Skip to main content

OpenAI Agents SDK integration

View Markdown

Temporal's integration with the OpenAI Agents SDK for 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.

Like with other types of API calls, in a 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 for more details.
  • If you are new to Temporal, we recommend you read the Understanding Temporal document or take the 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 guide. When you are done, leave the Temporal Development Server running if you want to test your code locally.

Install​

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

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

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 takes retry_policy, task_queue, priority, summary_override, and use_local_activity.

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

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.

ToolRuns inUse for
activity_as_tool()Temporal ActivityExternal I/O and other non-deterministic work
FunctionTool / @function_toolWorkflowDeterministic, Workflow-safe computation
OpenAI-hosted toolModel providerProvider-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

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

@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

@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 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:

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

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

danger

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

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

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

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

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

@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


Secrets for hosted tools​

caution

This feature is experimental and may change before general availability.

A hosted tool's credentials would otherwise have to be written into Workflow code, where they would land in Workflow history. Use temporal_worker_env_ref() to name an environment variable instead of passing the credential itself:

from agents import HostedMCPTool
from temporalio.contrib.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"),
    }
)

Every Worker that runs model Activities must both set MY_MCP_TOKEN and declare it resolvable:

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

Names are matched exactly, with no globbing. A "*" anywhere in the list allows every environment variable on the Worker. A reference can also 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 Worker substitutes the value 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

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:

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 to start a fresh Execution carrying the history forward.

In this example each user turn arrives as a Workflow Update, 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

@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

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

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

@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—convenient locally, but it means the agent gets a shell on that machine. In production, register a remote client such as DaytonaSandboxClient or E2BSandboxClient from agents.extensions.sandbox instead. Only the Worker changes; the Workflow still just 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 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

@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

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:

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​

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:

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:

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:

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​