diff --git a/docs/develop/python/index.mdx b/docs/develop/python/index.mdx index 756a8c1cbc..44da3b2c41 100644 --- a/docs/develop/python/index.mdx +++ b/docs/develop/python/index.mdx @@ -92,7 +92,7 @@ From there, you can dive deeper into any of the Temporal primitives to start bui - [Langfuse integration](https://langfuse.com/integrations/frameworks/temporal) - [LangGraph integration](/develop/python/integrations/langgraph) - [LangSmith integration](/develop/python/integrations/langsmith) -- [OpenAI Agents SDK integration](https://github.com/temporalio/sdk-python/blob/main/temporalio/contrib/openai_agents/README.md) +- [OpenAI Agents SDK integration](/develop/python/integrations/openai-agents) - [OpenBox integration](https://docs.openbox.ai/getting-started/temporal) - [Parseable integration](https://github.com/parseablehq/temporal-plugin-python/blob/main/INTEGRATION.MD) - [Pydantic AI integration](https://ai.pydantic.dev/durable_execution/temporal/) diff --git a/docs/develop/python/integrations/openai-agents.mdx b/docs/develop/python/integrations/openai-agents.mdx new file mode 100644 index 0000000000..0ed1db1adc --- /dev/null +++ b/docs/develop/python/integrations/openai-agents.mdx @@ -0,0 +1,861 @@ +--- +id: openai-agents +title: OpenAI Agents SDK integration +sidebar_label: OpenAI Agents SDK integration +toc_max_heading_level: 3 +tags: + - OpenAI Agents SDK + - Python SDK + - Temporal SDKs +description: Run OpenAI Agents SDK agents as durable Temporal Workflows in Python, with model calls executed as Activities. +--- + +Temporal's integration with the [OpenAI Agents SDK for Python](https://openai.github.io/openai-agents-python/) lets you +run agents as Temporal Workflows. Agent orchestration—the agent loop, tool selection, and handoffs—runs inside the +Workflow, while model calls run as [Activities](/glossary#activity). + +Like with other types of API calls, in a [Temporal Application](/glossary#temporal-application), you make LLM calls in +your Activities. This integration handles that for you: model calls are executed as Activities, so they retry durably +and are not repeated during Workflow replay. Your agents survive Worker restarts and can run for extended periods +without losing state. + +## Prerequisites + +- This guide assumes you are already familiar with the OpenAI Agents SDK. If you aren't, refer to the + [OpenAI Agents SDK documentation](https://openai.github.io/openai-agents-python/) for more details. +- If you are new to Temporal, we recommend you read the [Understanding Temporal](/evaluate/understanding-temporal) + document or take the [Temporal 101](https://learn.temporal.io/courses/temporal_101/) course to understand the basics + of Temporal. +- Ensure you have set up your local development environment by following the + [Set up your local with the Python SDK](/develop/python/set-up-your-local-python) guide. When you are done, leave the + Temporal Development Server running if you want to test your code locally. + +## Install + +```bash +uv add "temporalio[openai-agents]" +``` + +The extra pulls in `openai-agents` and `mcp` alongside the Temporal SDK. + +Two import paths cover most applications. `temporalio.contrib.openai_agents` holds what you configure on the Worker and +Client—`OpenAIAgentsPlugin`, `ModelActivityParameters`, and the MCP and sandbox providers. +`temporalio.contrib.openai_agents.workflow` holds what you call from inside a Workflow, such as `activity_as_tool` and +the MCP server handles. + +## Run your first durable agent + +A Temporal-backed agent needs three pieces: a Workflow that runs the agent, a Worker configured with the integration +plugin, and a Client configured with the same plugin. + +### Write the Workflow + +Inside the Workflow, write ordinary OpenAI Agents SDK code. The plugin redirects `Runner.run` so that each model call +becomes an Activity—there is no Temporal-specific runner to learn. + + +[openai_agents/basic/workflows/hello_world_workflow.py](https://github.com/temporalio/samples-python/blob/main/openai_agents/basic/workflows/hello_world_workflow.py) +```py +from agents import Agent, Runner +from temporalio import workflow + + +@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 + + +``` + + +### Configure the Worker + +Register `OpenAIAgentsPlugin` on the Client. The plugin registers the model Activity, configures the Pydantic data +converter, propagates tracing context, and registers any configured MCP server or sandbox providers. + + +[openai_agents/basic/run_worker.py](https://github.com/temporalio/samples-python/blob/main/openai_agents/basic/run_worker.py) +```py +client = await Client.connect( + "localhost:7233", + plugins=[ + OpenAIAgentsPlugin( + model_params=ModelActivityParameters( + start_to_close_timeout=timedelta(seconds=30) + ) + ), + ], +) +``` + + +Workers built from that Client pick up the plugin automatically, so all a Worker has to do is register the Workflow. + +`ModelActivityParameters` controls how the model Activity is scheduled. Alongside `start_to_close_timeout`, which +defaults to 60 seconds, it accepts `schedule_to_close_timeout`, `schedule_to_start_timeout`, `heartbeat_timeout`, +`retry_policy`, `cancellation_type`, `versioning_intent`, `task_queue`, `priority`, `summary_override`, +`use_local_activity`, `streaming_topic`, and `streaming_batch_interval`. + +You must ensure the Worker process has access to your model-provider credentials. Most provider SDKs read credentials +from environment variables. + +### Start the Workflow + +Attach the same plugin to the Client that starts the Workflow, so payloads are converted the same way on both sides. + + +[openai_agents/basic/run_hello_world_workflow.py](https://github.com/temporalio/samples-python/blob/main/openai_agents/basic/run_hello_world_workflow.py) +```py +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="openai-agents-basic-task-queue", +) +print(f"Result: {result}") +``` + + +## Tools + +Where a tool runs depends on how you define it. + +| Tool | Runs in | Use for | +| :---------------------------------- | :--------------- | :--------------------------------------------------------- | +| `activity_as_tool()` | Temporal Activity | External I/O and other non-deterministic work | +| `FunctionTool` / `@function_tool` | Workflow | Deterministic, Workflow-safe computation | +| OpenAI-hosted tool | Model provider | Provider-hosted features run as part of the model call | + +Model calls are always routed through Activities. Tools are not: a `@function_tool` runs in the Workflow unless you back +it with an Activity, so any tool that performs I/O needs `activity_as_tool()` or a Nexus Operation. + +### Activity-backed tools + +Use `activity_as_tool` for HTTP calls, database access, file system work, or other I/O. Write an ordinary Temporal +Activity: + + +[openai_agents/basic/activities/get_weather_activity.py](https://github.com/temporalio/samples-python/blob/main/openai_agents/basic/activities/get_weather_activity.py) +```py +from dataclasses import dataclass + +from temporalio import activity + + +@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.") + + +``` + + +Then pass it through `activity_as_tool` when you build the agent: + + +[openai_agents/basic/workflows/tools_workflow.py](https://github.com/temporalio/samples-python/blob/main/openai_agents/basic/workflows/tools_workflow.py) +```py +@workflow.defn +class ToolsWorkflow: + @workflow.run + async def run(self, question: str) -> str: + agent = Agent( + name="Hello world", + instructions="You are a helpful agent.", + tools=[ + temporal_agents.workflow.activity_as_tool( + get_weather, start_to_close_timeout=timedelta(seconds=10) + ) + ], + ) + + result = await Runner.run(agent, input=question) + return result.final_output + + +``` + + +`activity_as_tool` controls how the agent invokes the Activity; it does not register the Activity with the Worker. Pass +the Activity function to the Worker's `activities` argument as well. + +Because the Activity may not run in the same process as the Workflow, an Activity-backed tool receives a copy of the +agent context and cannot mutate it. A tool that runs in the Workflow can. + +### Inline and hosted tools + +For deterministic computation, use the standard `@function_tool` decorator and call it directly from the Workflow. Do +not perform network, database, or file system I/O from these tools—use `activity_as_tool` instead. + +Hosted tools such as `WebSearchTool`, `FileSearchTool`, `CodeInterpreterTool`, and `ImageGenerationTool` are executed by +the model provider during the model Activity, so they need no extra wiring: + + +[openai_agents/tools/workflows/web_search_workflow.py](https://github.com/temporalio/samples-python/blob/main/openai_agents/tools/workflows/web_search_workflow.py) +```py +@workflow.defn +class WebSearchWorkflow: + @workflow.run + async def run(self, question: str, user_city: str = "New York") -> str: + agent = Agent( + name="Web searcher", + instructions="You are a helpful agent.", + tools=[ + WebSearchTool(user_location={"type": "approximate", "city": user_city}) + ], + ) + + result = await Runner.run(agent, question) + return result.final_output + + +``` + + +`LocalShellTool` and `ComputerTool` are not supported, because they assume a single long-lived local process. + +### Nexus operation tools + +Use `nexus_operation_as_tool` to expose a [Nexus](/nexus) Operation as an agent tool. The Workflow starts the Operation +through a Nexus client and feeds the result back to the agent, which lets an agent call across a Namespace boundary: + +```python +from temporalio.contrib.openai_agents.workflow import nexus_operation_as_tool + +weather_tool = nexus_operation_as_tool( + WeatherService.get_weather, + service=WeatherService, + endpoint="weather-endpoint", +) +``` + +### Nested agent tools + +`Agent.as_tool()` from the OpenAI Agents SDK works unchanged. The nested agent's model calls become Activities like any +others, so a multi-agent run stays durable throughout: + + +[openai_agents/agent_patterns/workflows/agents_as_tools_workflow.py](https://github.com/temporalio/samples-python/blob/main/openai_agents/agent_patterns/workflows/agents_as_tools_workflow.py) +```py +def orchestrator_agent() -> Agent: + spanish_agent = Agent( + name="spanish_agent", + instructions="You translate the user's message to Spanish", + handoff_description="An english to spanish translator", + ) + + french_agent = Agent( + name="french_agent", + instructions="You translate the user's message to French", + handoff_description="An english to french translator", + ) + + italian_agent = Agent( + name="italian_agent", + instructions="You translate the user's message to Italian", + handoff_description="An english to italian translator", + ) + + orchestrator_agent = Agent( + name="orchestrator_agent", + instructions=( + "You are a translation agent. You use the tools given to you to translate." + "If asked for multiple translations, you call the relevant tools in order." + "You never translate on your own, you always use the provided tools." + ), + tools=[ + spanish_agent.as_tool( + tool_name="translate_to_spanish", + tool_description="Translate the user's message to Spanish", + ), + french_agent.as_tool( + tool_name="translate_to_french", + tool_description="Translate the user's message to French", + ), + italian_agent.as_tool( + tool_name="translate_to_italian", + tool_description="Translate the user's message to Italian", + ), + ], + ) + return orchestrator_agent + + +``` + + +## MCP servers + +Temporal's durability does not extend to [MCP](https://modelcontextprotocol.io/) servers, which run independently of the +Workflow. The integration offers two wrappers so you can pick the one that matches how your server behaves. + +A **stateless** server treats each operation as independent—`get_weather(location)` carries everything it needs—so it +can be reconnected to without changing behavior. A **stateful** server keeps session state between calls, as a server +where `set_location(location)` precedes `get_weather()` does, and loses that state if the session drops. Prefer +stateless when you have the choice: its durability guarantees are stronger. + +:::warning + +Both `stateless_mcp_server()` and `stateful_mcp_server()` accept a `factory_argument` that is passed to the registered +factory. It is an Activity argument, so it is recorded in Workflow history and, without a payload codec, visible in the +Web UI. Do not pass secrets, credentials, or API keys through it—resolve those Worker-side inside the server factory. + +::: + +### Stateless MCP servers + +Register a `StatelessMCPServerProvider` with a factory that creates the server, and give it a name: + + +[openai_agents/mcp/run_file_system_worker.py](https://github.com/temporalio/samples-python/blob/main/openai_agents/mcp/run_file_system_worker.py) +```py +file_system_server = StatelessMCPServerProvider( + "FileSystemServer", + lambda: MCPServerStdio( + name="FileSystemServer", + params={ + "command": "npx", + "args": ["-y", "@modelcontextprotocol/server-filesystem", samples_dir], + }, + ), +) + +# Create client connected to server at the given address +config = ClientConfig.load_client_connect_config() +config.setdefault("target_host", "localhost:7233") +client = await Client.connect( + **config, + plugins=[ + OpenAIAgentsPlugin( + model_params=ModelActivityParameters( + start_to_close_timeout=timedelta(seconds=60) + ), + mcp_server_providers=[file_system_server], + ), + ], +) +``` + + +Reference the same name from Workflow code with `stateless_mcp_server`: + + +[openai_agents/mcp/workflows/file_system_workflow.py](https://github.com/temporalio/samples-python/blob/main/openai_agents/mcp/workflows/file_system_workflow.py) +```py +server: MCPServer = openai_agents.workflow.stateless_mcp_server( + "FileSystemServer" +) +agent = Agent( + name="Assistant", + instructions="Use the tools to read the filesystem and answer questions based on those files.", + mcp_servers=[server], +) +``` + + +### Stateful MCP servers + +Register a `StatefulMCPServerProvider` instead. The plugin runs a dedicated Worker that holds the connection open for +the life of the Workflow run. + + +[openai_agents/mcp/run_memory_research_scratchpad_worker.py](https://github.com/temporalio/samples-python/blob/main/openai_agents/mcp/run_memory_research_scratchpad_worker.py) +```py +memory_server_provider = StatefulMCPServerProvider( + "MemoryServer", + lambda _: MCPServerStdio( + name="MemoryServer", + params={ + "command": "npx", + "args": ["-y", "@modelcontextprotocol/server-memory"], + }, + ), +) + +# Create client connected to server at the given address +config = ClientConfig.load_client_connect_config() +config.setdefault("target_host", "localhost:7233") +client = await Client.connect( + **config, + plugins=[ + OpenAIAgentsPlugin( + model_params=ModelActivityParameters( + start_to_close_timeout=timedelta(seconds=60) + ), + mcp_server_providers=[memory_server_provider], + ), + ], +) +``` + + +In the Workflow, `stateful_mcp_server` is an async context manager, which ties the session's lifetime to the block: + + +[openai_agents/mcp/workflows/memory_research_scratchpad_workflow.py](https://github.com/temporalio/samples-python/blob/main/openai_agents/mcp/workflows/memory_research_scratchpad_workflow.py) +```py +async with temporal_openai_agents.workflow.stateful_mcp_server( + "MemoryServer", +) as server: + with trace(workflow_name="MCP Memory Scratchpad Example"): + agent = Agent( + name="Research Scratchpad Agent", + instructions=( + "Use the Memory MCP tools to persist, query, update, and delete notes." + " Keep IDs short and consistent. Synthesis must rely only on recalled notes and include simple" + " citations of the form '(Note: id)'. Keep the brief to 5 bullets." + ), + mcp_servers=[server], + model_settings=ModelSettings(tool_choice="required"), + ) +``` + + +If the dedicated Worker fails—a network problem, or the server itself going away—the session state is gone and Temporal +cannot recreate it. The integration raises an `ApplicationError` so your Workflow can decide what to do; recovering +means retrying at the application level, not relying on Activity retries. + +### Hosted MCP tool + +For a network-accessible server, `HostedMCPTool` uses an MCP client hosted by OpenAI, so there is nothing to register +on the Worker: + + +[openai_agents/hosted_mcp/workflows/simple_mcp_workflow.py](https://github.com/temporalio/samples-python/blob/main/openai_agents/hosted_mcp/workflows/simple_mcp_workflow.py) +```py +@workflow.defn +class SimpleMCPWorkflow: + @workflow.run + async def run( + self, question: str, server_url: str = "https://gitmcp.io/openai/codex" + ) -> str: + agent = Agent( + name="Assistant", + tools=[ + HostedMCPTool( + tool_config={ + "type": "mcp", + "server_label": "gitmcp", + "server_url": server_url, + "require_approval": "never", + } + ) + ], + ) + + result = await Runner.run(agent, question) + return result.final_output + + +``` + + +## Conversation history and human-in-the-loop + +Because the agent loop runs inside a Workflow, conversation history and pending approvals have to be replay safe. + +### Carry conversation history across turns + +Hold the history in Workflow state and rebuild each turn's input from the previous result. `result.to_input_list()` +returns the conversation as input items, and `result.last_agent` records which agent a handoff left you on: + +```python +self.input_items.append({"content": user_input, "role": "user"}) +result = await Runner.run(self.current_agent, self.input_items) +self.input_items = result.to_input_list() +self.current_agent = result.last_agent +``` + +Workflow state is replay safe, so history survives Worker restarts within a run. `SQLiteSession` is not supported: it +keeps history in a local file, which no longer identifies one conversation once Workers are distributed. + +### Handle long-running conversations + +A chat-style Workflow accumulates history with every turn, and over a long session the event history can grow large +enough to hit Temporal's per-Workflow limit. Use +[Continue-as-New](/develop/python/workflows/continue-as-new) to start a fresh Execution carrying the history forward. + +In this example each user turn arrives as a Workflow +[Update](/develop/python/workflows/message-passing#updates), so the caller gets the agent's reply back from the same +call. The `run` method waits until Temporal suggests continuing, drains in-flight handlers, then continues as new with +the accumulated state: + + +[openai_agents/customer_service/workflows/customer_service_workflow.py](https://github.com/temporalio/samples-python/blob/main/openai_agents/customer_service/workflows/customer_service_workflow.py) +```py +@workflow.run +async def run( + self, customer_service_state: CustomerServiceWorkflowState | None = None +): + await workflow.wait_condition( + lambda: workflow.info().is_continue_as_new_suggested() + and workflow.all_handlers_finished() + ) + workflow.continue_as_new( + CustomerServiceWorkflowState( + printed_history=self.printed_history, + current_agent_name=self.current_agent.name, + context=self.context, + input_items=self.input_items, + ) + ) + +``` + + +### Add human approval + +An agent action that should not proceed unattended can pause for a person. A `HostedMCPTool` configured with +`require_approval` calls your `on_approval_request` callback, which runs in Workflow context—so it must be +deterministic, and it can wait on a Signal or Update to get the answer from outside: + + +[openai_agents/hosted_mcp/workflows/approval_mcp_workflow.py](https://github.com/temporalio/samples-python/blob/main/openai_agents/hosted_mcp/workflows/approval_mcp_workflow.py) +```py +def approval_callback(request: MCPToolApprovalRequest) -> MCPToolApprovalFunctionResult: + """Simple approval callback that logs the request and approves by default. + + In a real application, user input would be provided through a UI or API. + The approval callback executes within the Temporal workflow, so the application + can use signals or updates to receive user input. + """ + workflow.logger.info(f"MCP tool approval requested for: {request.data.name}") + + result: MCPToolApprovalFunctionResult = {"approve": True} + return result + + +``` + + +## Sandbox + +:::caution + +Sandbox support is pre-release and may change before general availability. + +::: + +`SandboxAgent` gives an agent a machine to work on: a shell to run commands in and a filesystem to read and write. The +plugin dispatches every sandbox operation—creating the session, each command, each read and write, PTY interaction, and +teardown—as its own Temporal Activity. Each one is individually retryable and visible in Workflow history, and the +session state is serialized with the Workflow, so a Worker restart part-way through a run resumes against the same +session. + +Register a `SandboxClientProvider` for each backend you want to reach, under a unique name: + + +[openai_agents/sandbox/run_worker.py](https://github.com/temporalio/samples-python/blob/main/openai_agents/sandbox/run_worker.py) +```py +client = await Client.connect( + "localhost:7233", + plugins=[ + OpenAIAgentsPlugin( + model_params=ModelActivityParameters( + start_to_close_timeout=timedelta(seconds=60) + ), + # The plugin registers one set of sandbox activities per + # provider, prefixed with the provider name. Register several + # providers to let one worker serve several backends. + sandbox_clients=[ + SandboxClientProvider(SANDBOX_PROVIDER, UnixLocalSandboxClient()), + ], + ), + ], +) +``` + + +In the Workflow, `temporal_sandbox_client()` resolves a name to that backend and goes in the `RunConfig`: + + +[openai_agents/sandbox/workflows/local_sandbox_workflow.py](https://github.com/temporalio/samples-python/blob/main/openai_agents/sandbox/workflows/local_sandbox_workflow.py) +```py +@workflow.defn +class LocalSandboxWorkflow: + @workflow.run + async def run(self, prompt: str) -> str: + # A default SandboxAgent already carries the Filesystem, Shell, and + # Compaction capabilities, so there are no tools to declare here. + agent = SandboxAgent[None]( + name="Sandbox Assistant", + instructions=( + "You have a sandbox with a shell and a filesystem. Use it to do " + "the work rather than answering from memory, then report what " + "the commands returned." + ), + ) + + result = await Runner.run( + starting_agent=agent, + input=prompt, + run_config=RunConfig( + sandbox=SandboxRunConfig( + # Must match the name registered on the worker. + client=temporal_sandbox_client(SANDBOX_PROVIDER), + options=UnixLocalSandboxClientOptions(), + ), + ), + ) + return result.final_output_as(str, raise_if_incorrect_type=True) + + +``` + + +The name becomes the prefix of that backend's Activity names, which is what lets several backends share one Worker. A +single Workflow can target more than one by calling `temporal_sandbox_client()` once per name. Names must match the +Worker's registration exactly. + +The sample above uses `UnixLocalSandboxClient`, which runs commands on the Worker host and copies the Worker's entire +environment into each agent-run process. An agent can read values such as `OPENAI_API_KEY`. Command output returns as an +Activity result and is stored in Workflow history unless you protect payloads with a codec. + +Use `UnixLocalSandboxClient` only for local development with trusted prompts. In production, register a remote client +such as `DaytonaSandboxClient` or `E2BSandboxClient` from `agents.extensions.sandbox` instead. Only the Worker changes; +the Workflow still names a provider. + +## Streaming + +:::caution + +Streaming is experimental and may change before general availability. + +::: + +`Runner.run_streamed` works inside a Workflow. The model call runs as a streaming Activity that consumes +`Model.stream_response` and publishes each event to a [Workflow Stream](/workflow-streams) topic as the model produces +it, so an external client can watch a run live while it stays durable. + +Set the topic on `ModelActivityParameters.streaming_topic` and host a `WorkflowStream` in the Workflow. The topic is +required: without it, `run_streamed` raises before scheduling any Activity. + + +[openai_agents/streaming/workflows/stream_text_workflow.py](https://github.com/temporalio/samples-python/blob/main/openai_agents/streaming/workflows/stream_text_workflow.py) +```py +@workflow.defn +class StreamTextWorkflow: + @workflow.init + def __init__(self, input: StreamTextInput) -> None: + # WorkflowStream requires construction from a method named __init__ + # (it checks its caller's frame and raises otherwise), and + # @workflow.init is what makes the run argument — and the + # stream_state it carries across continue-as-new — available here. + self.stream = WorkflowStream(prior_state=input.stream_state) + self.done = self.stream.topic(TOPIC_DONE, type=bool) + + @workflow.run + async def run(self, input: StreamTextInput) -> str: + agent = Agent( + name="Joker", + instructions="You are a helpful assistant.", + ) + result = Runner.run_streamed(agent, input=input.prompt) + + # The workflow only sees these events once the activity returns, so + # the loop just counts them. External subscribers receive them as the + # activity publishes them. + deltas = 0 + async for event in result.stream_events(): + if event.type == "raw_response_event" and isinstance( + event.data, ResponseTextDeltaEvent + ): + deltas += 1 + workflow.logger.info("collected %d delta events", deltas) + + # In-band terminator so the subscriber can stop without racing the + # workflow's completion, then a brief pause to let its next poll + # deliver the tail of the stream — the log lives in workflow memory + # and is gone once this run completes. + self.done.publish(True) + await workflow.sleep(DRAIN_INTERVAL) + # final_output is typed Any and is None when a run ends without + # message output, so assert the str this signature promises rather + # than letting a None through. + return result.final_output_as(str, raise_if_incorrect_type=True) + + +``` + + +Subscribe from outside with `WorkflowStreamClient`: + + +[openai_agents/streaming/run_stream_text_workflow.py](https://github.com/temporalio/samples-python/blob/main/openai_agents/streaming/run_stream_text_workflow.py) +```py +stream = WorkflowStreamClient.create(client, workflow_id) +converter = client.data_converter.payload_converter + +# A single iterator over both topics — one subscriber, no cancellation race +# between concurrent ones. result_type=RawValue delivers the underlying +# Payload so heterogeneous topics can be decoded per item.topic. The loop +# ends on the in-band terminator, or by the iterator exhausting if the +# workflow reaches a terminal state without publishing one (e.g. on +# failure); either way handle.result() below surfaces the outcome. +last_sequence = -1 +response_in_flight = False +async for item in stream.subscribe( + [TOPIC_EVENTS, TOPIC_DONE], result_type=RawValue +): + if item.topic == TOPIC_DONE: + break + # Subscribers receive native OpenAI events, not the agents-SDK + # StreamEvent wrappers that stream_events() yields in the workflow. + event: Any = converter.from_payload(item.data.payload, EVENT_TYPE) + + # Every event carries a sequence_number that starts at 0 per response, + # so a number that does not advance means a new response is streaming. + # That is a retry only if the previous one never completed: each turn + # of a multi-turn run is its own response and restarts the count too. + # The retry is an independently sampled answer rather than a + # continuation, so mark the seam instead of letting the failed + # attempt's partial text run into the new one. The workflow's return + # value is unaffected — stream_events() there sees only the attempt + # that succeeded. + sequence = event.sequence_number + if sequence <= last_sequence and response_in_flight: + print("\n\n[model activity retried — output restarts here]\n") + last_sequence = sequence + response_in_flight = not isinstance(event, ResponseCompletedEvent) + + if isinstance(event, ResponseTextDeltaEvent): + print(event.delta, end="", flush=True) +``` + + +Two things to know: + +- Streaming is incompatible with `use_local_activity`, because Local Activities support neither heartbeats nor the + Workflow stream signal channel. +- Activity retries are visible to stream subscribers but not to `stream_events()`. An attempt that fails mid-response + leaves its events on the stream and the retry publishes a second sequence, while `stream_events()` sees only the + successful attempt's collected events. + +## Tracing + +OpenAI Agents SDK tracing works across Client, Workflow, and Activity boundaries, with the plugin propagating trace +context for you. + +### OpenAI hosted traces + +Hosted tracing needs no setup beyond the plugin. To start a trace on the Client—so the whole Workflow Execution is part +of a larger trace—open `plugin.tracing_context()` first: + +```python +plugin = OpenAIAgentsPlugin() +client = await Client.connect("localhost:7233", plugins=[plugin]) + +with plugin.tracing_context(): + with trace("Customer support workflow"): + result = await client.execute_workflow( + CustomerSupportAgent.run, + "Help me with my order", + id="customer-support-123", + task_queue="my-task-queue", + ) +``` + +`plugin.tracing_context()` is required when starting traces outside a Worker; without it the trace does not propagate +into the Workflow. + +### OpenTelemetry + +:::caution + +The OpenTelemetry integration is Public Preview and may change before general availability. + +::: + +If you already collect traces with OpenTelemetry, the integration can emit the agent's spans through your pipeline, so +model calls, tools, and orchestration land in the same backend as the rest of your traces. + +Install the additional dependencies: + +```bash +uv add openinference-instrumentation-openai-agents opentelemetry-sdk opentelemetry-exporter-otlp +``` + +Then set a global replay-safe tracer provider before connecting the Client, and turn on instrumentation in the plugin: + +```python +from opentelemetry import trace +from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter +from opentelemetry.sdk.trace.export import SimpleSpanProcessor +from temporalio.contrib.opentelemetry import create_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)], +) +``` + +The provider must come from `create_tracer_provider()`. It replays safely, exporting spans only when a Workflow actually +completes rather than on every replay, and generates deterministic span identifiers so they correlate across replays. +Passing any other provider raises a `ValueError`. + +To call the OpenTelemetry API directly from Workflow code, allow the module through the Workflow sandbox and open an +agents-SDK span first, so your spans are parented rather than becoming roots: + +```python +import opentelemetry.trace + +from agents import custom_span +from temporalio.worker.workflow_sandbox import SandboxedWorkflowRunner, SandboxRestrictions + +worker = Worker( + client, + task_queue="my-task-queue", + workflows=[MyWorkflow], + workflow_runner=SandboxedWorkflowRunner( + SandboxRestrictions.default.with_passthrough_modules("opentelemetry") + ), +) + +# Inside the Workflow: +with custom_span("Workflow coordination"): + tracer = opentelemetry.trace.get_tracer(__name__) + with tracer.start_as_current_span("Custom workflow span"): + ... +``` + +## Resources + +- [OpenAI Agents SDK samples](https://github.com/temporalio/samples-python/tree/main/openai_agents) — runnable examples + for the patterns in this guide. +- [`temporalio.contrib.openai_agents` README](https://github.com/temporalio/sdk-python/blob/main/temporalio/contrib/openai_agents/README.md) + — the full plugin reference, including the complete feature-support matrix. +- [OpenAI Agents SDK for Python](https://openai.github.io/openai-agents-python/) +- [Temporal Plugins guide](/develop/plugins-guide) — the Plugin system this integration is built on, which you can also + use to build your own integrations. diff --git a/sidebars.js b/sidebars.js index cf8ab67006..6e0856d747 100644 --- a/sidebars.js +++ b/sidebars.js @@ -718,6 +718,7 @@ const developPythonCategory = { 'develop/python/integrations/google-genai', 'develop/python/integrations/langgraph', 'develop/python/integrations/langsmith', + 'develop/python/integrations/openai-agents', 'develop/python/integrations/strands-agents', ], }, diff --git a/src/components/IntegrationsGrid/integrations-data.json b/src/components/IntegrationsGrid/integrations-data.json index ee7831a1e7..ebd8c6d781 100644 --- a/src/components/IntegrationsGrid/integrations-data.json +++ b/src/components/IntegrationsGrid/integrations-data.json @@ -159,7 +159,7 @@ "Agent framework" ], "sdk": "Python", - "href": "https://github.com/temporalio/sdk-python/blob/main/temporalio/contrib/openai_agents/README.md" + "href": "/develop/python/integrations/openai-agents" }, { "name": "OpenAI Agents SDK",