temporalio.contrib.openai_agents._trace_interceptor.OpenAIAgentsContextPropagationInterceptor

class documentation

class OpenAIAgentsContextPropagationInterceptor(temporalio.client.Interceptor, temporalio.worker.Interceptor): (source)

Known subclasses: temporalio.contrib.openai_agents._otel_trace_interceptor.OTelOpenAIAgentsContextPropagationInterceptor

Constructor: OpenAIAgentsContextPropagationInterceptor(payload_converter, add_temporal_spans, start_traces)

View In Hierarchy

Interceptor that propagates OpenAI agent tracing context through Temporal workflows and activities.

Warning

This API is experimental and may change in future versions. Use with caution in production environments.

This interceptor enables tracing of OpenAI agent operations across Temporal workflows and activities. It propagates trace context through workflow and activity boundaries, allowing for end-to-end tracing of agent operations.

The interceptor handles: 1. Propagating trace context from client to workflow 2. Propagating trace context from workflow to activities 3. Maintaining trace context across workflow and activity boundaries

Example usage:
interceptor = OpenAIAgentsTracingInterceptor() client = await Client.connect("localhost:7233", interceptors=[interceptor]) worker = Worker(client, task_queue="my-task-queue", interceptors=[interceptor])
Method __init__ Initialize the interceptor with a payload converter.
Method context_from_header Extracts and initializes trace information the input header.
Method get_header_contents Extract trace context information from input headers.
Method header_contents Gets the OpenAI Agents trace/span data for the input header.
Method intercept_activity Intercepts activity calls to propagate trace context.
Method intercept_client Intercepts client calls to propagate trace context.
Method maybe_span Context manager that conditionally creates a span.
Method set_header_from_context Inserts the OpenAI Agents trace/span data in the input header.
Method span_context_from_header_contents Initialize span context from header contents.
Method trace_context_from_header_contents Initialize trace context from header contents.
Method workflow_interceptor_class Returns the workflow interceptor class to propagate trace context.
Instance Variable _add_temporal_spans Undocumented
Instance Variable _payload_converter Undocumented
Instance Variable _start_traces Undocumented

Inherited from Interceptor (via Interceptor):

Method intercept_nexus_operation Method called for intercepting a Nexus operation.
def __init__(self, payload_converter: temporalio.converter.PayloadConverter = temporalio.converter.default().payload_converter, add_temporal_spans: bool = True, start_traces: bool = False): (source)
overridden in temporalio.contrib.openai_agents._otel_trace_interceptor.OTelOpenAIAgentsContextPropagationInterceptor

Initialize the interceptor with a payload converter.

Parameters
payload_converter:temporalio.converter.PayloadConverterThe payload converter to use for serializing/deserializing trace context. Defaults to the default Temporal payload converter.
add_temporal_spans:boolWhether to add temporal-specific spans to traces.
start_traces:boolWhether to start new traces if none exist. This will cause duplication if the underlying trace provider actually process start events. Primarily designed for use with Open Telemetry integration.
def context_from_header(self, input: _InputWithHeaders): (source)
overridden in temporalio.contrib.openai_agents._otel_trace_interceptor.OTelOpenAIAgentsContextPropagationInterceptor

Extracts and initializes trace information the input header.

def get_header_contents(self, input: _InputWithHeaders) -> dict[str, Any] | None: (source)

Extract trace context information from input headers.

Parameters
input:_InputWithHeadersInput with headers containing trace information.
Returns
dict[str, Any] | NoneDictionary containing trace context or None if no headers present.
def header_contents(self) -> dict[str, Any]: (source)
overridden in temporalio.contrib.openai_agents._otel_trace_interceptor.OTelOpenAIAgentsContextPropagationInterceptor

Gets the OpenAI Agents trace/span data for the input header.

def intercept_activity(self, next: temporalio.worker.ActivityInboundInterceptor) -> temporalio.worker.ActivityInboundInterceptor: (source)
overrides temporalio.worker.Interceptor.intercept_activity

Intercepts activity calls to propagate trace context.

Parameters
next:temporalio.worker.ActivityInboundInterceptorThe next interceptor in the chain.
Returns
temporalio.worker.ActivityInboundInterceptorAn interceptor that propagates trace context for activity operations.
def intercept_client(self, next: temporalio.client.OutboundInterceptor) -> temporalio.client.OutboundInterceptor: (source)
overrides temporalio.client.Interceptor.intercept_client

Intercepts client calls to propagate trace context.

Parameters
next:temporalio.client.OutboundInterceptorThe next interceptor in the chain.
Returns
temporalio.client.OutboundInterceptorAn interceptor that propagates trace context for client operations.
def maybe_span(self, span_name: str, data: dict[str, Any] | None): (source)

Context manager that conditionally creates a span.

Parameters
span_name:strName for the span.
data:dict[str, Any] | NoneOptional data to attach to the span.
Yields
Context with optional span tracking.
def set_header_from_context(self, input: _InputWithHeaders): (source)

Inserts the OpenAI Agents trace/span data in the input header.

def span_context_from_header_contents(self, span_info: dict[str, Any]): (source)

Initialize span context from header contents.

Parameters
span_info:dict[str, Any]Dictionary containing span information from headers.
def trace_context_from_header_contents(self, span_info: dict[str, Any]): (source)

Initialize trace context from header contents.

Parameters
span_info:dict[str, Any]Dictionary containing trace information from headers.
def workflow_interceptor_class(self, input: temporalio.worker.WorkflowInterceptorClassInput) -> type[_ContextPropagationWorkflowInboundInterceptor]: (source)
overrides temporalio.worker.Interceptor.workflow_interceptor_class

Returns the workflow interceptor class to propagate trace context.

Parameters
input:temporalio.worker.WorkflowInterceptorClassInputThe input for creating the workflow interceptor.
Returns
type[_ContextPropagationWorkflowInboundInterceptor]The class of the workflow interceptor that propagates trace context.
_add_temporal_spans = (source)

Undocumented

_payload_converter = (source)

Undocumented

_start_traces = (source)

Undocumented