temporalio.contrib.openai_agents.trace_interceptor.OpenAIAgentsTracingInterceptor

class documentation

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

Constructor: OpenAIAgentsTracingInterceptor(payload_converter)

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 intercept_activity Intercepts activity calls to propagate trace context.
Method intercept_client Intercepts client calls to propagate trace context.
Method workflow_interceptor_class Returns the workflow interceptor class to propagate trace context.
Instance Variable _payload_converter Undocumented
def __init__(self, payload_converter: temporalio.converter.PayloadConverter = temporalio.converter.default().payload_converter): (source)

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.
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 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.
_payload_converter = (source)

Undocumented