temporalio.contrib.langsmith._interceptor._LangSmithWorkflowInboundInterceptor

class documentation

class _LangSmithWorkflowInboundInterceptor(temporalio.worker.WorkflowInboundInterceptor): (source)

Constructor: _LangSmithWorkflowInboundInterceptor(next)

View In Hierarchy

Instruments workflow execution with LangSmith runs.

Async Method execute_workflow Called to run the workflow.
Async Method handle_query Called to handle a query.
Async Method handle_signal Called to handle a signal.
Async Method handle_update_handler Called to handle an update's handler.
Method handle_update_validator Called to handle an update's validation stage.
Method init Initialize with an outbound interceptor.
Method _workflow_maybe_run Workflow-specific run creation with metadata.
Class Variable _config Undocumented

Inherited from WorkflowInboundInterceptor:

Method __init__ Create the inbound interceptor.
Instance Variable next Undocumented
async def execute_workflow(self, input: temporalio.worker.ExecuteWorkflowInput) -> Any: (source)
overrides temporalio.worker.WorkflowInboundInterceptor.execute_workflow

Called to run the workflow.

async def handle_query(self, input: temporalio.worker.HandleQueryInput) -> Any: (source)
overrides temporalio.worker.WorkflowInboundInterceptor.handle_query

Called to handle a query.

async def handle_signal(self, input: temporalio.worker.HandleSignalInput): (source)
overrides temporalio.worker.WorkflowInboundInterceptor.handle_signal

Called to handle a signal.

async def handle_update_handler(self, input: temporalio.worker.HandleUpdateInput) -> Any: (source)
overrides temporalio.worker.WorkflowInboundInterceptor.handle_update_handler

Called to handle an update's handler.

def handle_update_validator(self, input: temporalio.worker.HandleUpdateInput): (source)
overrides temporalio.worker.WorkflowInboundInterceptor.handle_update_validator

Called to handle an update's validation stage.

def init(self, outbound: temporalio.worker.WorkflowOutboundInterceptor): (source)
overrides temporalio.worker.WorkflowInboundInterceptor.init

Initialize with an outbound interceptor.

To add a custom outbound interceptor, wrap the given interceptor before sending to the next init call.

def _workflow_maybe_run(self, name: str, headers: Mapping[str, Payload] | None = None) -> Iterator[None]: (source)

Workflow-specific run creation with metadata.

Extracts parent from headers (if provided) and sets up tracing_context so @traceable functions called from workflow code can discover the parent and LangSmith client, independent of the add_temporal_runs toggle.

_config: ClassVar[LangSmithInterceptor] = (source)

Undocumented