Temporal Python SDK

For simple, deterministic computations that do not require external I/O, you can define tools directly within the workflow using the standard OpenAI Agents SDK @functiontool annotation.

Important Restrictions:

• Tools running directly in the workflow must respect workflow execution restrictions (no I/O or non-deterministic operations).
• Tools running in the workflow can update OpenAI Agents context (which is read-only for tools run as Temporal activities).
• If a tool needs to perform I/O, it should be implemented as a Temporal Activity and converted via activity_as_tool instead.

from temporalio import workflow
from agents import Agent, Runner, 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="Use the available tools to help with calculations.",
            tools=[calculate_circle_area],
        )
        result = await Runner.run(agent, input=message)
        return result.final_output

When using S3StorageDriver, keep the following configuration behaviors in mind:

• Payload Size Threshold: Only payloads at or above ExternalStorage.payload_size_threshold (default: 256 KiB) are offloaded to S3. Set this to 0 to offload every payload.
• Max Payload Size: S3StorageDriver.max_payload_size (default: 50 MiB) is a hard limit on the serialized size of a single payload. A ValueError is raised if a payload exceeds this. Increase this value if your workflows produce larger payloads.
• Driver Name: Override S3StorageDriver.driver_name only if you are registering multiple S3StorageDriver instances with different configurations in the same ExternalStorage.drivers list.
• Bucket Existence: The target S3 bucket must already exist; the driver will not create it.
• Symmetry: Any driver used to store payloads must also be configured on the component that retrieves them. For example, if a Client uses the driver to store inputs, the Worker must also include that driver in its ExternalStorage.drivers list to read them.

Nexus is a synchronous RPC protocol that allows Temporal to call operations. Temporal supports calling Nexus operations exclusively from within a workflow.

There are two types of supported Nexus operations:

1. @temporalio.nexus.workflow_run_operation: Backed by a Temporal workflow. The handler starts a workflow and returns a token. When the workflow completes, the result is automatically delivered to the caller.
2. @nexusrpc.handler.sync_operation: A synchronous operation (can be def or async def) that must respond within 10 seconds.

To use Nexus, you must:

1. Create namespaces and a Nexus endpoint via the Temporal CLI.
2. Define a service contract using @nexusrpc.service.
3. Implement handlers using @service_handler and the appropriate operation decorator.
4. Register the handler with a Temporal Worker using the nexus_service_handlers argument.
5. Call operations from a workflow using workflow.create_nexus_client.

# 1. Define Service Contract
@nexusrpc.service
class MyNexusService:
    my_sync_operation: nexusrpc.Operation[MyInput, MyOutput]
    my_workflow_run_operation: nexusrpc.Operation[MyInput, MyOutput]

# 2. Implement Handlers
@service_handler(service=MyNexusService)
class MyNexusServiceHandler:
    @sync_operation
    async def my_sync_operation(self, ctx: StartOperationContext, input: MyInput) -> MyOutput:
        return MyOutput(message=f"Hello {input.name}!")

    @workflow_run_operation
    async def my_workflow_run_operation(self, ctx: WorkflowRunOperationContext, input: MyInput) -> nexus.WorkflowHandle[MyOutput]:
        return await ctx.start_workflow(WorkflowStartedByNexusOperation.run, input, id=str(uuid.uuid4()))

# 3. Register with Worker
client = await Client.connect("localhost:7233", namespace="my-handler-namespace")
worker = Worker(
    client,
    task_queue="my-handler-task-queue",
    workflows=[WorkflowStartedByNexusOperation],
    nexus_service_handlers=[MyNexusServiceHandler()],
)
await worker.run()

# 4. Call from Workflow
@workflow.defn
class CallerWorkflow:
    def __init__(self):
        self.nexus_client = workflow.create_nexus_client(
            service=MyNexusService, endpoint="my-nexus-endpoint"
        )

    @workflow.run
    async def run(self, name: str) -> tuple[MyOutput, MyOutput]:
        # Execute and wait for result
        wf_result = await self.nexus_client.execute_operation(
            MyNexusService.my_workflow_run_operation,
            MyInput(name),
        )
        return wf_result

The integration supports Model Context Protocol (MCP) servers via two wrapper types. Because MCP servers operate independently of Temporal, their durability is not automatically managed by Temporal workflows. You must choose the wrapper that matches your server's design:

1. Stateless MCP Servers: Treat each operation independently (e.g., a weather lookup). These are safe to restart or reconnect. Use StatelessMCPServerProvider to register them with the OpenAIAgentsPlugin in the Worker. In the workflow, access them using openai_agents.workflow.stateless_mcp_server("SERVER_NAME").

2. Stateful MCP Servers: Maintain session state between calls (e.g., a server requiring a set_location call before get_weather). If the connection fails, Temporal raises an ApplicationError. Because the server state is lost, you must implement your own application-level retry logic to handle these failures.

Security Warning: When using stateless_mcp_server() or stateful_mcp_server(), you can pass an optional factory_argument. Do not pass secrets, credentials, or API keys through factory_argument, as it is recorded in the Temporal workflow history and may be visible in the Web UI. Resolve credentials inside the server factory instead.

# Worker Configuration for Stateless MCP
from temporalio.contrib.openai_agents import ( 
    ModelActivityParameters, 
    OpenAIAgentsPlugin, 
    StatelessMCPServerProvider
)

filesystem_server = StatelessMCPServerProvider(
    lambda: MCPServerStdio(
        name="FileSystemServer",
        params={
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-filesystem", "/path/to/files"],
        },
    )
)

client = await Client.connect(
    "localhost:7233",
    plugins=[
        OpenAIAgentsPlugin(
            model_params=ModelActivityParameters(start_to_close_timeout=timedelta(seconds=60)),
            mcp_server_providers=[filesystem_server],
        ),
    ],
)

# Workflow Implementation
@workflow.defn
class FileSystemWorkflow:
    @workflow.run
    async def run(self, query: str) -> str:
        server = openai_agents.workflow.stateless_mcp_server("FileSystemServer")
        agent = Agent(
            name="File Assistant",
            instructions="Use the filesystem tools to read files.",
            mcp_servers=[server],
        )
        result = await Runner.run(agent, input=query)
        return result.final_output

The StrandsPlugin allows you to map model names to factory functions. Each factory is called lazily on the worker (outside the workflow sandbox) and the resulting model is cached.

In your workflow, you select a model by passing its name to TemporalAgent(model="name", ...).

If models is omitted from StrandsPlugin, it defaults to a single factory named "bedrock" using BedrockModel().

from strands.models.anthropic import AnthropicModel
from strands.models.bedrock import BedrockModel

# workflow
@workflow.defn
class MultiModelWorkflow:
    def __init__(self) -> None:
        self.agent_a = TemporalAgent(
            model="claude",
            start_to_close_timeout=timedelta(seconds=60),
        )
        self.agent_b = TemporalAgent(
            model="bedrock",
            start_to_close_timeout=timedelta(seconds=60),
        )

# worker
Worker(..., plugins=[StrandsPlugin(models={
    "claude": lambda: AnthropicModel(client_args={"api_key": "..."}),
    "bedrock": lambda: BedrockModel(),
})])