strands-agent-builder

name

strands-agent-builder

description

Build AWS Strands agents and make them durable with Temporal. Use when creating a Strands Agent, defining @tool functions, wiring AnthropicModel/Claude, or wrapping a Strands agent loop in a Temporal workflow via temporalio.contrib.strands (StrandsPlugin, TemporalAgent, activity_as_tool). Triggers on "Strands agent", "strands @tool", "durable agent", "temporalio.contrib.strands", "TemporalAgent", "make my agent durable".

Strands Agent Builder (+ Temporal durability)

Build a Strands agent once, then make it durable by wrapping — not rewriting. The agent logic is identical; durability comes from running the agent loop inside a Temporal workflow where the model call and each tool call become Temporal activities.

API verified against strands-agents 1.45.0 and temporalio 1.29.0 (mid-2026). Do not trust memorized signatures — temporalio.contrib.strands does not expose top-level TemporalModel or TemporalActivityTool. When unsure, check the strands-docs and temporal-docs MCP servers.

1. Plain Strands agent (no durability)

import os
from strands import Agent, tool
from strands.models.anthropic import AnthropicModel  # needs strands-agents[anthropic]

@tool
def check_service_health(url: str) -> dict:
    """Check a service's health endpoint and return status + latency."""
    import httpx
    r = httpx.get(f"{url}/health", timeout=10)
    return {"status_code": r.status_code, "body": r.json()}

model = AnthropicModel(
    client_args={"api_key": os.environ["ANTHROPIC_API_KEY"]},
    model_id=os.environ.get("CLAUDE_MODEL_ID", "claude-opus-4-8"),
    max_tokens=1024,
)
agent = Agent(model=model, tools=[check_service_health], system_prompt="You are an SRE agent.")
print(agent("Is the service at http://localhost:8080 healthy?"))

@tool turns a typed, docstringed function into a tool schema automatically (first docstring paragraph = description; type hints = parameter schema). Agent(...) accepts a Model or a model id.

2. Make it durable with Temporal

Three moves. The pure tool logic stays the same — keep it in plain functions and add thin wrappers.

a) Tools become @activity.defn functions (non-deterministic I/O must be activities):

from temporalio import activity

@activity.defn
async def check_service_health_activity(url: str) -> dict:
    import httpx
    async with httpx.AsyncClient(timeout=10) as c:
        r = await c.get(f"{url}/health")
    return {"status_code": r.status_code, "body": r.json()}

b) The agent runs inside a workflow, using TemporalAgent (model call → activity) and activity_as_tool (each tool call → activity):

from datetime import timedelta
from temporalio import workflow
from temporalio.contrib.strands import TemporalAgent
from temporalio.contrib.strands.workflow import activity_as_tool

@workflow.defn
class SREAgentWorkflow:
    @workflow.run
    async def run(self, prompt: str) -> str:
        agent = TemporalAgent(
            model="claude",  # NAME of a factory registered in StrandsPlugin(models={...})
            tools=[activity_as_tool(check_service_health_activity,
                                    start_to_close_timeout=timedelta(seconds=30))],
            system_prompt="You are an SRE agent.",
        )
        result = await agent.invoke_async(prompt)
        return str(result)

c) The worker registers the model factory + activities via StrandsPlugin:

import os, asyncio
from temporalio.client import Client
from temporalio.worker import Worker
from temporalio.contrib.strands import StrandsPlugin
from strands.models.anthropic import AnthropicModel

def make_model() -> AnthropicModel:
    return AnthropicModel(
        client_args={"api_key": os.environ["ANTHROPIC_API_KEY"]},
        model_id=os.environ.get("CLAUDE_MODEL_ID", "claude-opus-4-8"),
        max_tokens=1024,
    )

async def main():
    client = await Client.connect("localhost:7233", plugins=[StrandsPlugin(models={"claude": make_model})])
    worker = Worker(client, task_queue="sre-agent",
                    workflows=[SREAgentWorkflow],
                    activities=[check_service_health_activity])
    await worker.run()

asyncio.run(main())

Key rules

• Never call an LLM or do network I/O directly in workflow code — only inside activities. Workflows must be deterministic (they replay from history).
• Register a model factory by name in StrandsPlugin(models={"name": factory}); reference that name in TemporalAgent(model="name").
• activity_as_tool(fn, *, start_to_close_timeout=..., retry_policy=...) forwards its kwargs to workflow.execute_activity — set timeouts/retries per tool.
• Human-in-the-loop: gate a sensitive tool by waiting on a @workflow.signal with await workflow.wait_condition(lambda: self.approved, timeout=...) before invoking it.
• MCP tools: register transports worker-side with StrandsPlugin(mcp_clients={...}) and reference them in-workflow via TemporalMCPClient("server").

Serverless worker on AWS Lambda

Same code; swap the worker entrypoint to temporalio.contrib.aws.lambda_worker.run_worker(version, configure). Worker Versioning is mandatory. See the agentcore-harness skill and the repo's durable/worker_lambda.py.