Add Strands plugin samples

README.md

@@ -89,6 +89,7 @@ Some examples require extra dependencies. See each sample's directory for specif
 * [pydantic_converter](pydantic_converter) - Data converter for using Pydantic models.
 * [schedules](schedules) - Demonstrates a Workflow Execution that occurs according to a schedule.
 * [sentry](sentry) - Report errors to Sentry.
+* [strands_plugin](strands_plugin) - Run Strands Agents as durable Temporal workflows (model calls, tools, MCP, HITL).
 * [trio_async](trio_async) - Use asyncio Temporal in Trio-based environments.
 * [updatable_timer](updatable_timer) - A timer that can be updated while sleeping.
 * [worker_specific_task_queues](worker_specific_task_queues) - Use unique task queues to ensure activities run on specific workers.

pyproject.toml

@@ -61,6 +61,13 @@ openai-agents = [
 ]
 pydantic-converter = ["pydantic>=2.10.6,<3"]
 sentry = ["sentry-sdk>=2.13.0"]
+strands-agents = [
+    "strands-agents>=1.39.0",
+    "strands-agents-tools>=0.5.2",
+    "mcp>=1.0.0",
+    "boto3>=1.34.92,<2",
+    "temporalio[strands-agents,pydantic]>=1.27.0",
+]
 trio-async = ["trio>=0.28.0,<0.29", "trio-asyncio>=0.15.0,<0.16"]
 cloud-export-to-parquet = [
     "pandas>=2.3.3,<3 ; python_version >= '3.10' and python_version < '4.0'",
@@ -71,14 +78,18 @@ cloud-export-to-parquet = [
 
 [tool.uv]
 constraint-dependencies = [
-    # langsmith 0.7.34 changed its aio_to_thread signature; temporalio.contrib.langsmith
-    # 1.27.2 still patches the older signature, causing workflow task retries to hang CI.
-    "langsmith<0.7.34",
+    "langsmith>=0.7.34",
     # yarl 1.24.0 was published without an sdist and only has cp310 wheels, so it cannot
     # install on the Python 3.14 CI jobs.
     "yarl!=1.24.0",
 ]
 
+# Temporary: the strands extra of temporalio is shipping in an upcoming release.
+# Point at the strands branch of sdk-python until it's published.
+# Remove this section once `temporalio[strands]` is on PyPI.
+[tool.uv.sources]
+temporalio = { git = "https://github.com/temporalio/sdk-python.git", branch = "strands" }
+
 [tool.hatch.metadata]
 allow-direct-references = true
 
@@ -118,6 +129,7 @@ packages = [
     "schedules",
     "sentry",
     "sleep_for_days",
+    "strands_plugin",
     "tests",
     "trio_async",
     "updatable_timer",

strands_plugin/README.md

@@ -0,0 +1,84 @@
+# Strands Agents Samples
+
+These samples demonstrate the [Temporal Strands plugin](https://github.com/temporalio/sdk-python/tree/main/temporalio/contrib/strands), which runs [Strands Agents](https://strandsagents.com/) inside Temporal Workflows. Model invocations, tool calls, and MCP tool calls all execute as Temporal Activities, so you get durable execution, Temporal-managed retries, and timeouts.
+
+## Samples
+
+| Sample | Description |
+|--------|-------------|
+| [hello_world](hello_world) | Minimal `TemporalAgent` invocation. Start here. |
+| [tools](tools) | Three tool patterns side by side: in-workflow `@tool`, custom `@activity.defn` wrapped via `activity_as_tool`, and a `strands_tools` tool wrapped as a Temporal activity. |
+| [human_in_the_loop](human_in_the_loop) | Pause a tool call on `BeforeToolCallEvent.interrupt()`, resume via Temporal signal. The canonical Strands HITL pattern. |
+| [tool_interrupt](tool_interrupt) | Raise `InterruptException` from a Temporal activity to surface a HITL prompt across the activity boundary. Plugin-specific feature. |
+| [hooks](hooks) | `HookProvider` with both an in-workflow callback and an `activity_as_hook` callback for I/O. |
+| [mcp](mcp) | Connect to an MCP server (`FastMCP` echo) via `TemporalMCPClient`. |
+| [structured_output](structured_output) | Pydantic-typed agent output via `structured_output_model`. |
+| [streaming](streaming) | Forward model chunks to an external subscriber via `streaming_topic` + `WorkflowStream`. |
+| [continue_as_new](continue_as_new) | Chat-style workflow that hands off `agent.messages` when history grows large. |
+
+## Prerequisites
+
+1. Install dependencies:
+
+   ```bash
+   uv sync --group strands
+   ```
+
+   > The `strands` extra of `temporalio` is shipping in an upcoming release. Until then, install the SDK from the strands branch:
+   >
+   > ```bash
+   > uv pip install -e ../sdk-python --extra strands-agents --extra pydantic
+   > ```
+
+2. Configure AWS credentials. The samples use the plugin's default `BedrockModel()`, which picks up the standard AWS SDK credential chain. Make sure the credentials grant access to a Bedrock model in your selected region (e.g., `us-west-2`).
+
+   ```bash
+   export AWS_REGION=us-west-2
+   # plus AWS_ACCESS_KEY_ID / AWS_SECRET_ACCESS_KEY or an SSO profile
+   ```
+
+   You can pick a specific model by passing it to `BedrockModel(model_id="...")` in each sample's worker.
+
+3. Start a [Temporal dev server](https://docs.temporal.io/cli#start-dev-server):
+
+   ```bash
+   temporal server start-dev
+   ```
+
+## Running a Sample
+
+Each sample has two scripts. Start the Worker first, then the Workflow starter in a separate terminal:
+
+```bash
+# Terminal 1: start the Worker
+uv run strands_plugin/<sample>/run_worker.py
+
+# Terminal 2: start the Workflow
+uv run strands_plugin/<sample>/run_workflow.py
+```
+
+For example, to run the tools sample:
+
+```bash
+# Terminal 1
+uv run strands_plugin/tools/run_worker.py
+
+# Terminal 2
+uv run strands_plugin/tools/run_workflow.py
+```
+
+## Key Features Demonstrated
+
+- **Durable model invocation** — every model call runs in an `invoke_model` activity with configurable timeouts and retries.
+- **Three ways to define tools** — pure Strands `@tool`, custom Temporal activities, and ecosystem `strands_tools` wrapped as activities.
+- **Human-in-the-loop** — both hook-based (`BeforeToolCallEvent.interrupt()`) and tool-body (`raise InterruptException`) styles.
+- **Hook system** — deterministic in-workflow callbacks plus I/O callbacks dispatched via `activity_as_hook`.
+- **MCP integration** — connect to MCP servers at worker startup; tool calls dispatched through per-server activities.
+- **Structured output** — Pydantic-typed agent results via the plugin's `pydantic_data_converter`.
+- **Streaming** — forward model chunks live to external subscribers.
+- **Long-lived chats** — hand off `agent.messages` via `continue-as-new` to stay under Temporal's history limit.
+
+## Related
+
+- [Temporal Strands plugin docs](https://github.com/temporalio/sdk-python/tree/main/temporalio/contrib/strands)
+- [Strands Agents](https://strandsagents.com/)

strands_plugin/continue_as_new/README.md

@@ -0,0 +1,30 @@
+# Continue-as-new
+
+A chat-style workflow accumulates history with every turn and will eventually hit Temporal's per-workflow history limit. `workflow.info().is_continue_as_new_suggested()` flips `True` once the server decides history has grown large enough; this sample checks it after each turn and hands off to a fresh run with `agent.messages` as input.
+
+## What This Sample Demonstrates
+
+- Driving a multi-turn chat with **updates**, so each caller gets the assistant's reply back from the same call
+- Seeding a new `TemporalAgent` with prior `agent.messages`
+- Using `workflow.info().is_continue_as_new_suggested()` + `workflow.continue_as_new(...)` to keep the workflow alive indefinitely
+- Draining in-flight update handlers with `workflow.all_handlers_finished` before continue-as-new
+
+## Running the Sample
+
+```bash
+# Terminal 1
+uv run strands_plugin/continue_as_new/run_worker.py
+
+# Terminal 2
+uv run strands_plugin/continue_as_new/run_workflow.py
+```
+
+The starter calls the `turn` update for each user message and prints the assistant's reply, then signals `end_chat`. In a real chatbot, a UI would drive the updates and the workflow would run indefinitely, continuing-as-new whenever history gets large.
+
+## Files
+
+| File | Description |
+|------|-------------|
+| `workflow.py` | `ChatInput`, `ChatWorkflow` with `turn` update, `end_chat` signal, and `messages` query |
+| `run_worker.py` | Registers `StrandsPlugin`, starts the worker |
+| `run_workflow.py` | Starts the chat, sends a few turns, ends it |

strands_plugin/continue_as_new/run_worker.py

@@ -0,0 +1,30 @@
+"""Worker for the chat continue-as-new sample."""
+
+import asyncio
+import os
+
+from temporalio.client import Client
+from temporalio.contrib.strands import StrandsPlugin
+from temporalio.worker import Worker
+
+from strands_plugin.continue_as_new.workflow import ChatWorkflow
+
+
+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-chat",
+        workflows=[ChatWorkflow],
+    )
+    print("Worker started. Ctrl+C to exit.")
+    await worker.run()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

strands_plugin/continue_as_new/run_workflow.py

@@ -0,0 +1,38 @@
+"""Start the chat workflow, send a few turns, then end it."""
+
+import asyncio
+import os
+
+from temporalio.client import Client
+from temporalio.contrib.strands import StrandsPlugin
+
+from strands_plugin.continue_as_new.workflow import ChatInput, ChatWorkflow
+
+
+async def main() -> None:
+    client = await Client.connect(
+        os.environ.get("TEMPORAL_ADDRESS", "localhost:7233"),
+        plugins=[StrandsPlugin()],
+    )
+
+    handle = await client.start_workflow(
+        ChatWorkflow.run,
+        ChatInput(),
+        id="strands-chat",
+        task_queue="strands-chat",
+    )
+
+    for prompt in [
+        "Hi! What is durable execution?",
+        "Give me a one-sentence summary.",
+    ]:
+        reply = await handle.execute_update(ChatWorkflow.turn, prompt)
+        print(f"user: {prompt}")
+        print(f"assistant: {reply}\n")
+
+    await handle.signal(ChatWorkflow.end_chat)
+    await handle.result()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

strands_plugin/continue_as_new/workflow.py

@@ -0,0 +1,63 @@
+"""Chat-style workflow that continues-as-new before history grows too large.
+
+Each user turn arrives as a Temporal **update**, so the caller gets the
+assistant's reply back from the same call. Once Temporal suggests
+continue-as-new, the workflow drains any in-flight update handlers and hands
+``agent.messages`` off to a fresh run.
+"""
+
+import asyncio
+from dataclasses import dataclass, field
+from datetime import timedelta
+
+from strands.types.content import Messages
+from temporalio import workflow
+from temporalio.contrib.strands import TemporalAgent
+
+
+@dataclass
+class ChatInput:
+    messages: Messages = field(default_factory=list)
+
+
+@workflow.defn
+class ChatWorkflow:
+    def __init__(self) -> None:
+        self._done = False
+        self._lock = asyncio.Lock()
+        self._agent: TemporalAgent | None = None
+
+    @workflow.update
+    async def turn(self, prompt: str) -> str:
+        # Updates can arrive before ``run`` has constructed the agent.
+        await workflow.wait_condition(lambda: self._agent is not None)
+        # Serialize turns so concurrent updates can't interleave on ``agent.messages``.
+        async with self._lock:
+            assert self._agent is not None
+            result = await self._agent.invoke_async(prompt)
+            return str(result).strip()
+
+    @workflow.signal
+    def end_chat(self) -> None:
+        self._done = True
+
+    @workflow.query
+    def messages(self) -> Messages:
+        return list(self._agent.messages) if self._agent else []
+
+    @workflow.run
+    async def run(self, input: ChatInput) -> None:
+        self._agent = TemporalAgent(
+            start_to_close_timeout=timedelta(seconds=60),
+            messages=list(input.messages),
+        )
+
+        await workflow.wait_condition(
+            lambda: self._done or workflow.info().is_continue_as_new_suggested()
+        )
+
+        # Let any in-flight ``turn`` updates finish before we exit or hand off.
+        await workflow.wait_condition(workflow.all_handlers_finished)
+
+        if not self._done:
+            workflow.continue_as_new(ChatInput(messages=self._agent.messages))

strands_plugin/hello_world/README.md

@@ -0,0 +1,29 @@
+# Hello World
+
+The simplest Strands + Temporal sample: one `TemporalAgent` invoked once. Every model call runs as an `invoke_model` Temporal activity, so it gets durable retries, timeouts, and crash recovery for free.
+
+## What This Sample Demonstrates
+
+- Wiring `StrandsPlugin` onto the client and worker
+- Constructing a `TemporalAgent` with no explicit model (defaults to `BedrockModel()`)
+- Invoking the agent from a `@workflow.defn`
+
+## Running the Sample
+
+Prerequisites: `uv sync --group strands`, AWS credentials with Bedrock access, and a running Temporal dev server (`temporal server start-dev`).
+
+```bash
+# Terminal 1
+uv run strands_plugin/hello_world/run_worker.py
+
+# Terminal 2
+uv run strands_plugin/hello_world/run_workflow.py
+```
+
+## Files
+
+| File | Description |
+|------|-------------|
+| `workflow.py` | `HelloWorldWorkflow` with a single `TemporalAgent` |
+| `run_worker.py` | Registers `StrandsPlugin`, starts the worker |
+| `run_workflow.py` | Executes the workflow and prints the result |

strands_plugin/hello_world/run_worker.py

@@ -0,0 +1,30 @@
+"""Worker for the hello world sample."""
+
+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())

strands_plugin/hello_world/run_workflow.py

@@ -0,0 +1,25 @@
+"""Start the hello world workflow."""
+
+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())