OpenAI Agents SDK Integration for Temporal

We welcome questions and feedback in the #python-sdk Slack channel at temporalio.slack.com.

Install

uv add temporalio-openai-agents

With Temporal 1.33, both distributions must be installed into the same physical site-packages/temporalio directory, as they are in a standard non-editable virtual environment. Split-directory installations—including editable installs, separate user and system sites, layered deployments, and --target installs—require Temporal 1.34 or later so that temporalio extends its package search path.

Introduction

This integration combines OpenAI Agents SDK with Temporal's durable execution. It allows you to build durable agents that never lose their progress and handle long-running, asynchronous, and human-in-the-loop workflows with production-grade reliability.

Temporal and OpenAI Agents SDK are complementary technologies, both of which contribute to simplifying what it takes to build highly capable, high-quality AI systems. Temporal provides a crash-proof system foundation, taking care of the distributed systems challenges inherent to production agentic systems. OpenAI Agents SDK offers a lightweight yet powerful framework for defining those agents.

This document is organized as follows:

• Hello World Durable Agent. Your first durable agent example.
• Background Concepts. Background on durable execution and AI agents.
• Full Example Running the Hello World Durable Agent example.
• Tool Calling. Calling agent Tools in Temporal.
• Sandbox Support. Running sandbox agents in Temporal.
• Feature Support. Compatibility matrix.

The samples repository contains examples including basic usage, common agent patterns, and more complete samples.

Hello World Durable Agent

The code below shows how to wrap an agent for durable execution.

File 1: Durable Agent (hello_world.py)

from temporalio import workflow
from agents import Agent, Runner

@workflow.defn
class HelloWorldAgent:
    @workflow.run
    async def run(self, prompt: str) -> str:
        agent = Agent(
            name="Assistant",
            instructions="You only respond in haikus.",
        )

        result = await Runner.run(agent, input=prompt)
        return result.final_output

In this example, Temporal provides the durable execution wrapper: the HelloWorldAgent.run method. The content of that method, is regular OpenAI Agents SDK code.

If you are familiar with Temporal and with Open AI Agents SDK, this code will look very familiar. The @workflow.defn annotation on the HelloWorldAgent indicates that this class will contain durable execution logic. The @workflow.run annotation defines the entry point. We use the Agent class from OpenAI Agents SDK to define a simple agent, instructing it to always respond with haikus. We then run that agent, using the Runner class from OpenAI Agents SDK, passing through prompt as an argument.

We will complete this example below. Before digging further into the code, we will review some background that will make it easier to understand.

Background Concepts

We encourage you to review this section thoroughly to gain a solid understanding of AI agents and durable execution with Temporal. This knowledge will make it easier to design and build durable agents. If you are already well versed in these topics, feel free to skim this section or skip ahead.

AI Agents

In the OpenAI Agents SDK, an agent is an AI model configured with instructions, tools, MCP servers, guardrails, handoffs, context, and more.

We describe each of these briefly:

• AI model. An LLM such as OpenAI's GPT, Google's Gemini, or one of many others.
• Instructions. Also known as a system prompt, the instructions contain the initial input to the model, which configures it for the job it will do.
• Tools. Typically, Python functions that the model may choose to invoke. Tools are functions with text-descriptions that explain their functionality to the model.
• MCP servers. Best known for providing tools, MCP offers a pluggable standard for interoperability, including file-like resources, prompt templates, and human approvals. MCP servers may be accessed over the network or run in a local process.
• Guardrails. Checks on the input or the output of an agent to ensure compliance or safety. Guardrails may be implemented as regular code or as AI agents.
• Handoffs. A handoff occurs when an agent delegates a task to another agent. During a handoff the conversation history remains the same, and passes to a new agent with its own model, instructions, tools.
• Context. This is an overloaded term. Here, context refers to a framework object that is shared across tools and other code, but is not passed to the model.

Now, let's see how these components work together. In a common pattern, the model first receives user input and then reasons about which tool to invoke. The tool's response is passed back to the model, which may call additional tools, repeating this loop until the task is complete.

The diagram below illustrates this flow.

           +-------------------+
           |     User Input    |
           +-------------------+
                     |
                     v
          +---------------------+
          |  Reasoning (Model)  |  <--+
          +---------------------+     |
                     |                |
           (decides which action)     |
                     v                |
          +---------------------+     |
          |       Action        |     |
          | (e.g., use a Tool)  |     |
          +---------------------+     |
                     |                |
                     v                |
          +---------------------+     |
          |     Observation     |     |
          |    (Tool Output)    |     |
          +---------------------+     |
                     |                |
                     +----------------+
          (loop: uses new info to reason
           again, until task is complete)

Even in a simple example like this, there are many places where things can go wrong. Tools call APIs that sometimes fail, while models can encounter rate limits, requiring retries. The longer the agent runs, the more costly it is to start the job over. We next describe durable execution, which handles such failures seamlessly.

Durable Execution

In Temporal's durable execution implementation, a program that crashes or encounters an exception while interacting with a model or API will retry until it can successfully complete.

Temporal relies primarily on a replay mechanism to recover from failures. As the program makes progress, Temporal saves key inputs and decisions, allowing a re-started program to pick up right where it left off.

The key to making this work is to separate the applications repeatable (deterministic) and non-repeatable (non-deterministic) parts:

1. Deterministic pieces, termed workflows, execute the same way when re-run with the same inputs.
2. Non-deterministic pieces, termed activities, can run arbitrary code, performing I/O and any other operations.

Workflow code can run for extended periods and, if interrupted, resume exactly where it left off. Activity code faces no restrictions on I/O or external interactions, but if it fails part-way through it restarts from the beginning.

In this integration, model invocations are automatically routed through Temporal activities, while the logic that coordinates them lives in the workflow. Tools that perform I/O or other non-deterministic work should run as Temporal activities, while deterministic, workflow-safe tools can run directly in the workflow. This pattern generalizes to more sophisticated agents. We refer to the coordinating logic as agent orchestration.

The diagram below shows the overall architecture of an agentic application in Temporal. The Temporal Server is responsible to tracking program execution and making sure associated state is preserved reliably (i.e., stored to a database, possibly replicated across cloud regions). Temporal Server manages data in encrypted form, so all data processing occurs on the Worker, which runs the workflow and activities.

            +---------------------+
            |   Temporal Server   |      (Stores workflow state,
            +---------------------+       schedules activities,
                     ^                    persists progress)
                     |
        Save state,  |   Schedule Tasks,
        progress,    |   load state on resume
        timeouts     |
                     |
+------------------------------------------------------+
|                      Worker                          |
|   +----------------------------------------------+   |
|   |              Workflow Code                   |   |
|   |  (Agent orchestration + deterministic tools) |   |
|   +----------------------------------------------+   |
|          |          |                |               |
|          v          v                v               |
|   +-----------+ +-----------+ +-------------+        |
|   | Activity  | | Activity  | |  Activity   |        |
|   | (I/O Tool | | (I/O Tool | | (Model API) |        |
|   |     1)    | |     2)    | |             |        |
|   +-----------+ +-----------+ +-------------+        |
|         |           |                |               |
+------------------------------------------------------+
          |           |                |
          v           v                v
      [External APIs, services, databases, etc.]

See the Temporal documentation for more information.

Complete Example

To make the Hello World durable agent shown earlier available in Temporal, we need to create a worker program. To see it run, we also need a client to launch it. We show these files below.

File 2: Launch Worker (run_worker.py)

# File: run_worker.py

import asyncio
from datetime import timedelta

from temporalio.client import Client
from temporalio.openai_agents import OpenAIAgentsPlugin, ModelActivityParameters
from temporalio.worker import Worker

from hello_world_workflow import HelloWorldAgent


async def worker_main():
    # Use the plugin to configure Temporal for use with OpenAI Agents SDK
    client = await Client.connect(
        "localhost:7233",
        plugins=[
            OpenAIAgentsPlugin(
                model_params=ModelActivityParameters(
                    start_to_close_timeout=timedelta(seconds=30)
                )
            ),
        ],
    )

    worker = Worker(
        client,
        task_queue="my-task-queue",
        workflows=[HelloWorldAgent],
    )
    await worker.run()


if __name__ == "__main__":
    asyncio.run(worker_main())

We use the OpenAIAgentsPlugin to configure Temporal for use with OpenAI Agents SDK. The plugin automatically handles several important setup tasks:

• Ensures proper serialization of Pydantic types
• Propagates context for OpenAI Agents tracing.
• Registers an activity for invoking model calls with the Temporal worker.
• Configures OpenAI Agents SDK to run model calls as Temporal activities.

File 3: Client Execution (run_hello_world_workflow.py)

# File: run_hello_world_workflow.py

import asyncio

