Skip to main content

Strands Agents integration

View Markdown

Temporal's integration with Strands Agents is an SDK Plugin that gives your Strands agents Durable Execution via the Temporal platform. The plugin routes model invocations, tool calls, MCP tool calls, and hooks through Temporal Activities, so every step your agent takes is recorded in Workflow history and can survive crashes, restarts, and infrastructure failures.

Code snippets in this guide are taken from the Strands Agents plugin samples. Refer to the samples for the complete code.

Get started​

Install the plugin, then run a minimal Strands agent inside a Temporal Workflow.

Prerequisites​

Install the plugin​

Install the Temporal Python SDK with Strands Agents support (requires temporalio 1.28.0 or later):

uv add "temporalio[strands-agents]"

or with pip:

pip install "temporalio[strands-agents]"

Run a Strands agent with Durable Execution​

The following example runs a Strands agent inside a Temporal Workflow. Model calls execute as Temporal Activities, which means they get automatic retries, timeouts, and durable execution. If the Worker process crashes mid-conversation, Temporal replays the Workflow and resumes from the last completed Activity.

1. Define the Workflow

Create a Workflow that holds a TemporalAgent and invokes it with a prompt. The start_to_close_timeout sets the maximum time each model call Activity can run:

strands_plugin/hello_world/workflow.py

from datetime import timedelta

from temporalio import workflow
from temporalio.contrib.strands import TemporalAgent


@workflow.defn
class HelloWorldWorkflow:
    def __init__(self) -> None:
        self.agent = TemporalAgent(start_to_close_timeout=timedelta(seconds=60))

    @workflow.run
    async def run(self, prompt: str) -> str:
        result = await self.agent.invoke_async(prompt)
        return str(result)


caution

Inside a Workflow, always call agent.invoke_async(message), not agent(message). The synchronous form spawns a worker thread, which the Workflow sandbox blocks.

2. Start a Worker

Create a Worker that registers the Workflow and the StrandsPlugin. The plugin automatically registers the Activities that handle model calls:

strands_plugin/hello_world/run_worker.py

import asyncio
import os

from temporalio.client import Client
from temporalio.contrib.strands import StrandsPlugin
from temporalio.worker import Worker

from strands_plugin.hello_world.workflow import HelloWorldWorkflow


async def main() -> None:
    plugin = StrandsPlugin()
    client = await Client.connect(
        os.environ.get("TEMPORAL_ADDRESS", "localhost:7233"),
        plugins=[plugin],
    )

    worker = Worker(
        client,
        task_queue="strands-hello-world",
        workflows=[HelloWorldWorkflow],
    )
    print("Worker started. Ctrl+C to exit.")
    await worker.run()


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

3. Run the Workflow

Start the Workflow from a separate client script. This example sends the prompt "Write a haiku about durable execution" and prints the agent's response:

strands_plugin/hello_world/run_workflow.py

import asyncio
import os

from temporalio.client import Client

from strands_plugin.hello_world.workflow import HelloWorldWorkflow


async def main() -> None:
    client = await Client.connect(os.environ.get("TEMPORAL_ADDRESS", "localhost:7233"))

    result = await client.execute_workflow(
        HelloWorldWorkflow.run,
        "Write a haiku about durable execution.",
        id="strands-hello-world",
        task_queue="strands-hello-world",
    )

    print(f"Result: {result}")


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

Build the agent​

Customize which model provider your agent uses, add tools that run as Activities, subscribe to lifecycle events with hooks, and connect to MCP servers.

Choose and configure models​

By default, StrandsPlugin uses Strands' own default model (BedrockModel). To use a different model, pass a models mapping to StrandsPlugin on the Worker. When you provide a custom models mapping, each TemporalAgent must specify which model to use by name.

Each entry in the mapping pairs a name with a factory function that creates a model provider (such as AnthropicModel or BedrockModel). The provider is created on first use and reused for the Worker's lifetime:

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(),
})])

Each TemporalAgent carries its own Activity options (timeouts, retry policy, task queue, streaming topic) and dispatches to a shared model Activity, which resolves the model name against the registered factories at runtime. A model name not present in the models mapping raises ValueError inside the Activity.

