feat: add Claude Agent SDK example with Temporal

Add a new cookbook example demonstrating how to run the Claude Agent SDK
(claude-code-sdk) inside a Temporal workflow for durable agent execution.

Key patterns demonstrated:
- Background heartbeats via asyncio.create_task() (independent of event stream)
- Staleness guard to detect hung agents (15-min idle threshold)
- Response deduplication (AssistantMessage vs StreamEvent)
- Errors modeled as completions (not exceptions)

Includes workflow tests (mocked activities) and activity tests (mocked SDK).

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: 2 people

agents/claude_agent_sdk_python/README.md

@@ -0,0 +1,209 @@
+"""
+Agent Execution Activity — Claude Agent SDK with Temporal
+
+This activity wraps the Claude Agent SDK (claude-agent-sdk) to provide durable,
+observable agent execution inside a Temporal workflow.
+
+Key patterns demonstrated:
+1. Background heartbeats — keeps Temporal informed during long-running tool calls
+2. Staleness guard — stops heartbeating when the agent appears hung
+3. Response deduplication — only accumulates text from AssistantMessage events
+"""
+
+import asyncio
+import os
+import time
+from datetime import datetime, timezone
+
+from temporalio import activity
+
+from models import AgentInput, AgentOutput
+
+
+# How often to send a heartbeat to Temporal (seconds).
+HEARTBEAT_INTERVAL = 60
+
+# If no SDK events arrive for this long, stop heartbeating and let Temporal
+# kill the activity.  This prevents a truly hung agent from blocking the
+# full start_to_close_timeout (30 min).
+MAX_IDLE_SECONDS = 15 * 60  # 15 minutes
+
+
+@activity.defn
+async def execute_agent_activity(input_data: AgentInput) -> AgentOutput:
+    """
+    Execute a Claude agent via the Claude Agent SDK and collect results.
+
+    The Claude Agent SDK manages the agentic loop internally — tool selection,
+    execution, and multi-turn conversation are all handled by the SDK.  This
+    activity simply streams events from the SDK and collects the final response.
+
+    Heartbeat pattern:
+        A background asyncio task sends heartbeats every 60 seconds, independent
+        of the SDK event stream.  This is critical because the SDK may execute
+        long-running tools (e.g. git clone, large file reads) that emit no events
+        for extended periods.  Without background heartbeats, Temporal would kill
+        the activity for missing its heartbeat_timeout.
+
+    Staleness guard:
+        If no SDK events arrive for MAX_IDLE_SECONDS (15 min), the heartbeat
+        loop exits.  Temporal's heartbeat_timeout (10 min) then fires ~25 min
+        after the last event, killing a truly hung agent instead of letting it
+        block the full 30-minute start_to_close_timeout.
+
+    Response deduplication:
+        The Claude Agent SDK emits both StreamEvent (incremental text chunks)
+        and AssistantMessage (complete text blocks).  Both contain the same text,
+        so we only accumulate from AssistantMessage to avoid duplication.
+    """
+    # Lazy import — avoids loading the SDK at module level, which keeps the
+    # Temporal worker startup fast and avoids sandbox issues.
+    from claude_agent_sdk import query
+    from claude_agent_sdk.types import ClaudeAgentOptions, AssistantMessage, ResultMessage
+
+    activity.logger.info(
+        f"Starting agent execution: prompt_length={len(input_data.prompt)}, "
+        f"model={input_data.model}"
+    )
+
+    start_time = datetime.now(timezone.utc)
+
+    try:
+        # Build SDK options
+        #
+        # NOTE: The SDK merges os.environ with options.env ({**os.environ, **env}).
+        # If the worker runs inside Claude Code, CLAUDECODE will be set, and the
+        # bundled CLI binary refuses to launch ("cannot nest Claude Code sessions").
+        # We override it to empty string so the subprocess doesn't see it.
+        options = ClaudeAgentOptions(
+            model=input_data.model,
+            max_turns=input_data.max_turns,
+            permission_mode=input_data.permission_mode,
+            env={"CLAUDECODE": ""},
+        )
+        if input_data.system_prompt:
+            options.system_prompt = input_data.system_prompt
+
+        # Heartbeat state shared between the event loop and the background task.
+        heartbeat_state = {
+            "event_count": 0,
+            "last_event_time": time.time(),
+            "done": False,
+        }
+
+        async def _heartbeat_loop():
+            """
+            Background task that sends Temporal heartbeats at a fixed interval.
+
+            This runs independently of the SDK event stream so that heartbeats
+            continue even when the SDK is executing a long-running tool that
+            produces no events.
+            """
+            while not heartbeat_state["done"]:
+                await asyncio.sleep(HEARTBEAT_INTERVAL)
+                if heartbeat_state["done"]:
+                    break
+
+                idle_seconds = time.time() - heartbeat_state["last_event_time"]
+
+                # Staleness guard: if no events for too long, the agent may be
+                # stuck.  Stop heartbeating and let Temporal's heartbeat_timeout
+                # kill the activity.
+                if idle_seconds > MAX_IDLE_SECONDS:
+                    activity.logger.warning(
+                        f"No events for {idle_seconds:.0f}s — stopping heartbeat "
+                        f"(agent may be stuck)"
+                    )
+                    break
+
+                activity.heartbeat(
+                    f"events={heartbeat_state['event_count']}, "
+                    f"idle={idle_seconds:.0f}s"
+                )
+
+        heartbeat_task = asyncio.create_task(_heartbeat_loop())
+
+        # Collect response and events
+        response_text = ""
+        total_tokens = 0
+        event_count = 0
+
+        try:
+            async for event in query(
+                prompt=input_data.prompt,
+                options=options,
+            ):
+                event_count += 1
+                heartbeat_state["event_count"] = event_count
+                heartbeat_state["last_event_time"] = time.time()
+
+                # Response deduplication: The SDK emits both StreamEvent
+                # (incremental chunks) and AssistantMessage (complete blocks).
+                # Both contain the same text, so we ONLY accumulate from
+                # AssistantMessage to avoid duplicating the response.
+                if isinstance(event, AssistantMessage):
+                    # AssistantMessage.content is a list of content blocks
+                    for block in event.content:
+                        if hasattr(block, "text"):
+                            response_text += block.text
+
+                # Capture token usage from the final result event
+                if isinstance(event, ResultMessage):
+                    total_tokens = getattr(event, "total_tokens", 0) or 0
+
+        finally:
+            # Always clean up the heartbeat task
+            heartbeat_state["done"] = True
+            heartbeat_task.cancel()
+            try:
+                await heartbeat_task
+            except asyncio.CancelledError:
+                pass
+
+        end_time = datetime.now(timezone.utc)
+        processing_time = (end_time - start_time).total_seconds()
+
+        activity.logger.info(
+            f"Agent execution completed: events={event_count}, "
+            f"response_length={len(response_text)}, "
+            f"processing_time={processing_time:.2f}s"
+        )
+
+        return AgentOutput(
+            status="success",
+            response=response_text,
+            total_tokens=total_tokens,
+            num_events=event_count,
+            processing_time_seconds=processing_time,
+        )
+
+    except Exception as e:
+        activity.logger.error(f"Agent execution failed: {e}", exc_info=True)
+
+        end_time = datetime.now(timezone.utc)
+        processing_time = (end_time - start_time).total_seconds()
+
+        return AgentOutput(
+            status="error",
+            response="",
+            error_message=str(e),
+            processing_time_seconds=processing_time,
+        )
+
+
+@activity.defn
+async def log_result_activity(output: AgentOutput) -> None:
+    """
+    Log the agent execution result.
+
+    In a production system, this would persist results to a database.
+    Here we keep it simple for the cookbook example.
+    """
+    if output.status == "success":
+        activity.logger.info(
+            f"Agent succeeded: {len(output.response)} chars, "
+            f"{output.total_tokens} tokens, "
+            f"{output.processing_time_seconds:.2f}s"
+        )
+    else:
+        activity.logger.error(f"Agent failed: {output.error_message}")