from temporalio.client import Client
from temporalio.common import WorkflowIDReusePolicy
from temporalio.openai_agents import OpenAIAgentsPlugin

from hello_world_workflow import HelloWorldAgent

async def main():
    # Create client connected to server at the given address
    client = await Client.connect(
        "localhost:7233",
        plugins=[OpenAIAgentsPlugin()],
    )

    # Execute a workflow
    result = await client.execute_workflow(
        HelloWorldAgent.run,
        "Tell me about recursion in programming.",
        id="my-workflow-id",
        task_queue="my-task-queue",
        id_reuse_policy=WorkflowIDReusePolicy.ALLOW_DUPLICATE,
        id_conflict_policy=WorkflowIDConflictPolicy.TERMINATE_EXISTING,
    )
    print(f"Result: {result}")

if __name__ == "__main__":
    asyncio.run(main())

This file is a standard Temporal launch script. We also configure the client with the OpenAIAgentsPlugin to ensure serialization is compatible with the worker.

To run this example, see the detailed instructions in the Temporal Python Samples Repository.

Tool Calling

Model invocations are automatically routed through Temporal activities. OpenAI-hosted tools are passed through the model invocation and executed by the model provider. User-defined FunctionTools, including tools created with @function_tool, are not automatically converted into Temporal activities; they execute in the workflow unless explicitly backed by a Temporal activity. Where a tool executes depends on how it is defined:

Temporal Activities as OpenAI Agents Tools

One of the powerful features of this integration is the ability to convert Temporal activities into agent tools using activity_as_tool. This allows your agent to leverage Temporal's durable execution for tool calls. activity_as_tool() creates an OpenAI Agents FunctionTool whose invocation schedules the underlying Temporal activity.

In the example below, we apply the @activity.defn decorator to the get_weather function to create a Temporal activity. We then pass this through the activity_as_tool helper function to create an OpenAI Agents tool that is passed to the Agent.

from dataclasses import dataclass
from datetime import timedelta
from temporalio import activity, workflow
from temporalio.contrib import openai_agents
from agents import Agent, Runner

@dataclass
class Weather:
    city: str
    temperature_range: str
    conditions: str

@activity.defn
async def get_weather(city: str) -> Weather:
    """Get the weather for a given city."""
    return Weather(city=city, temperature_range="14-20C", conditions="Sunny with wind.")

@workflow.defn
class WeatherAgent:
    @workflow.run
    async def run(self, question: str) -> str:
        agent = Agent(
            name="Weather Assistant",
            instructions="You are a helpful weather agent.",
            tools=[
                openai_agents.workflow.activity_as_tool(
                    get_weather,
                    start_to_close_timeout=timedelta(seconds=10)
                )
            ],
        )
        result = await Runner.run(starting_agent=agent, input=question)
        return result.final_output

The activity must also be registered with a Worker. activity_as_tool() controls how the Agent invokes the activity; it does not register the activity with the Worker.

from temporalio.worker import Worker

worker = Worker(
    client,
    task_queue="my-task-queue",
    workflows=[WeatherAgent],
    activities=[get_weather],
)

Calling OpenAI Agents Tools inside Temporal Workflows

For simple computations that don't involve external calls, you can call the tool directly from the workflow by using the standard OpenAI Agents SDK @function_tool decorator.

from temporalio import workflow
from agents import Agent, Runner
from agents import function_tool

@function_tool
def calculate_circle_area(radius: float) -> float:
    """Calculate the area of a circle given its radius."""
    import math
    return math.pi * radius ** 2

@workflow.defn
class MathAssistantAgent:
    @workflow.run
    async def run(self, message: str) -> str:
        agent = Agent(
            name="Math Assistant",
            instructions="You are a helpful math assistant. Use the available tools to help with calculations.",
            tools=[calculate_circle_area],
        )
        result = await Runner.run(agent, input=message)
        return result.final_output

Use regular @function_tool tools only for deterministic, workflow-safe logic. Do not perform network, database, filesystem, or other external I/O directly from these tools. Use a Temporal activity with activity_as_tool() instead. Code running in the workflow can also invoke a Temporal activity directly when needed.

Tools that run in the workflow can also update OpenAI Agents context, which is read-only for tools run as Temporal activities.

MCP Support

The durable MCP integration uses MCP Python SDK v2 and the optional mcp dependencies:

uv add "temporalio-openai-agents[mcp]"

Register named OpenAI MCP server factories on the worker, then reference the same name from workflow code:

from agents.mcp import MCPServerStreamableHttp

