temporalio.contrib.langsmith._interceptor._LangSmithWorkflowOutboundInterceptor

class documentation

class _LangSmithWorkflowOutboundInterceptor(temporalio.worker.WorkflowOutboundInterceptor): (source)

Constructor: _LangSmithWorkflowOutboundInterceptor(next, config)

View In Hierarchy

Instruments all outbound calls from workflow code.

Method __init__ Create the outbound interceptor.
Method continue_as_new Called for every temporalio.workflow.continue_as_new call.
Async Method signal_child_workflow Called for every temporalio.workflow.ChildWorkflowHandle.signal call.
Async Method signal_external_workflow Called for every temporalio.workflow.ExternalWorkflowHandle.signal call.
Method start_activity Called for every temporalio.workflow.start_activity and temporalio.workflow.execute_activity call.
Async Method start_child_workflow Called for every temporalio.workflow.start_child_workflow and temporalio.workflow.execute_child_workflow call.
Method start_local_activity Called for every temporalio.workflow.start_local_activity and temporalio.workflow.execute_local_activity call.
Async Method start_nexus_operation Called for every temporalio.workflow.NexusClient.start_operation call.
Method _traced_outbound Outbound workflow run creation with context injection into input.headers.
Instance Variable _config Undocumented

Inherited from WorkflowOutboundInterceptor:

Method info Called for every temporalio.workflow.info call.
Instance Variable next Undocumented
def __init__(self, next: temporalio.worker.WorkflowOutboundInterceptor, config: LangSmithInterceptor): (source)
overrides temporalio.worker.WorkflowOutboundInterceptor.__init__

Create the outbound interceptor.

Parameters
next:temporalio.worker.WorkflowOutboundInterceptorThe next interceptor in the chain. The default implementation of all calls is to delegate to the next interceptor.
config:LangSmithInterceptorUndocumented
def continue_as_new(self, input: temporalio.worker.ContinueAsNewInput) -> NoReturn: (source)
overrides temporalio.worker.WorkflowOutboundInterceptor.continue_as_new
async def signal_child_workflow(self, input: temporalio.worker.SignalChildWorkflowInput): (source)
overrides temporalio.worker.WorkflowOutboundInterceptor.signal_child_workflow
async def signal_external_workflow(self, input: temporalio.worker.SignalExternalWorkflowInput): (source)
overrides temporalio.worker.WorkflowOutboundInterceptor.signal_external_workflow
def start_activity(self, input: temporalio.worker.StartActivityInput) -> temporalio.workflow.ActivityHandle[Any]: (source)
overrides temporalio.worker.WorkflowOutboundInterceptor.start_activity
async def start_child_workflow(self, input: temporalio.worker.StartChildWorkflowInput) -> temporalio.workflow.ChildWorkflowHandle[Any, Any]: (source)
overrides temporalio.worker.WorkflowOutboundInterceptor.start_child_workflow
def start_local_activity(self, input: temporalio.worker.StartLocalActivityInput) -> temporalio.workflow.ActivityHandle[Any]: (source)
overrides temporalio.worker.WorkflowOutboundInterceptor.start_local_activity
async def start_nexus_operation(self, input: temporalio.worker.StartNexusOperationInput[Any, Any]) -> temporalio.workflow.NexusOperationHandle[Any]: (source)
overrides temporalio.worker.WorkflowOutboundInterceptor.start_nexus_operation
def _traced_outbound(self, name: str, input: _InputWithHeaders) -> Iterator[None]: (source)

Outbound workflow run creation with context injection into input.headers.

Uses ambient context so @traceable step functions that wrap outbound calls correctly parent the outbound run under themselves.

_config = (source)

Undocumented