Metadata-Version: 2.4
Name: temporalio-openai-agents
Version: 1.0.0
Summary: Temporal integration for the OpenAI Agents SDK (temporalio.openai_agents)
Author: Temporal Technologies Inc
Author-email: Temporal Technologies Inc <sdk@temporal.io>
License-Expression: MIT
License-File: LICENSE
Classifier: Development Status :: 5 - Production/Stable
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Typing :: Typed
Requires-Dist: temporalio[opentelemetry,pydantic]>=1.33.0,<2
Requires-Dist: openai-agents>=0.20,<0.21
Requires-Dist: openai>=2.45,<3
Requires-Dist: mcp>=1.24,<3
Requires-Dist: pydantic>=2,<3
Requires-Dist: typing-extensions>=4.2,<5
Requires-Dist: mcp>=2,<3 ; extra == 'mcp'
Requires-Dist: mcp-types>=2,<3 ; extra == 'mcp'
Requires-Dist: temporalio-mcp>=0.2.0,<0.3 ; extra == 'mcp'
Requires-Dist: openinference-instrumentation-openai-agents>=0.1.0 ; extra == 'openinference'
Requires-Dist: opentelemetry-sdk>=1.26,<2 ; extra == 'openinference'
Requires-Python: >=3.10
Project-URL: Homepage, https://github.com/temporalio/ai-integrations/tree/main/python/openai_agents
Project-URL: Repository, https://github.com/temporalio/ai-integrations
Project-URL: Documentation, https://docs.temporal.io/develop/python/integrations/openai-agents
Provides-Extra: mcp
Provides-Extra: openinference
Description-Content-Type: text/markdown

# OpenAI Agents SDK Integration for Temporal

We welcome questions and feedback in the [#python-sdk](https://temporalio.slack.com/archives/CTT84RS0P) Slack channel at [temporalio.slack.com](https://temporalio.slack.com/).

## Install

```bash
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](https://github.com/openai/openai-agents-python) with [Temporal's durable execution](https://docs.temporal.io/evaluate/understanding-temporal#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](#hello-world-durable-agent).** Your first durable agent example.
- **[Background Concepts](#core-concepts).** Background on durable execution and AI agents.
- **[Full Example](#full-example)** Running the Hello World Durable Agent example.
- **[Tool Calling](#tool-calling).** Calling agent Tools in Temporal.
- **[Sandbox Support](#sandbox-support).** Running sandbox agents in Temporal.
- **[Feature Support](#feature-support).** Compatibility matrix.

The [samples repository](https://github.com/temporalio/samples-python/tree/main/openai_agents) 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`)

```python
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](#full-example).
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.

```text
           +-------------------+
           |     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.

```text
            +---------------------+
            |   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](https://docs.temporal.io/evaluate/understanding-temporal#temporal-application-the-building-blocks) for more information.

## Complete Example

To make the [Hello World durable agent](#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`)

```python
# 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](https://openai.github.io/openai-agents-python/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`)

```python
# 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](https://github.com/temporalio/samples-python/tree/main/openai_agents).

## 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 `FunctionTool`s, 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:

| Tool | Execution | Use for |
| --- | --- | --- |
| `activity_as_tool()` | Temporal activity | External I/O and non-deterministic operations |
| `FunctionTool` / `@function_tool` | Workflow | Deterministic, workflow-safe computation |
| OpenAI-hosted tool | Model provider | Provider-hosted features executed as part of the model invocation |

### 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`.

```python
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.

```python
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.

```python
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:

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

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

```python
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"},
        ),
    },
)
```

```python
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:

```python
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:

```python
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](#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:

```python
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.

```python
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:

```python
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](#sandbox-support) manifest, in place of the value itself:

```python
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

```text
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.

```python
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`:

```python
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:

```python
# 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`](https://github.com/temporalio/sdk-python/blob/main/temporalio/contrib/workflow_streams/README.md)
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:

```python
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`](https://github.com/temporalio/sdk-python/blob/main/temporalio/contrib/workflow_streams/README.md) 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](https://github.com/temporalio/sdk-python/blob/main/temporalio/contrib/workflow_streams/README.md) 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](#streaming) above.
Certain tools are not suitable for a distributed computing environment, so these have been disabled as well.

### Model Providers

| Model Provider | Supported |
| :------------- | :-------: |
| OpenAI         |    Yes    |
| LiteLLM        |    Yes    |

### Model Response format

| Model Response | Supported            |
| :------------- | :------------------: |
| Get Response   |         Yes          |
| Streaming      | Yes (experimental)   |

### Tools

#### Tool Type

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

| Tool Type           | Supported |
| :------------------ | :-------: |
| FunctionTool        |    Yes    |
| LocalShellTool      |    No     |
| WebSearchTool       |    Yes    |
| FileSearchTool      |    Yes    |
| HostedMCPTool       |    Yes    |
| ImageGenerationTool |    Yes    |
| CodeInterpreterTool |    Yes    |
| ShellTool           |    Yes    |
| ComputerTool        |    No     |

#### Tool Context

As described in [Tool Calling](#tool-calling), context propagation is read-only when Temporal activities are used as tools.

| Context Propagation                    | Supported |
| :------------------------------------- | :-------: |
| Activity Tool receives copy of context |    Yes    |
| Activity Tool can update context       |    No     |
| Function Tool received context         |    Yes    |
| Function Tool can update context       |    Yes    |

### 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.

| MCP v2 client transport | Supported |
| :---------------------- | :-------: |
| Stdio                   |    Yes    |
| Streamable HTTP         |    Yes    |
| In-process server       |    Yes    |
| Custom transport        |    Yes    |

### Guardrails

| Guardrail Type | Supported |
| :------------- | :-------: |
| Code           |    Yes    |
| Agent          |    Yes    |

### Sessions

SQLite storage is not suited to a distributed environment.

| Feature       | Supported |
| :------------ | :-------: |
| SQLiteSession |    No     |

### Tracing

| Tracing Provider | Supported |
| :--------------- | :-------: |
| OpenAI platform  |    Yes    |

## 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`:

```python
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:

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

Choose the appropriate OTEL exporter for your monitoring system:

```bash
# 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

```python
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:

```python
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:

```python
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

```python
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:

```python
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

| Mode                     | Supported     |
| :----------------------- | :-----------: |
| Voice agents (pipelines) | Yes [^voice]  |
| Realtime agents          |      No       |

[^voice]: `VoicePipeline` runs in your process and delegates the agent
    step (`VoiceWorkflowBase.run`) to a Temporal workflow that uses
    `Runner.run` or `Runner.run_streamed`. STT and TTS run outside
    Temporal; the agent loop is durable.

### Utilities

The REPL utility is not suitable for a distributed setting.

| Utility | Supported |
| :------ | :-------: |
| REPL    |    No     |

## Additional Examples

You can find additional examples in the [Temporal Python Samples Repository](https://github.com/temporalio/samples-python/tree/main/openai_agents).