Run non-deterministic tools as Activities​

Strands tools that perform I/O, access external services, or produce non-deterministic results need to run as Temporal Activities rather than inline in the Workflow. Wrap each tool in an @activity.defn function, register the Activities on the Worker, and pass them to the agent using activity_as_tool.

Define an Activity for the tool:

strands_plugin/tools/workflow.py

@activity.defn
async def fetch_weather(city: str) -> dict:
    """Stub weather lookup — replace with a real HTTP call in production."""
    return {
        "city": city,
        "temperature_f": 72,
        "conditions": "sunny",
    }


Pass the Activity to the agent in the Workflow using activity_as_tool:

strands_plugin/tools/workflow.py

@workflow.defn
class ToolsWorkflow:
    def __init__(self) -> None:
        self.agent = TemporalAgent(
            start_to_close_timeout=timedelta(seconds=60),
            tools=[
                letter_counter,
                activity_as_tool(
                    fetch_weather,
                    start_to_close_timeout=timedelta(seconds=30),
                ),
                activity_as_tool(
                    environment_activity,
                    start_to_close_timeout=timedelta(seconds=30),
                ),
            ],
        )

    @workflow.run
    async def run(self, prompt: str) -> str:
        result = await self.agent.invoke_async(prompt)
        return str(result)


Register the Activity functions on the Worker:

strands_plugin/tools/run_worker.py

import asyncio
import os

from temporalio.client import Client
from temporalio.contrib.strands import StrandsPlugin
from temporalio.worker import Worker

from strands_plugin.tools.workflow import (
    ToolsWorkflow,
    environment_activity,
    fetch_weather,
)


async def main() -> None:
    plugin = StrandsPlugin()
    client = await Client.connect(
        os.environ.get("TEMPORAL_ADDRESS", "localhost:7233"),
        plugins=[plugin],
    )

    worker = Worker(
        client,
        task_queue="strands-tools",
        workflows=[ToolsWorkflow],
        activities=[fetch_weather, environment_activity],
    )
    print("Worker started. Ctrl+C to exit.")
    await worker.run()


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

If you are using built-in strands_tools, wrap them in a thin async function decorated with @activity.defn so they run as Temporal Activities.

React to agent lifecycle events​

Strands' hook system lets you subscribe callbacks to events in the agent lifecycle, such as invocation start/end, model call before/after, tool call before/after, and message added. Use hooks to add logging, metrics, or custom logic at each stage.

Pass hooks=[MyHookProvider()] to TemporalAgent. Hook callbacks fire in Workflow context, so deterministic callbacks work without any extra setup.

For callbacks that need I/O (audit logging, metrics, alerting), use activity_as_hook to dispatch the work as a Temporal Activity. The following example shows both patterns in one HookProvider. The _record callback runs in Workflow context (deterministic), while persist_tool_call runs as an Activity (I/O-safe):

strands_plugin/hooks/workflow.py

@activity.defn
async def persist_tool_call(tool_name: str) -> None:
    # In production, write to a database / S3 / your audit pipeline.
    activity.logger.info(f"audit: tool {tool_name} completed")


strands_plugin/hooks/workflow.py

class AuditHook(HookProvider):
    def __init__(self) -> None:
        self.fired: list[str] = []

    def register_hooks(self, registry: HookRegistry, **kwargs: object) -> None:
        registry.add_callback(AfterToolCallEvent, self._record)
        registry.add_callback(
            AfterToolCallEvent,
            activity_as_hook(
                persist_tool_call,
                activity_input=lambda event: event.tool_use["name"],
                start_to_close_timeout=timedelta(seconds=15),
            ),
        )

    def _record(self, event: AfterToolCallEvent) -> None:
        self.fired.append(event.tool_use["name"])


caution

Hook callbacks run in Workflow context, so they must be deterministic. Do not use time.time(), uuid.uuid4(), or I/O inside hook callbacks. Use activity_as_hook for anything that requires I/O.

The activity_input parameter extracts serializable values from the event to pass as the Activity's input. Use a dataclass or Pydantic model for multiple values. This is needed because hook events hold references to Agent, AgentTool instances, and other objects that cannot cross the Activity boundary.

Connect to MCP servers​

