Add Claude Agent SDK samples

README.md

@@ -63,6 +63,7 @@ Some examples require extra dependencies. See each sample's directory for specif
 * [batch_sliding_window](batch_sliding_window) - Batch processing with a sliding window of child workflows.
 * [bedrock](bedrock) - Orchestrate a chatbot with Amazon Bedrock.
 * [bedrock_agentcore/strands_agent](bedrock_agentcore/strands_agent) - Run a AWS Strands Agent with Temporal Plugin on AgentCore Worker.
+* [claude_agent_sdk_plugin](claude_agent_sdk_plugin) - Run Claude Code agents as durable workflows with the Claude Agent SDK.
 * [cloud_export_to_parquet](cloud_export_to_parquet) - Set up schedule workflow to process exported files on an hourly basis
 * [context_propagation](context_propagation) - Context propagation through workflows/activities via interceptor.
 * [custom_converter](custom_converter) - Use a custom payload converter to handle custom types.

claude_agent_sdk_plugin/README.md

@@ -0,0 +1,58 @@
+# Claude Agent SDK Samples
+
+⚠️ **Pre-release**: this integration and its interfaces may change before General Availability.
+
+These samples demonstrate the [Temporal Claude Agent SDK integration](https://github.com/temporalio/ai-integrations/tree/main/python/claude_agent_sdk), which runs [Claude Code](https://code.claude.com/docs/en/agent-sdk/overview) agents as Temporal Workflows. You write ordinary Claude Agent SDK code in a Workflow, and Claude Code runs unmodified. If a Worker fails, the agent picks up where it left off, and approvals, hooks, and custom tools run as durable Workflow code.
+
+## Samples
+
+| Category | Sample | What it shows |
+| --- | --- | --- |
+| Agent Chat | [agent_chat](agent_chat) | A multi-turn chat driven by Updates, with Continue-As-New that resumes Claude's session in the next run. Start here. |
+| Tools | [tools](tools) | One custom tool computed in the Workflow and one that runs an Activity, both via `@tool` and `create_sdk_mcp_server`. |
+| Streaming | [streaming](streaming) | Token-by-token output through a Workflow Stream, with or without a subscriber. |
+| Human in the Loop | [human_in_the_loop](human_in_the_loop) | `can_use_tool` waits durably for a person's Update: approve, deny, or cancel. |
+| Structured Output | [structured_output](structured_output) | A JSON-schema result validated as a Pydantic model at the Workflow boundary. |
+| Fork Session | [fork_session](fork_session) | One conversation forked with `fork_session` into parallel branches, each with a session ID from Claude Code. |
+| Multi-Agent | [multi_agent](multi_agent) | Claude Code's native subagents: a coordinator delegating to a researcher and a writer. |
+| Plugin with Custom Transport | [plugin_with_custom_transport](plugin_with_custom_transport) | A Claude Code plugin with a skill and an MCP server, copied into a Docker Sandbox (`sbx`) by a custom connector that runs Claude Code there. |
+| Observability | [open_telemetry](open_telemetry) | One OpenTelemetry trace across the caller, the Workflow, its Activities, and Claude Code's model and tool spans. |
+
+Most samples run Claude Code as a subprocess of the Worker. `plugin_with_custom_transport` runs it in a Docker Sandbox instead.
+
+## Prerequisites
+
+Install the [Temporal CLI](https://docs.temporal.io/cli), then the dependencies:
+
+```bash
+uv sync --group claude-agent-sdk
+```
+
+The `claude-agent-sdk` package bundles Claude Code, so you don't need to install it separately. Each sample's README has the commands to run it.
+
+## Stopping a Worker
+
+To see a Worker fail mid-turn and the restarted Worker pick up where it left off, stop the Worker with SIGTERM, as process managers do. For example, while the human-in-the-loop sample waits for an approval:
+
+```bash
+pkill -f claude_agent_sdk_plugin/human_in_the_loop/run_worker.py
+```
+
+The Worker shuts down and stops Claude Code, leaving the turn for the restarted Worker to resume, so a pending tool approval is asked again.
+
+Stopping a Worker mid-turn these ways isn't recommended:
+
+- `kill -9` on the Worker process alone. Claude Code outlives it, sees its stdin close, and records a pending tool approval as failed.
+- Ctrl+C in the Worker's terminal. Claude Code gets the Ctrl+C too, and records a pending tool approval as rejected.
+
+Either way, the resumed session continues as if the call failed or was denied, instead of asking again.
+
+## Tests
+
+The tests run every sample against the scripted API and a local dev server, with no credentials or Docker:
+
+```bash
+uv run --group claude-agent-sdk pytest tests/claude_agent_sdk_plugin
+```
+
+They're skipped on Windows, where the default transport's shell scripts don't run.

claude_agent_sdk_plugin/agent_chat/README.md

@@ -0,0 +1,70 @@
+# Agent Chat
+
+A multi-turn chat with Claude that lives in one Workflow. Each message is a `send_message` Update that returns Claude's reply, and the `end_conversation` Signal ends the chat. Start here.
+
+## What this shows
+
+- **A chat is just a Workflow.** You write ordinary `ClaudeSDKClient` code, and each message becomes a durable Update with Claude's reply as the result.
+- **Chats can run forever.** The Workflow continues as new every few turns, so its history stays small no matter how long the conversation gets.
+- **Claude doesn't forget.** Only the session ID crosses to the next run. Claude Code reloads the full conversation from its own session file, so nothing is lost and nothing large is stored in Temporal.
+- **A Worker restart loses no message.** If the Worker stops mid-reply, the restarted Worker resumes the session. Claude Code finishes the cut-off reply itself, so the Workflow sends the message again only if Claude Code never got it. Each message gets exactly one reply.
+
+## Run it
+
+All commands run from the repository root. Start a Temporal dev server in a separate terminal:
+
+```bash
+temporal server start-dev
+```
+
+Start the Worker in another terminal.
+
+```bash
+uv run --group claude-agent-sdk claude_agent_sdk_plugin/agent_chat/run_worker.py
+```
+
+The Worker runs offline: everything runs for real except the model, which replies with how many replies came before. To use the real model, put `ANTHROPIC_API_KEY=...` in front of the command.
+
+### Chat
+
+Start the chat:
+
+```bash
+temporal workflow start \
+  --type ChatWorkflow \
+  --task-queue claude-agent-sdk-agent-chat \
+  --workflow-id claude-agent-sdk-agent-chat \
+  --input '{}'
+```
+
+Send messages, one at a time:
+
+```bash
+temporal workflow update execute \
+  --workflow-id claude-agent-sdk-agent-chat \
+  --name send_message \
+  --input '"Hi, my name is Ada."'
+```
+
+Each command prints Claude's reply. Send four messages to cross into a new run, then check that Claude remembers.
+
+Each run takes three messages. After the third reply the run is full and hands off to the next one. A message sent in that moment is rejected with `The chat is continuing as new. Send it again.`, so if your fourth message fails, send it again.
+
+Once the fourth reply arrives, check:
+
+- **Live:** ask "What is my name?" and Claude still knows it.
+- **Offline:** the fourth reply is `"Offline reply 4 in this conversation."`. The scripted model only counts the earlier replies in the request Claude Code sends, so the 4 shows Claude Code reloaded them in the new run.
+
+See the runs, one `ContinuedAsNew` and one `Running`:
+
+```bash
+temporal workflow list --query 'WorkflowId="claude-agent-sdk-agent-chat"'
+```
+
+End the chat:
+
+```bash
+temporal workflow signal \
+  --workflow-id claude-agent-sdk-agent-chat \
+  --name end_conversation
+```

claude_agent_sdk_plugin/agent_chat/run_worker.py

@@ -0,0 +1,42 @@
+import asyncio
+import os
+import signal
+from typing import Any
+
+from temporalio.claude_agent_sdk import ClaudeAgentSDKPlugin
+from temporalio.client import Client
+from temporalio.worker import Worker
+
+from claude_agent_sdk_plugin.agent_chat.workflow import ChatWorkflow
+from claude_agent_sdk_plugin.offline import Answer, offline_api_without_key
+
+
+def offline_reply(request: dict[str, Any]) -> str:
+    earlier = sum(message["role"] == "assistant" for message in request["messages"])
+    return f"Offline reply {earlier + 1} in this conversation."
+
+
+OFFLINE_ANSWERS: dict[str, Answer] = {"": offline_reply}
+
+
+async def main() -> None:
+    client = await Client.connect(os.environ.get("TEMPORAL_ADDRESS", "localhost:7233"))
+
+    # Shut the Worker down on SIGTERM, so the plugin stops Claude Code and a
+    # restarted Worker can resume its session.
+    stop = asyncio.Event()
+    asyncio.get_running_loop().add_signal_handler(signal.SIGTERM, stop.set)
+
+    with offline_api_without_key(OFFLINE_ANSWERS):
+        async with Worker(
+            client,
+            task_queue="claude-agent-sdk-agent-chat",
+            workflows=[ChatWorkflow],
+            plugins=[ClaudeAgentSDKPlugin()],
+        ):
+            print(f"Worker started. To stop it gracefully: kill {os.getpid()}")
+            await stop.wait()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

claude_agent_sdk_plugin/agent_chat/workflow.py

@@ -0,0 +1,117 @@
+import asyncio
+from dataclasses import dataclass
+
+from claude_agent_sdk import (
+    ClaudeAgentOptions,
+    ClaudeSDKClient,
+    CLIConnectionError,
+    ResultMessage,
+    SystemMessage,
+)
+from temporalio import workflow
+from temporalio.claude_agent_sdk import TemporalTransport
+from temporalio.exceptions import ApplicationError
+
+
+@dataclass
+class ChatInput:
+    session_id: str | None = None
+    turns_per_run: int = 3
+
+
+@workflow.defn
+class ChatWorkflow:
+    @workflow.init
+    def __init__(self, input: ChatInput) -> None:
+        self.session_id = input.session_id
+        self.turns_per_run = input.turns_per_run
+        self.turns = 0
+        self.ended = False
+        self.client = self.new_client()
+        self.connected = False
+        self.lock = asyncio.Lock()
+
+    @workflow.run
+    async def run(self, input: ChatInput) -> None:
+        await workflow.wait_condition(self.run_is_done)
+
+        await self.client.disconnect()
+
+        if self.ended:
+            return
+
+        workflow.continue_as_new(ChatInput(self.session_id, self.turns_per_run))
+
+    def new_client(self) -> ClaudeSDKClient:
+        """A client that starts Claude Code with a new session, or resumes the session."""
+        options = ClaudeAgentOptions(
+            setting_sources=[],
+            env={"CLAUDE_CODE_RESUME_INTERRUPTED_TURN": "1"},
+            resume=self.session_id,  # None starts a new session.
+        )
+
+        # The default SubprocessConnector keeps session files on the Worker host,
+        # so only that host can resume. With Workers on several hosts, pass
+        # activity_options={"connect": {"task_queue": ...}} with a task queue
+        # only that host polls.
+        return ClaudeSDKClient(options=options, transport=TemporalTransport(options))
+
+    @workflow.update
+    async def send_message(self, message: str) -> str:
+        """Answer the message in the session, resuming it if the connection is lost."""
+        self.turns += 1
+
+        # One turn at a time, so each message gets its own reply.
+        async with self.lock:
+            sent = False
+
+            while True:
+                try:
+                    # Claude Code starts on the first message, and again after a resume.
+                    if not self.connected:
+                        await self.client.connect()
+                        self.connected = True
+
+                    # Keep sending the message until Claude Code has received it in a
+                    # session it reported. After that, a resume finishes the turn.
+                    if not sent or self.session_id is None:
+                        await self.client.query(message)
+                        sent = True
+
+                    async for response in self.client.receive_response():
+                        if (
+                            isinstance(response, SystemMessage)
+                            and response.subtype == "init"
+                        ):
+                            self.session_id = response.data["session_id"]
+                        elif isinstance(response, ResultMessage):
+                            return response.result or "\n".join(response.errors or [])
+
+                    raise RuntimeError("Claude Code ended the turn without a result")
+
+                except CLIConnectionError:
+                    await self.client.disconnect()
+                    self.connected = False
+                    self.client = self.new_client()
+
+    @send_message.validator
+    def validate_send_message(self, message: str) -> None:
+        if self.ended:
+            raise ApplicationError("The conversation has ended")
+
+        if self.run_is_full():
+            raise ApplicationError("The chat is continuing as new. Send it again.")
+
+    @workflow.signal
+    def end_conversation(self) -> None:
+        self.ended = True
+
+    def run_is_done(self) -> bool:
+        """The chat ended or this run is full, and every message has its reply."""
+        return (self.ended or self.run_is_full()) and workflow.all_handlers_finished()
+
+    def run_is_full(self) -> bool:
+        return (
+            self.turns >= self.turns_per_run
+            or workflow.info().is_continue_as_new_suggested()
+        )

claude_agent_sdk_plugin/fork_session/README.md

@@ -0,0 +1,49 @@
+# Fork Session
+
+Answer a prompt once, then try several follow-ups at the same time, each in its own fork of that conversation. Every fork starts from the same history, and the original session stays as it was.
+
+## What this shows
+
+- **Branching a conversation.** `fork_session=True` with `resume` starts a new session from a copy of an existing one, as in the Claude Agent SDK.
+- **Session IDs from Claude Code.** The Workflow reads each session's ID from Claude Code's `init` message, including each fork's. If the connection is lost, it resumes that session, or starts it again if Claude Code hadn't yet started it.
+- **Forks in parallel.** Each fork is its own Claude Code. The Workflow runs them together with `asyncio.gather`.
+
+## Run it
+
+All commands run from the repository root. Start a Temporal dev server in a separate terminal:
+
+```bash
+temporal server start-dev
+```
+
+Start the Worker in another terminal:
+
+```bash
+uv run --group claude-agent-sdk claude_agent_sdk_plugin/fork_session/run_worker.py
+```
+
+The Worker runs offline: everything runs for real except the model, which answers with fixed text. To use the real model, put `ANTHROPIC_API_KEY=...` in front of the command.
+
+Run the Workflow from a third terminal, with the prompt and a list of follow-ups:
+
+```bash
+temporal workflow execute \
+  --type ForkSessionWorkflow \
+  --task-queue claude-agent-sdk-fork-session \
+  --workflow-id fork-session \
+  --input '"Plan a weekend in Lisbon."' \
+  --input '["Redo it on a tight budget.", "Redo it with no budget limit."]'
+```
+
+## Expected behavior
+
+The result has one branch per follow-up, each with its own session ID. Offline, it is:
+
+```json
+[
+  {"answer": "Walk Alfama, catch a free fado night, take the tram to Belém.", "prompt": "Redo it on a tight budget.", "session_id": "8889ec93-..."},
+  {"answer": "A private fado dinner, then a yacht along the Tagus to Belém.", "prompt": "Redo it with no budget limit.", "session_id": "3a14ddda-..."}
+]
+```
+
+Each fork's model request has the weekend plan from the base conversation, and neither has the other fork's follow-up. To continue a branch later, start a client with `resume` set to its session ID.

claude_agent_sdk_plugin/fork_session/run_worker.py

@@ -0,0 +1,39 @@
+import asyncio
+import os
+import signal
+
+from temporalio.claude_agent_sdk import ClaudeAgentSDKPlugin
+from temporalio.client import Client
+from temporalio.worker import Worker
+
+from claude_agent_sdk_plugin.fork_session.workflow import ForkSessionWorkflow
+from claude_agent_sdk_plugin.offline import offline_api_without_key
+
+OFFLINE_ANSWERS = {
+    "Plan a weekend in Lisbon": "Saturday: Alfama and a fado show. Sunday: Belém.",
+    "on a tight budget": "Walk Alfama, catch a free fado night, take the tram to Belém.",
+    "with no budget limit": "A private fado dinner, then a yacht along the Tagus to Belém.",
+}
+
+
+async def main() -> None:
+    client = await Client.connect(os.environ.get("TEMPORAL_ADDRESS", "localhost:7233"))
+
+    # Shut the Worker down on SIGTERM, so the plugin stops Claude Code and a
+    # restarted Worker can resume its session.
+    stop = asyncio.Event()
+    asyncio.get_running_loop().add_signal_handler(signal.SIGTERM, stop.set)
+
+    with offline_api_without_key(OFFLINE_ANSWERS):
+        async with Worker(
+            client,
+            task_queue="claude-agent-sdk-fork-session",
+            workflows=[ForkSessionWorkflow],
+            plugins=[ClaudeAgentSDKPlugin()],
+        ):
+            print(f"Worker started. To stop it gracefully: kill {os.getpid()}")
+            await stop.wait()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())