from temporalio.openai_agents import OpenAIAgentsPlugin

plugin = OpenAIAgentsPlugin(
    mcp_servers={
        "weather": lambda: MCPServerStreamableHttp(
            name="weather",
            params={"url": "https://example.com/mcp"},
        ),
    },
)

from agents import Agent
from temporalio.openai_agents.workflow import temporal_mcp_server

server = temporal_mcp_server("weather")
agent = Agent(name="weather", mcp_servers=[server])

Configure transport, connection, retry, and message-handling behavior on the worker-side OpenAI MCPServer. Configure workflow-facing behavior such as tool_filter, require_approval, failure_error_function, and metadata resolvers on temporal_mcp_server(...), where the OpenAI agent can use it. Custom MCPServer method implementations still execute worker-side. A callable tool_filter on the worker-side server is rejected, because the run context and agent it receives exist only in the workflow. A callable tool_filter passed to temporal_mcp_server(...) therefore executes during workflow replay and must be deterministic: it must not perform I/O or depend on the system clock, randomness, mutable global state, or other external state.

Every MCP operation is a Temporal Activity. The workflow-side tool list is cached by default; pass cache_tools_list=False to refresh it on every Agents SDK listing. A parameterless factory's connection is also reused for up to five idle minutes when it negotiates the modern, sessionless protocol. Set mcp_connection_idle_timeout=None to keep cached modern connections until plugin shutdown, or timedelta(0) to close them as soon as they become idle. Connections that negotiate a legacy handshake are closed after the current Activity and are never shared across workflows.

OpenAIAgentsPlugin(mcp_servers={"x": ...}) and MCPPlugin(clients={"x": ...}) both register Activities under temporalio.mcp.x.*. Do not register the same MCP server name through both plugins on one worker; Temporal rejects the duplicate Activity types.

An optional factory_argument can select worker-side configuration such as a tenant endpoint:

server = temporal_mcp_server(
    "weather",
    factory_argument={"tenant": "acme"},
)

A non-None argument creates a fresh client for each Activity. Omitting the argument or passing None uses the zero-argument factory and permits connection caching. Caching by only the registered name could otherwise reuse one tenant's endpoint or authorization for another tenant; caching by the argument itself is unsafe because arguments may be unhashable, high-cardinality, or refer to configuration that changes worker-side. The argument is recorded in workflow history, so it must be a non-secret stable identifier. Resolve secrets inside the factory.

MCP v2 uses httpx2, not legacy httpx. OpenAI Agents owns the custom HTTP client lifecycle supplied through its MCP server parameters:

import httpx2
from agents.mcp import MCPServerStreamableHttp


def http_client_factory(headers=None, timeout=None, auth=None):
    return httpx2.AsyncClient(
        headers={**(headers or {}), "X-Client": "temporal"},
        timeout=timeout,
        auth=auth,
    )


def weather_server() -> MCPServerStreamableHttp:
    return MCPServerStreamableHttp(
        name="weather",
        params={
            "url": "https://example.com/mcp",
            "httpx_client_factory": http_client_factory,
        },
    )

StatelessMCPServerProvider, StatefulMCPServerProvider, the plugin's mcp_server_providers option, stateless_mcp_server(), and stateful_mcp_server() are deprecated. They remain supported for source and workflow-history compatibility and can still run with MCP Python SDK v1 when the mcp extra is not installed. New integrations should use mcp_servers and temporal_mcp_server(), which require the mcp extra and MCP Python SDK v2. The legacy stateful path retains its dedicated per-workflow worker and persistent-session behavior.

Hosted MCP Tool

For network-accessible MCP servers, you can also use HostedMCPTool from the OpenAI Agents SDK, which uses an MCP client hosted by OpenAI.

Secrets from the Worker's Environment

⚠️ Experimental - This functionality is subject to change prior to General Availability.

A credential an agent needs can stay in the worker process's environment instead of being written into your workflow. Where the value would otherwise go, you name the environment variable that holds it, and the worker reads that variable when the value is actually needed.

There are two forms, and which one you use follows from where the value goes:

• For a hosted tool credential, use temporal_worker_env_ref(). It is substituted only in the fields listed under Hosted Tool Credentials.
• For a sandbox environment variable, use TemporalWorkerEnvValue.

Both are gated by resolvable_worker_env_vars, an allowlist of the variable names a worker is willing to read. On every worker that runs model or sandbox activities, set the variable and add its name to that list:

plugin = OpenAIAgentsPlugin(resolvable_worker_env_vars=["MY_MCP_TOKEN"])

Names are matched exactly, with no globbing. Passing AllowAllWorkerEnvVars() in place of the list makes every environment variable on the worker resolvable, so a workflow-authored sandbox manifest can name any variable on the worker and have its value land inside the container.

from temporalio.openai_agents import AllowAllWorkerEnvVars

plugin = OpenAIAgentsPlugin(resolvable_worker_env_vars=AllowAllWorkerEnvVars())

The reference form never raises. A name the worker does not allow is sent on as the reference string, and a name it allows resolves to whatever the variable holds — an empty string when that variable is unset or empty.

Hosted Tool Credentials

Pass temporal_worker_env_ref() the name of an environment variable, in place of the credential itself:

from agents import HostedMCPTool
from temporalio.openai_agents import temporal_worker_env_ref

tool = HostedMCPTool(
    tool_config={
        "type": "mcp",
        "server_label": "my_server",
        "server_url": "https://example.com/mcp",
        "authorization": temporal_worker_env_ref("MY_MCP_TOKEN"),
    }
)

A reference can sit inside a larger value: in "Bearer " + temporal_worker_env_ref("MY_MCP_TOKEN"), the reference is replaced in place and the rest of the string is sent unchanged.

The environment variable's value is substituted in these fields and no others:

• authorization, and the value of each entry in headers, in a HostedMCPTool's tool_config
• value in each entry of network_policy.domain_secrets under a hosted ShellTool's environment
• value in each entry of network_policy.domain_secrets under the container in a CodeInterpreterTool's tool_config

Sandbox Environment Variables

Put a TemporalWorkerEnvValue in the environment of a sandbox manifest, in place of the value itself:

from agents.sandbox import Manifest
from agents.sandbox.manifest import Environment

from temporalio.openai_agents import TemporalWorkerEnvValue

manifest = Manifest(
    environment=Environment(
        value={
            "OPENAI_API_KEY": TemporalWorkerEnvValue(name="PROD_OPENAI_KEY"),
            "REGION": "us-west-2",
        }
    )
)

Pass that manifest to SandboxRunConfig(manifest=...). This reads PROD_OPENAI_KEY on the worker and sets OPENAI_API_KEY inside the sandbox, so the two names need not match.

Sandbox Support

⚠️ Pre-release - This functionality is subject to change prior to General Availability.

The sandbox integration lets SandboxAgent from the OpenAI Agents SDK execute inside a remote or local sandbox (Daytona, Docker, E2B, local Unix, etc.) while keeping all coordination durable in Temporal.

Every sandbox operation — creating a session, running commands, reading/writing files, PTY interactions — is dispatched as a Temporal activity. This means sandbox work is fully observable, retryable, and recoverable like any other activity, and sandbox session state is serialized with the workflow so it survives worker restarts.

Architecture

Workflow Code
  ↓
temporal_sandbox_client("daytona")   [returns TemporalSandboxClient]
  ↓
SandboxAgent.run(run_config=RunConfig(sandbox=SandboxRunConfig(client=...)))
  ↓
sandbox agent calls session.exec / session.read / session.write / …
  ↓
TemporalSandboxSession routes each call as a Temporal activity
("daytona-sandbox_session_exec", "daytona-sandbox_session_read", …)
  ↓
SandboxClientProvider activities on the worker call the real sandbox client
  ↓
Actual sandbox backend (Daytona, Docker, local, …)

Worker Configuration

Register one or more SandboxClientProvider instances with the plugin. Each provider pairs a unique name with a real BaseSandboxClient implementation. The plugin automatically registers all required activities on the worker.

import asyncio
from datetime import timedelta
from temporalio.client import Client
from temporalio.worker import Worker
from temporalio.openai_agents import OpenAIAgentsPlugin, SandboxClientProvider, ModelActivityParameters
from agents.extensions.sandbox.daytona import DaytonaSandboxClient
from agents.sandbox.sandboxes.unix_local import UnixLocalSandboxClient

async def main():
    client = await Client.connect(
        "localhost:7233",
        plugins=[
            OpenAIAgentsPlugin(
                model_params=ModelActivityParameters(
                    start_to_close_timeout=timedelta(seconds=30)
                ),
                sandbox_clients=[
                    SandboxClientProvider("daytona", DaytonaSandboxClient()),
                    SandboxClientProvider("local", UnixLocalSandboxClient()),
                ],
            ),
        ],
    )

    worker = Worker(
        client,
        task_queue="my-task-queue",
        workflows=[MyWorkflow],
    )
    await worker.run()

Provider names must be unique. Each name becomes the prefix for that backend's activities, allowing multiple backends to coexist on a single worker.

Workflow Usage

In the workflow, use temporal_sandbox_client() to create a reference to a registered backend by name. Pass it to SandboxRunConfig inside RunConfig:

from temporalio import workflow
from temporalio.openai_agents.workflow import temporal_sandbox_client
from agents import Runner
from agents.sandbox import SandboxAgent, SandboxRunConfig
from agents.run import RunConfig
from agents.extensions.sandbox.daytona import DaytonaSandboxClientOptions

@workflow.defn
class MyWorkflow:
    @workflow.run
    async def run(self, prompt: str) -> str:
        agent = SandboxAgent(
            name="Coding Assistant",
            instructions="You are a helpful coding assistant with access to a sandbox.",
        )

        result = await Runner.run(
            agent,
            prompt,
            run_config=RunConfig(
                sandbox=SandboxRunConfig(
                    client=temporal_sandbox_client("daytona"),
                    options=DaytonaSandboxClientOptions(pause_on_exit=False),
                ),
            ),
        )
        return result.final_output

The name passed to temporal_sandbox_client() must exactly match the name used in SandboxClientProvider on the worker.

Multiple Backends

A single workflow can target different backends by name. Register all backends on the worker and reference each by name in the workflow:

# Run a task on the "daytona" backend
result = await Runner.run(
    agent, prompt,
    run_config=RunConfig(sandbox=SandboxRunConfig(
        client=temporal_sandbox_client("daytona"),
        options=DaytonaSandboxClientOptions(pause_on_exit=False),
    )),
)

# Run a different task on the "local" backend
result = await Runner.run(
    agent, prompt,
    run_config=RunConfig(sandbox=SandboxRunConfig(
        client=temporal_sandbox_client("local"),
        options=UnixLocalSandboxClientOptions(),
    )),
)

Streaming

⚠️ Experimental - This functionality is subject to change prior to General Availability.

The integration supports streaming model responses via the SDK-native Runner.run_streamed API. Inside a workflow, model calls execute as a streaming activity (invoke_model_activity_streaming) that consumes Model.stream_response and returns the collected list of native OpenAI response events. The workflow surfaces those events to the caller through RunResultStreaming.stream_events(), which wraps them in the agents-SDK StreamEvent union (so raw model events arrive as RawResponsesStreamEvent.data).

External consumers (UIs, tracing pipelines, etc.) observe events as they arrive by hosting a WorkflowStream in the workflow and subscribing with WorkflowStreamClient. The streaming activity publishes each event to the topic configured on ModelActivityParameters.streaming_topic. The topic is required when using Runner.run_streamed; calling it without a configured topic raises before any activity is scheduled.

Example workflow consuming events via stream_events() while the streaming activity publishes to the "events" topic:

from agents import Agent, Runner
from agents.stream_events import RawResponsesStreamEvent

from temporalio import workflow

@workflow.defn
class MyAgent:
    @workflow.run
    async def run(self, prompt: str) -> str:
        agent = Agent(name="Assistant", instructions="...")
        result = Runner.run_streamed(agent, prompt)
        async for event in result.stream_events():
            if isinstance(event, RawResponsesStreamEvent):
                raw_event = event.data  # native OpenAI ResponseStreamEvent
                ...
        return result.final_output

To publish raw model events to external subscribers, host a WorkflowStream in the workflow and configure OpenAIAgentsPlugin(model_params=ModelActivityParameters(streaming_topic="events")). See temporalio.contrib.workflow_streams for the publisher and subscriber API.

RunResultStreaming.stream_events() yields the agents-SDK StreamEvent union (RawResponsesStreamEvent, RunItemStreamEvent, AgentUpdatedStreamEvent); native OpenAI response events arrive wrapped as RawResponsesStreamEvent.data. Workflow-stream subscribers, by contrast, receive the unwrapped native events directly because the streaming activity publishes them straight from Model.stream_response.

Streaming is incompatible with use_local_activity because local activities support neither activity heartbeats nor the workflow stream signal channel.