If your agent needs access to tools provided by an MCP server, configure the MCP clients on the Worker and reference them by name in the Workflow.

StrandsPlugin(mcp_clients=...) takes a mapping of name to MCPClient factory, mirroring the models pattern. The plugin registers a per-server Activity and connects at Worker startup to enumerate available tools. In the Workflow, TemporalMCPClient(server="name") is a handle that references the server by name and carries per-call Activity options.

Define the Workflow with a TemporalMCPClient:

strands_plugin/mcp/workflow.py

from datetime import timedelta

from temporalio import workflow
from temporalio.contrib.strands import TemporalAgent, TemporalMCPClient


@workflow.defn
class MCPWorkflow:
    def __init__(self) -> None:
        echo = TemporalMCPClient(
            server="echo",
            start_to_close_timeout=timedelta(seconds=30),
        )
        self.agent = TemporalAgent(
            start_to_close_timeout=timedelta(seconds=60),
            tools=[echo],
        )

    @workflow.run
    async def run(self, prompt: str) -> str:
        result = await self.agent.invoke_async(prompt)
        return str(result)


Register the MCP client factory on the Worker:

strands_plugin/mcp/run_worker.py

# ...
from mcp import StdioServerParameters, stdio_client
from strands.tools.mcp.mcp_client import MCPClient
from temporalio.client import Client
from temporalio.contrib.strands import StrandsPlugin
from temporalio.worker import Worker
# ...
def _make_echo_client() -> MCPClient:
    return MCPClient(
        lambda: stdio_client(
            StdioServerParameters(
                command=sys.executable,
                args=[str(ECHO_SERVER)],
            )
        )
    )
# ...
async def main() -> None:
    plugin = StrandsPlugin(mcp_clients={"echo": _make_echo_client})
    client = await Client.connect(
        os.environ.get("TEMPORAL_ADDRESS", "localhost:7233"),
        plugins=[plugin],
    )

    worker = Worker(
        client,
        task_queue="strands-mcp",
        workflows=[MCPWorkflow],
    )
    print("Worker started. Ctrl+C to exit.")
    await worker.run()

Each factory returns a fully configured MCPClient, so you can pass options like tool_filters, prefix, elicitation_callback, or tasks_config to it.

info

The plugin connects to each MCP server once at Worker startup to enumerate tools. The schema is frozen for the Worker's lifetime. Restart Workers to pick up MCP server changes. If a server is unavailable at startup, the Worker fails to start.

Interact with the agent​

Control the shape of agent responses, stream output in real time, and pause the agent for human approval.

Add human approval gates​

Some agent actions, such as deleting resources or sending messages, may require human approval before proceeding. Strands offers two ways to interrupt an agent and wait for a response. Both work with the plugin.

In each case, agent.invoke_async() returns AgentResult(stop_reason="interrupt", interrupts=[...]) instead of raising. Pair this with a Signal handler that supplies responses, then resume by calling agent.invoke_async(responses).

Interrupt from a hook​

A hook on an interruptible event such as BeforeToolCallEvent can pause the agent by calling event.interrupt(name, reason=...). The hook runs in Workflow context, so it must be deterministic.

Define the approval hook:

strands_plugin/human_in_the_loop/workflow.py

class ApprovalHook(HookProvider):
    def register_hooks(self, registry: HookRegistry, **kwargs: object) -> None:
        registry.add_callback(BeforeToolCallEvent, self._gate)

    def _gate(self, event: BeforeToolCallEvent) -> None:
        if event.tool_use["name"] != "delete_file":
            return
        approval = event.interrupt(
            "approval",
            reason=f"approve delete of {event.tool_use['input']['path']}?",
        )
        if approval != "approve":
            event.cancel_tool = "denied"


The Workflow waits for a Signal carrying the approval response, then resumes the agent:

strands_plugin/human_in_the_loop/workflow.py

@workflow.defn
class HumanInTheLoopWorkflow:
    def __init__(self) -> None:
        self.agent = TemporalAgent(
            start_to_close_timeout=timedelta(seconds=60),
            tools=[delete_file],
            hooks=[ApprovalHook()],
        )
        self._approval: str | None = None
        self._pending_reason: str | None = None