agents/claude_agent_sdk_python/activities/__init__.py

@@ -0,0 +1,47 @@
+"""
+Pydantic Models for Agent Execution
+
+Defines the input/output contract between the workflow and activity.
+"""
+
+from typing import Literal, Optional
+from pydantic import BaseModel, Field
+
+
+class AgentInput(BaseModel):
+    """Input for the agent execution activity."""
+
+    prompt: str = Field(..., description="User message to send to the agent")
+    model: str = Field(
+        default="claude-sonnet-4-5-20250929",
+        description="Claude model to use",
+    )
+    system_prompt: Optional[str] = Field(
+        default=None,
+        description="Optional system prompt for the agent",
+    )
+    max_turns: int = Field(
+        default=30,
+        description="Maximum number of agentic turns (tool call rounds)",
+    )
+    permission_mode: str = Field(
+        default="bypassPermissions",
+        description="Claude Code permission mode (e.g. 'bypassPermissions', 'default')",
+    )
+
+
+class AgentOutput(BaseModel):
+    """Output from the agent execution activity."""
+
+    status: Literal["success", "error"] = Field(
+        ..., description="Overall execution status"
+    )
+    response: str = Field(default="", description="Final assistant response text")
+    total_tokens: int = Field(default=0, description="Total tokens used")
+    num_events: int = Field(default=0, description="Number of SDK events processed")
+    processing_time_seconds: Optional[float] = Field(
+        default=None, description="Wall-clock time in seconds"
+    )
+    error_message: Optional[str] = Field(
+        default=None, description="Error details if status is 'error'"
+    )

agents/claude_agent_sdk_python/activities/agent_executor.py

@@ -0,0 +1,22 @@
+[project]
+name = "cookbook-claude-agent-sdk-python"
+version = "0.1"
+description = "Durable agent execution using Claude Agent SDK with Temporal"
+authors = [{ name = "Temporal Technologies Inc", email = "sdk@temporal.io" }]
+requires-python = ">=3.10"
+readme = "README.md"
+license = "MIT"
+dependencies = [
+    "temporalio>=1.15.0,<2",
+    "claude-agent-sdk==0.1.6",
+    "pydantic>=2.0.0",
+]
+
+[project.optional-dependencies]
+test = [
+    "pytest>=7.0",
+    "pytest-asyncio>=0.21",
+]
+
+[tool.pytest.ini_options]
+asyncio_mode = "strict"

agents/claude_agent_sdk_python/models.py

@@ -0,0 +1,47 @@
+"""
+Start an Agent Execution Workflow
+
+Submits a prompt to the Claude Agent SDK via Temporal and prints the result.
+
+Usage:
+    uv run python -m start_workflow "explain how binary search works"
+    uv run python -m start_workflow "what files are in the current directory?"
+"""
+
+import asyncio
+import sys
+import uuid
+
+from temporalio.client import Client
+from temporalio.contrib.pydantic import pydantic_data_converter
+
+from workflows.agent import AgentExecutionWorkflow
+from models import AgentInput
+
+
+async def main():
+    client = await Client.connect(
+        "localhost:7233",
+        data_converter=pydantic_data_converter,
+    )
+
+    prompt = sys.argv[1] if len(sys.argv) > 1 else "Tell me about recursion"
+
+    input_data = AgentInput(prompt=prompt)
+
+    result = await client.execute_workflow(
+        AgentExecutionWorkflow.run,
+        input_data,
+        id=f"claude-agent-sdk-{uuid.uuid4()}",
+        task_queue="claude-agent-sdk-task-queue",
+    )
+
+    print(f"\nStatus: {result.status}")
+    print(f"Tokens: {result.total_tokens}")
+    print(f"Events: {result.num_events}")
+    print(f"Time:   {result.processing_time_seconds:.2f}s")
+    print(f"\nResponse:\n{result.response}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())