Activity retries surface to workflow-stream subscribers but not to RunResultStreaming.stream_events(). Events are published to the stream as Model.stream_response produces them, so a partial attempt that fails mid-response leaves its emitted events on the stream and the retry attempt publishes a second sequence. stream_events() only sees the final successful attempt's collected events because it consumes the activity's return value. Workflow-stream subscribers should treat retries the same way as any other workflow_streams publisher — see Delivery semantics for the trade and the conventional RETRY event pattern for surfacing the transition to consumers.

Feature Support

This integration is presently subject to certain limitations. Realtime agents are not supported. Streaming is supported via Runner.run_streamed — see Streaming above. Certain tools are not suitable for a distributed computing environment, so these have been disabled as well.

Model Providers

Model Response format

Tools

Tool Type

LocalShellTool and ComputerTool are not suited to a distributed computing setting.

Tool Context

As described in Tool Calling, context propagation is read-only when Temporal activities are used as tools.

MCP

The integration supports MCP Python SDK v2 clients. Modern MCP protocol connections are sessionless; transport connections may still be reused as an optimization. The OpenAI MCP server factory can use stdio, streamable HTTP, an in-process server, or a custom v2 transport.

Note that when using network-accessible MCP servers, you also can also use the tool HostedMCPTool, which is part of the OpenAI Responses API and uses an MCP client hosted by OpenAI.

Guardrails

Sessions

SQLite storage is not suited to a distributed environment.

Tracing

OpenTelemetry Integration

⚠️ Public Preview - This functionality is subject to change prior to General Availability.

This integration provides seamless export of OpenAI agent telemetry to OpenTelemetry (OTEL) endpoints for observability and monitoring. The integration automatically handles workflow replay semantics, ensuring spans are only exported when workflows actually complete.

Quick Start

To enable OTEL telemetry export, you need to set up a global ReplaySafeTracerProvider and enable the integration in the OpenAIAgentsPlugin:

from datetime import timedelta
from temporalio.client import Client
from temporalio.openai_agents import OpenAIAgentsPlugin, ModelActivityParameters
from temporalio.contrib.opentelemetry import create_tracer_provider
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry import trace
from opentelemetry.sdk.trace.export import SimpleSpanProcessor

# Configure your OTEL exporters

# Set up the global tracer provider
tracer_provider = create_tracer_provider()
tracer_provider.add_span_processor(SimpleSpanProcessor(OTLPSpanExporter(endpoint="http://localhost:4317")))
trace.set_tracer_provider(tracer_provider)

client = await Client.connect(
    "localhost:7233",
    plugins=[
        OpenAIAgentsPlugin(
            use_otel_instrumentation=True,  # Enable OTEL integration
            model_params=ModelActivityParameters(
                start_to_close_timeout=timedelta(seconds=30)
            )
        ),
    ],
)

Features

• Multiple Exporters: Send telemetry to multiple OTEL endpoints simultaneously via the global tracer provider
• Replay-Safe: Spans are only exported when workflows actually complete, not during replays
• Deterministic IDs: Consistent span IDs across workflow replays for reliable correlation
• Automatic Setup: No manual instrumentation required - just enable the flag and set up the global tracer provider
• Graceful Degradation: Works seamlessly whether OTEL dependencies are installed or not

Dependencies

OTEL integration requires additional dependencies:

pip install openinference-instrumentation-openai-agents opentelemetry-sdk

Choose the appropriate OTEL exporter for your monitoring system:

# For OTLP (works with most OTEL collectors and monitoring systems)
pip install opentelemetry-exporter-otlp

# Other exporters available for specific systems
pip install opentelemetry-exporter-<your-system>

ConsoleSpanExporter (development/debugging) ships with opentelemetry-sdk, so it needs no extra package.

Example: Multiple Exporters

from temporalio.contrib.opentelemetry import create_tracer_provider
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter
from opentelemetry import trace

exporters = [
    # Production monitoring system
    OTLPSpanExporter(
        endpoint="https://your-monitoring-system:4317",
        headers={"api-key": "your-api-key"}
    ),
    
    # Secondary monitoring endpoint
    OTLPSpanExporter(endpoint="https://backup-collector:4317"),
    
    # Development debugging
    ConsoleSpanExporter(),
]

# Set up the global tracer provider with one span processor per exporter
tracer_provider = create_tracer_provider()
for exporter in exporters:
    tracer_provider.add_span_processor(BatchSpanProcessor(exporter))
trace.set_tracer_provider(tracer_provider)

plugin = OpenAIAgentsPlugin(use_otel_instrumentation=True)

Error Handling

If you enable OTEL instrumentation but the required dependencies are not installed, you'll receive a clear error message:

ImportError: OTEL dependencies not available. Install with: pip install openinference-instrumentation-openai-agents opentelemetry-sdk

If you enable OTEL instrumentation but don't have a proper global tracer provider set up, you'll get:

ValueError: Global tracer provider must a ReplaySafeTracerProvider. Use temporalio.contrib.opentelemtry.create_trace_provider to create one.

Direct OpenTelemetry API Calls in Workflows

When using direct OpenTelemetry API calls within workflows (e.g., opentelemetry.trace.get_tracer(__name__).start_as_current_span()), you need to ensure proper context bridging and sandbox configuration.

Sandbox Configuration

Workflows run in a sandbox that restricts module access. To use direct OTEL API calls, you must explicitly allow OpenTelemetry passthrough:

from temporalio.worker import Worker
from temporalio.worker.workflow_sandbox import SandboxedWorkflowRunner, SandboxRestrictions

# Configure worker with OpenTelemetry passthrough
worker = Worker(
    client,
    task_queue="my-task-queue",
    workflows=[MyWorkflow],
    workflow_runner=SandboxedWorkflowRunner(
        SandboxRestrictions.default.with_passthrough_modules("opentelemetry")
    )
)

Context Bridging Pattern

Direct OTEL spans must be created within an active OpenAI Agents SDK span to ensure proper parenting:

import opentelemetry.trace
from agents import custom_span
from temporalio import workflow

@workflow.defn
class MyWorkflow:
    @workflow.run
    async def run(self) -> str:
        # Start an SDK span first to establish OTEL context bridge
        with custom_span("Workflow coordination"):
            # Now direct OTEL spans will be properly parented
            tracer = opentelemetry.trace.get_tracer(__name__)
            with tracer.start_as_current_span("Custom workflow span"):
                # Your workflow logic here
                result = await self.do_work()
                return result

Why This Pattern is Required

• OpenInference instrumentation bridges OpenAI Agents SDK spans to OpenTelemetry context
• Direct OTEL API calls without an active SDK span become root spans with no parent
• SDK spans (custom_span()) establish the context bridge that allows subsequent direct OTEL spans to inherit proper trace parenting

Complete Example

import opentelemetry.trace
from agents import custom_span
from temporalio import workflow
from temporalio.worker import Worker
from temporalio.worker.workflow_sandbox import SandboxedWorkflowRunner, SandboxRestrictions

@workflow.defn
class TracedWorkflow:
    @workflow.run
    async def run(self) -> str:
        # Establish OTEL context with SDK span
        with custom_span("Main workflow"):
            # Create direct OTEL spans for fine-grained tracing
            tracer = opentelemetry.trace.get_tracer(__name__)
            
            with tracer.start_as_current_span("Data processing"):
                data = await self.process_data()
                
            with tracer.start_as_current_span("Business logic"):
                result = await self.execute_business_logic(data)
                
            return result

# Worker configuration
worker = Worker(
    client,
    task_queue="traced-workflows",
    workflows=[TracedWorkflow],
    workflow_runner=SandboxedWorkflowRunner(
        SandboxRestrictions.default.with_passthrough_modules("opentelemetry")
    )
)

This ensures your direct OTEL spans are properly parented within the trace hierarchy initiated by your client SDK traces.

Client-Side Trace Initialization

You can also start an Agents SDK trace on the client side before executing a workflow. This is useful when you want the entire workflow execution to be part of a larger trace context:

from agents import trace, custom_span
from temporalio.openai_agents import OpenAIAgentsPlugin

# Set up the plugin with OTEL integration
plugin = OpenAIAgentsPlugin(use_otel_instrumentation=True)

# Client setup
client = await Client.connect(
    "localhost:7233",
    plugins=[plugin]
)

# Start a trace on the client side
with plugin.tracing_context():
    with trace("Customer support workflow"):
        with custom_span("Workflow execution"):
            # Execute workflow within the trace context
            result = await client.execute_workflow(
                CustomerSupportAgent.run,
                "Help me with my order",
                id="customer-support-123",
                task_queue="my-task-queue",
            )
            print(f"Result: {result}")

The plugin.tracing_context() is required when starting traces outside of a worker context. This ensures proper instrumentation setup and trace propagation into the workflow execution.

If OTEL instrumentation is not enabled, the integration works normally without any OTEL setup.

Voice

Utilities

The REPL utility is not suitable for a distributed setting.

Additional Examples

You can find additional examples in the Temporal Python Samples Repository.