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: