diff --git a/README.md b/README.md index 587a3098..1dff1ea7 100644 --- a/README.md +++ b/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. diff --git a/claude_agent_sdk_plugin/README.md b/claude_agent_sdk_plugin/README.md new file mode 100644 index 00000000..0c0ba485 --- /dev/null +++ b/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. diff --git a/claude_agent_sdk_plugin/__init__.py b/claude_agent_sdk_plugin/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/claude_agent_sdk_plugin/agent_chat/README.md b/claude_agent_sdk_plugin/agent_chat/README.md new file mode 100644 index 00000000..e2b3686f --- /dev/null +++ b/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 +``` diff --git a/claude_agent_sdk_plugin/agent_chat/__init__.py b/claude_agent_sdk_plugin/agent_chat/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/claude_agent_sdk_plugin/agent_chat/run_worker.py b/claude_agent_sdk_plugin/agent_chat/run_worker.py new file mode 100644 index 00000000..9f9ecd03 --- /dev/null +++ b/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()) diff --git a/claude_agent_sdk_plugin/agent_chat/workflow.py b/claude_agent_sdk_plugin/agent_chat/workflow.py new file mode 100644 index 00000000..5f0bc5f6 --- /dev/null +++ b/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() + ) diff --git a/claude_agent_sdk_plugin/fork_session/README.md b/claude_agent_sdk_plugin/fork_session/README.md new file mode 100644 index 00000000..b170c628 --- /dev/null +++ b/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. diff --git a/claude_agent_sdk_plugin/fork_session/__init__.py b/claude_agent_sdk_plugin/fork_session/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/claude_agent_sdk_plugin/fork_session/run_worker.py b/claude_agent_sdk_plugin/fork_session/run_worker.py new file mode 100644 index 00000000..723b9d45 --- /dev/null +++ b/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()) diff --git a/claude_agent_sdk_plugin/fork_session/workflow.py b/claude_agent_sdk_plugin/fork_session/workflow.py new file mode 100644 index 00000000..6df6ed2c --- /dev/null +++ b/claude_agent_sdk_plugin/fork_session/workflow.py @@ -0,0 +1,91 @@ +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 + + +@dataclass +class Branch: + session_id: str + prompt: str + answer: str + + +@workflow.defn +class ForkSessionWorkflow: + @workflow.run + async def run(self, prompt: str, follow_ups: list[str]) -> list[Branch]: + """Answer the prompt, then each follow-up in a fork of that conversation.""" + base = await self.ask(prompt) + + # Each fork starts from the base conversation and leaves it as it was. + return list( + await asyncio.gather( + *(self.ask(follow_up, base.session_id) for follow_up in follow_ups) + ) + ) + + def options( + self, *, resume: str | None = None, fork_session: bool = False + ) -> ClaudeAgentOptions: + """Options that start a new session, or continue the one given by resume. + + With fork_session, resume is the session to fork from, and the fork is a + new session. + """ + return ClaudeAgentOptions( + setting_sources=[], + env={"CLAUDE_CODE_RESUME_INTERRUPTED_TURN": "1"}, + resume=resume, + fork_session=fork_session, + ) + + async def ask(self, prompt: str, fork_from: str | None = None) -> Branch: + """Answer the prompt, in a fork of fork_from if given. + + Resumes the session if the connection is lost. + """ + session_id = None # Claude Code's, once it has the prompt + + while True: + try: + if session_id is not None: + options = self.options(resume=session_id) + elif fork_from is not None: + options = self.options(resume=fork_from, fork_session=True) + else: + options = self.options() + + transport = TemporalTransport(options) + async with ClaudeSDKClient( + options=options, transport=transport + ) as client: + # Keep sending the prompt until Claude Code reports a session, which + # means it received it. After that, a resume finishes the turn. + if session_id is None: + await client.query(prompt) + + async for message in client.receive_response(): + if ( + isinstance(message, SystemMessage) + and message.subtype == "init" + ): + session_id = message.data["session_id"] + elif isinstance(message, ResultMessage): + return Branch( + message.session_id, prompt, message.result or "" + ) + + raise RuntimeError("Claude Code ended the turn without a result") + + except CLIConnectionError: + # Resume the fork if Claude Code started it, or start it again. + pass diff --git a/claude_agent_sdk_plugin/human_in_the_loop/README.md b/claude_agent_sdk_plugin/human_in_the_loop/README.md new file mode 100644 index 00000000..2ccacd07 --- /dev/null +++ b/claude_agent_sdk_plugin/human_in_the_loop/README.md @@ -0,0 +1,69 @@ +# Human in the Loop + +Claude Code pauses before each tool call until a person approves or denies it. + +## What this shows + +- **A person can sit in the loop for days.** Each approval is a durable wait in the Workflow, so it survives Worker restarts. +- **Approvals are just Workflow state.** A Query shows what's waiting, and an Update decides it. Anything that can call Temporal can be the approval UI. +- **Denial is an answer, not a failure.** A denied command doesn't run, and Claude hears why and carries on. +- **Cancel stops the agent.** Cancelling the Workflow mid-wait ends the turn and stops Claude Code. + +## 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. Claude Code works in the Worker's current directory. + +```bash +uv run --group claude-agent-sdk claude_agent_sdk_plugin/human_in_the_loop/run_worker.py +``` + +The Worker runs offline: everything runs for real except the model, which follows a script. To use the real model, put `ANTHROPIC_API_KEY=...` in front of the command. + +Start a task: + +```bash +temporal workflow start \ + --type HumanInTheLoopWorkflow \ + --task-queue claude-agent-sdk-human-in-the-loop \ + --workflow-id human-in-the-loop \ + --input '"Create hello.txt with a greeting."' +``` + +See what is waiting for a decision: + +```bash +temporal workflow query --workflow-id human-in-the-loop --name pending_tool_calls -o json +``` + +Approve it, using the `tool_use_id` from the Query. A live model picks its own IDs and may ask for more than one call; decide each one. + +```bash +temporal workflow update execute \ + --workflow-id human-in-the-loop \ + --name decide \ + --input '{"tool_use_id": "toolu_hello", "approved": true}' +``` + +Then get the result: + +```bash +temporal workflow result --workflow-id human-in-the-loop +``` + +Offline, the result is `"Created hello.txt."`, and `hello.txt` is in the Worker's directory, so Claude Code really ran the command. + +To deny instead, send `"approved": false`. Offline, the result is `"Understood. I did not create hello.txt."`, and no file appears. The script picks its reply from the tool result in Claude Code's request, so the two replies show your decision reached Claude Code. + +To cancel, start another task with a new Workflow ID, then cancel it while a call is pending: + +```bash +temporal workflow cancel --workflow-id +``` + +The Workflow ends as Canceled. Its history shows the `claude-agent-sdk-close` Activity that stopped Claude Code. diff --git a/claude_agent_sdk_plugin/human_in_the_loop/__init__.py b/claude_agent_sdk_plugin/human_in_the_loop/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/claude_agent_sdk_plugin/human_in_the_loop/run_worker.py b/claude_agent_sdk_plugin/human_in_the_loop/run_worker.py new file mode 100644 index 00000000..7c181f4d --- /dev/null +++ b/claude_agent_sdk_plugin/human_in_the_loop/run_worker.py @@ -0,0 +1,60 @@ +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.human_in_the_loop.workflow import HumanInTheLoopWorkflow +from claude_agent_sdk_plugin.offline import Answer, offline_api_without_key, tool_use + +TOOL_USE_ID = "toolu_hello" + + +def tool_result(request: dict[str, Any]) -> dict[str, Any]: + return next( + block + for message in request["messages"] + if isinstance(message["content"], list) + for block in message["content"] + if block.get("tool_use_id") == TOOL_USE_ID + ) + + +def after_decision(request: dict[str, Any]) -> str: + if tool_result(request).get("is_error"): + return "Understood. I did not create hello.txt." + return "Created hello.txt." + + +OFFLINE_ANSWERS: dict[str, Answer] = { + f"tool_result:{TOOL_USE_ID}": after_decision, + "hello.txt": tool_use( + TOOL_USE_ID, "Bash", command="echo 'Hello from Claude Code' > hello.txt" + ), +} + + +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-human-in-the-loop", + workflows=[HumanInTheLoopWorkflow], + plugins=[ClaudeAgentSDKPlugin()], + ): + print(f"Worker started. To stop it gracefully: kill {os.getpid()}") + await stop.wait() + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/claude_agent_sdk_plugin/human_in_the_loop/workflow.py b/claude_agent_sdk_plugin/human_in_the_loop/workflow.py new file mode 100644 index 00000000..416f9866 --- /dev/null +++ b/claude_agent_sdk_plugin/human_in_the_loop/workflow.py @@ -0,0 +1,109 @@ +from dataclasses import dataclass +from typing import Any + +from claude_agent_sdk import ( + ClaudeAgentOptions, + ClaudeSDKClient, + CLIConnectionError, + PermissionResult, + PermissionResultAllow, + PermissionResultDeny, + ResultMessage, + SystemMessage, + ToolPermissionContext, +) +from temporalio import workflow +from temporalio.claude_agent_sdk import TemporalTransport + + +@dataclass +class ToolRequest: + tool_use_id: str + tool_name: str + tool_input: dict[str, Any] + + +@dataclass +class Decision: + tool_use_id: str + approved: bool + + +@workflow.defn +class HumanInTheLoopWorkflow: + def __init__(self) -> None: + self.pending: dict[str, ToolRequest] = {} + self.decisions: dict[str, bool] = {} + + @workflow.run + async def run(self, task: str) -> str: + return await self.ask(task) + + def options(self, *, resume: str | None = None) -> ClaudeAgentOptions: + """Options that start a new session, or continue the one given by resume.""" + return ClaudeAgentOptions( + # With permission_mode="default", every Bash command asks can_use_tool. + tools=["Bash"], + permission_mode="default", + can_use_tool=self.ask_human, + setting_sources=[], + env={"CLAUDE_CODE_RESUME_INTERRUPTED_TURN": "1"}, + resume=resume, + ) + + async def ask(self, prompt: str) -> str: + """Answer the prompt, resuming the session if the connection is lost.""" + session_id = None # Claude Code's, once it has the prompt + + while True: + try: + options = self.options(resume=session_id) + transport = TemporalTransport(options) + async with ClaudeSDKClient( + options=options, transport=transport + ) as client: + # Keep sending the prompt until Claude Code reports a session, which + # means it received it. After that, a resume finishes the turn. + if session_id is None: + await client.query(prompt) + + async for message in client.receive_response(): + if ( + isinstance(message, SystemMessage) + and message.subtype == "init" + ): + session_id = message.data["session_id"] + elif isinstance(message, ResultMessage): + return message.result or "" + + raise RuntimeError("Claude Code ended the turn without a result") + + except CLIConnectionError: + # Resume the session if Claude Code started it, or start again. + pass + + async def ask_human( + self, tool_name: str, tool_input: dict[str, Any], context: ToolPermissionContext + ) -> PermissionResult: + tool_use_id = context.tool_use_id or "" + self.pending[tool_use_id] = ToolRequest(tool_use_id, tool_name, tool_input) + try: + await workflow.wait_condition(lambda: tool_use_id in self.decisions) + finally: + del self.pending[tool_use_id] + if self.decisions[tool_use_id]: + return PermissionResultAllow() + return PermissionResultDeny(message="A person denied this tool call") + + @workflow.query + def pending_tool_calls(self) -> list[ToolRequest]: + return list(self.pending.values()) + + @workflow.update + def decide(self, decision: Decision) -> None: + self.decisions[decision.tool_use_id] = decision.approved + + @decide.validator + def validate_decide(self, decision: Decision) -> None: + if decision.tool_use_id not in self.pending: + raise ValueError(f"No pending tool call {decision.tool_use_id}") diff --git a/claude_agent_sdk_plugin/multi_agent/README.md b/claude_agent_sdk_plugin/multi_agent/README.md new file mode 100644 index 00000000..a5556df7 --- /dev/null +++ b/claude_agent_sdk_plugin/multi_agent/README.md @@ -0,0 +1,43 @@ +# Multi-Agent + +Claude Code's native subagent delegation, run as a Temporal Workflow. A coordinator delegates to two subagents defined in `ClaudeAgentOptions(agents=...)`: a `researcher` that gathers facts and a `writer` that turns them into a briefing. + +## What this shows + +- **Claude Code's own subagents, made durable.** You define roles with `AgentDefinition` as usual, and Claude Code delegates to them inside one process. Temporal adds durability without changing how delegation works. +- **You can see who said what.** Every subagent message points back to the call that started it, so the Workflow can return and log each role's output separately. +- **No person needed.** Delegation is pre-approved, so the team runs end to end on its own. + +## 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/multi_agent/run_worker.py +``` + +The Worker runs offline: everything runs for real except the model, which follows a script. To use the real model, put `ANTHROPIC_API_KEY=...` in front of the command. + +Then run the Workflow: + +```bash +temporal workflow execute \ + --type MultiAgentWorkflow \ + --task-queue claude-agent-sdk-multi-agent \ + --workflow-id multi-agent \ + --input '"durable execution"' +``` + +## Expected behavior + +The result has the final briefing and, under `messages_by_agent`, what each of `coordinator`, `researcher`, and `writer` said. The Workflow also logs each message with its role. + +Offline, the researcher returns three facts, the writer one sentence, and the coordinator `Final briefing: Temporal keeps your code running through failures.` The script answers the researcher and writer only when it sees their prompts, so those replies show Claude Code really ran each subagent. The grouping by role comes from Claude Code's real messages. + +If the Worker restarts mid-run, the Workflow resumes the session on the restarted Worker. Claude Code finishes the cut-off turn. diff --git a/claude_agent_sdk_plugin/multi_agent/__init__.py b/claude_agent_sdk_plugin/multi_agent/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/claude_agent_sdk_plugin/multi_agent/run_worker.py b/claude_agent_sdk_plugin/multi_agent/run_worker.py new file mode 100644 index 00000000..f7bbd37b --- /dev/null +++ b/claude_agent_sdk_plugin/multi_agent/run_worker.py @@ -0,0 +1,59 @@ +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.multi_agent.workflow import MultiAgentWorkflow +from claude_agent_sdk_plugin.offline import Answer, offline_api_without_key, tool_use + +RESEARCH_NOTES = "- Workflows are durable.\n- Activities are retried.\n- Signals are delivered exactly once." +BRIEFING = "Temporal keeps your code running through failures." + + +OFFLINE_ANSWERS: dict[str, Answer] = { + "short briefing on this topic": tool_use( + "toolu_research", + "Agent", + description="Research the topic", + subagent_type="researcher", + prompt="Collect three key facts about the topic.", + run_in_background=False, + ), + "Collect three key facts": RESEARCH_NOTES, + "tool_result:toolu_research": tool_use( + "toolu_write", + "Agent", + description="Write the briefing", + subagent_type="writer", + prompt="Turn these notes into a one-sentence briefing.", + run_in_background=False, + ), + "Turn these notes": BRIEFING, + "tool_result:toolu_write": f"Final briefing: {BRIEFING}", +} + + +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-multi-agent", + workflows=[MultiAgentWorkflow], + plugins=[ClaudeAgentSDKPlugin()], + ): + print(f"Worker started. To stop it gracefully: kill {os.getpid()}") + await stop.wait() + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/claude_agent_sdk_plugin/multi_agent/workflow.py b/claude_agent_sdk_plugin/multi_agent/workflow.py new file mode 100644 index 00000000..205afe5e --- /dev/null +++ b/claude_agent_sdk_plugin/multi_agent/workflow.py @@ -0,0 +1,129 @@ +import json +from dataclasses import asdict, dataclass, field + +from claude_agent_sdk import ( + AgentDefinition, + AssistantMessage, + ClaudeAgentOptions, + ClaudeSDKClient, + CLIConnectionError, + ResultMessage, + SystemMessage, + TextBlock, + ToolUseBlock, +) +from temporalio import workflow +from temporalio.claude_agent_sdk import TemporalTransport + +COORDINATOR = "coordinator" + +AGENTS = { + "researcher": AgentDefinition( + description="Finds the key facts about a topic.", + prompt="You are a researcher. Reply with a short list of key facts.", + tools=["WebSearch", "WebFetch"], + ), + "writer": AgentDefinition( + description="Turns notes into a short, readable briefing.", + prompt="You are a writer. Turn the notes you are given into a briefing.", + tools=[], + ), +} +# The same agents, as the Claude Agent SDK sends them to Claude Code. +AGENTS_JSON = { + name: {key: value for key, value in asdict(agent).items() if value is not None} + for name, agent in AGENTS.items() +} + + +@dataclass +class Briefing: + result: str + messages_by_agent: dict[str, list[str]] = field(default_factory=dict) + + +@workflow.defn +class MultiAgentWorkflow: + def __init__(self) -> None: + # Kept across resumes, so a resumed session adds to them. + self.roles: dict[str, str] = {} + self.messages_by_agent: dict[str, list[str]] = {} + + @workflow.run + async def run(self, topic: str) -> Briefing: + prompt = ( + f"Write a short briefing on this topic: {topic}. Delegate the " + "research to the researcher, then give the notes to the writer. " + "Wait for each subagent's result before the next step (pass " + "run_in_background: false)." + ) + return await self.ask(prompt) + + def options(self, resume: str | None = None) -> ClaudeAgentOptions: + """Options that start a new session, or resume the one with this ID.""" + return ClaudeAgentOptions( + agents=AGENTS, + # A resumed Claude Code finishes the cut-off turn before it reads the + # agents above or the forward_subagent_text request, so give both on + # the command line too. + extra_args={ + "agents": json.dumps(AGENTS_JSON), + "forward-subagent-text": None, + }, + # A subagent's tools must also be listed here. "Agent" is the delegation + # tool in this Claude Code version. + tools=["Agent", "WebSearch", "WebFetch"], + allowed_tools=["Agent", "WebSearch", "WebFetch"], + permission_mode="default", + # Forward the subagents' text, not only their tool calls. + forward_subagent_text=True, + setting_sources=[], + env={"CLAUDE_CODE_RESUME_INTERRUPTED_TURN": "1"}, + resume=resume, + ) + + async def ask(self, prompt: str) -> Briefing: + """Answer the prompt in a session, resuming it if the connection is lost.""" + session_id = None # Claude Code's, once it has the prompt + + while True: + try: + options = self.options(resume=session_id) + transport = TemporalTransport(options) + async with ClaudeSDKClient( + options=options, transport=transport + ) as client: + # Keep sending the prompt until Claude Code reports a session, which + # means it received it. After that, a resume finishes the turn. + if session_id is None: + await client.query(prompt) + + async for message in client.receive_response(): + if ( + isinstance(message, SystemMessage) + and message.subtype == "init" + ): + session_id = message.data["session_id"] + elif isinstance(message, AssistantMessage): + self.record(message) + elif isinstance(message, ResultMessage): + return Briefing( + message.result or "", self.messages_by_agent + ) + + raise RuntimeError("Claude Code ended the turn without a result") + + except CLIConnectionError: + # Resume the session if Claude Code started it, or start again. + pass + + def record(self, message: AssistantMessage) -> None: + # A subagent's messages carry parent_tool_use_id, the Agent call that + # started it. + role = self.roles.get(message.parent_tool_use_id or "", COORDINATOR) + for block in message.content: + if isinstance(block, ToolUseBlock) and block.name == "Agent": + self.roles[block.id] = block.input["subagent_type"] + elif isinstance(block, TextBlock): + workflow.logger.info("%s: %s", role, block.text) + self.messages_by_agent.setdefault(role, []).append(block.text) diff --git a/claude_agent_sdk_plugin/offline.py b/claude_agent_sdk_plugin/offline.py new file mode 100644 index 00000000..0f5f56be --- /dev/null +++ b/claude_agent_sdk_plugin/offline.py @@ -0,0 +1,180 @@ +from __future__ import annotations + +import contextlib +import json +import os +import tempfile +import threading +import uuid +from collections.abc import Callable, Iterator, Mapping +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import Any, Union + +# Outlives the Worker, so a restarted Worker finds the sessions it can resume. +OFFLINE_CONFIG_DIR = os.path.join(tempfile.gettempdir(), "claude-agent-sdk-samples") + +Answer = Union[str, list[str], dict[str, Any], Callable[[dict[str, Any]], Any]] + + +def is_online() -> bool: + return bool( + os.environ.get("ANTHROPIC_API_KEY") or os.environ.get("CLAUDE_CODE_OAUTH_TOKEN") + ) + + +def tool_use(id: str, name: str, **tool_input: Any) -> dict[str, Any]: + return {"type": "tool_use", "id": id, "name": name, "input": tool_input} + + +class ScriptedApi: + def __init__(self, answers: Mapping[str, Answer]) -> None: + self.answers = dict(answers) + self.requests: list[dict[str, Any]] = [] + self.asked: list[str] = [] + self._server = ThreadingHTTPServer(("127.0.0.1", 0), self._handler()) + + @property + def url(self) -> str: + return f"http://127.0.0.1:{self._server.server_address[1]}" + + def answer(self, body: dict[str, Any]) -> Any: + self.requests.append(body) + prompt = prompt_of(body) + key = next((k for k in self.answers if k in prompt), None) + if key is None: + return "" + self.asked.append(key) + answer = self.answers[key] + return answer(body) if callable(answer) else answer + + def _handler(self) -> type[BaseHTTPRequestHandler]: + api = self + + class Handler(BaseHTTPRequestHandler): + def do_POST(self) -> None: + body = json.loads(self.rfile.read(int(self.headers["Content-Length"]))) + data = stream(api.answer(body), body.get("model", "claude")) + with contextlib.suppress(OSError): + self.send_response(200) + self.send_header("Content-Type", "text/event-stream") + self.send_header("Content-Length", str(len(data))) + self.end_headers() + self.wfile.write(data) + + def log_message(self, format: str, *args: Any) -> None: + pass + + return Handler + + +@contextlib.contextmanager +def scripted_api(answers: Mapping[str, Answer]) -> Iterator[ScriptedApi]: + api = ScriptedApi(answers) + threading.Thread(target=api._server.serve_forever, daemon=True).start() + env = { + "ANTHROPIC_API_KEY": "offline", + "ANTHROPIC_BASE_URL": api.url, + "CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC": "1", + "CLAUDE_CONFIG_DIR": OFFLINE_CONFIG_DIR, + } + inherited = [ + name + for name in os.environ + if name == "CLAUDECODE" or name.startswith("CLAUDE_CODE_") + ] + saved = {name: os.environ.get(name) for name in [*inherited, *env]} + for name in inherited: + del os.environ[name] + os.environ.update(env) + try: + yield api + finally: + for name, value in saved.items(): + if value is None: + os.environ.pop(name, None) + else: + os.environ[name] = value + api._server.shutdown() + api._server.server_close() + + +@contextlib.contextmanager +def offline_api_without_key(answers: Mapping[str, Answer]) -> Iterator[None]: + if is_online(): + yield + return + + print("No ANTHROPIC_API_KEY, so running offline against a scripted Anthropic API.") + with scripted_api(answers): + yield + + +def prompt_of(body: dict[str, Any]) -> str: + content = next( + m["content"] for m in reversed(body["messages"]) if m["role"] == "user" + ) + if isinstance(content, str): + return content + return " ".join( + block.get("text") or f"tool_result:{block.get('tool_use_id')}" + for block in content + ) + + +def stream(answer: Any, model: str = "claude") -> bytes: + message = { + "id": f"msg_{uuid.uuid4().hex}", + "type": "message", + "model": model, + "role": "assistant", + "content": [], + "usage": {"input_tokens": 1}, + } + events: list[dict[str, Any]] = [{"type": "message_start", "message": message}] + if isinstance(answer, dict): + events += [ + { + "type": "content_block_start", + "index": 0, + "content_block": {**answer, "input": {}}, + }, + { + "type": "content_block_delta", + "index": 0, + "delta": { + "type": "input_json_delta", + "partial_json": json.dumps(answer["input"]), + }, + }, + ] + stop_reason = "tool_use" + else: + chunks = answer if isinstance(answer, list) else [answer] + events.append( + { + "type": "content_block_start", + "index": 0, + "content_block": {"type": "text", "text": ""}, + } + ) + events += [ + { + "type": "content_block_delta", + "index": 0, + "delta": {"type": "text_delta", "text": chunk}, + } + for chunk in chunks + ] + stop_reason = "end_turn" + events += [ + {"type": "content_block_stop", "index": 0}, + { + "type": "message_delta", + "delta": {"stop_reason": stop_reason}, + "usage": {"output_tokens": 1}, + }, + {"type": "message_stop"}, + ] + return "".join( + f"event: {e['type']}\ndata: {json.dumps(e)}\n\n" for e in events + ).encode() diff --git a/claude_agent_sdk_plugin/open_telemetry/README.md b/claude_agent_sdk_plugin/open_telemetry/README.md new file mode 100644 index 00000000..c90e7fb2 --- /dev/null +++ b/claude_agent_sdk_plugin/open_telemetry/README.md @@ -0,0 +1,78 @@ +# OpenTelemetry + +Trace a Workflow and the Claude Code it runs as one OpenTelemetry trace: the caller, the Workflow, its Activities, and Claude Code's own model and tool spans. + +## What this shows + +- **One trace across processes.** `OpenTelemetryPlugin` carries the trace from the starter into the Workflow and its Activities. The plugin passes it on to Claude Code in `TRACEPARENT`, so Claude Code's spans join the same trace. +- **Claude Code's own spans.** Each turn is a `claude_code.interaction` span, with a `claude_code.llm_request` span for each model call and a `claude_code.tool` span for each tool call. +- **Replay-safe Workflow spans.** `create_tracer_provider()` gives Workflow spans IDs that stay the same on replay, and exports them once. + +## Run it + +All commands run from the repository root. Start a Temporal dev server in a separate terminal: + +```bash +temporal server start-dev +``` + +Start [Jaeger](https://www.jaegertracing.io/) to receive and show the traces. It takes OTLP on port 4317: + +```bash +docker run --rm -p 16686:16686 -p 4317:4317 jaegertracing/all-in-one:latest +``` + +Start the Worker in another terminal: + +```bash +uv run --group claude-agent-sdk claude_agent_sdk_plugin/open_telemetry/run_worker.py +``` + +The Worker runs offline: everything runs for real except the model, which asks for one Bash command and then answers with fixed text. To use the real model, put `ANTHROPIC_API_KEY=...` in front of the command. + +The Worker sets Claude Code's tracing variables in its own environment, from [telemetry.py](telemetry.py): + +```bash +CLAUDE_CODE_ENABLE_TELEMETRY=1 +CLAUDE_CODE_ENHANCED_TELEMETRY_BETA=1 +OTEL_TRACES_EXPORTER=otlp +OTEL_EXPORTER_OTLP_PROTOCOL=grpc +OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4317 +``` + +Run the Workflow from a third terminal: + +```bash +uv run --group claude-agent-sdk python -m claude_agent_sdk_plugin.open_telemetry.run_workflow +``` + +## Expected behavior + +The starter prints the result and the trace ID: + +``` +Result: I counted the Python files with ls. +Trace ID: aa8b95b37c1b39f76cc9102fa3d1d5b9 +``` + +Open the trace at `http://localhost:16686/trace/`. It spans three services: + +``` +ask-agent claude-agent-sdk-starter +└─ StartWorkflow:OpenTelemetryWorkflow claude-agent-sdk-starter + └─ RunWorkflow:OpenTelemetryWorkflow claude-agent-sdk-worker + ├─ StartActivity:claude-agent-sdk-connect claude-agent-sdk-worker + │ └─ RunActivity:claude-agent-sdk-connect claude-agent-sdk-worker + │ └─ claude_code.interaction claude-code + │ ├─ claude_code.llm_request + │ ├─ claude_code.tool + │ │ ├─ claude_code.tool.blocked_on_user + │ │ └─ claude_code.tool.execution + │ └─ claude_code.llm_request + ├─ StartActivity:claude-agent-sdk-write (one per write) + │ └─ RunActivity:claude-agent-sdk-write + ├─ StartActivity:claude-agent-sdk-read (one per read) + │ └─ RunActivity:claude-agent-sdk-read + └─ StartActivity:claude-agent-sdk-close + └─ RunActivity:claude-agent-sdk-close +``` diff --git a/claude_agent_sdk_plugin/open_telemetry/__init__.py b/claude_agent_sdk_plugin/open_telemetry/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/claude_agent_sdk_plugin/open_telemetry/run_worker.py b/claude_agent_sdk_plugin/open_telemetry/run_worker.py new file mode 100644 index 00000000..d587e0e3 --- /dev/null +++ b/claude_agent_sdk_plugin/open_telemetry/run_worker.py @@ -0,0 +1,53 @@ +import asyncio +import os +import signal + +from temporalio.claude_agent_sdk import ClaudeAgentSDKPlugin +from temporalio.client import Client +from temporalio.contrib.opentelemetry import OpenTelemetryPlugin +from temporalio.worker import Worker + +from claude_agent_sdk_plugin.offline import Answer, offline_api_without_key, tool_use +from claude_agent_sdk_plugin.open_telemetry.telemetry import ( + CLAUDE_CODE_TRACING, + install_tracer_provider, +) +from claude_agent_sdk_plugin.open_telemetry.workflow import OpenTelemetryWorkflow + +OFFLINE_ANSWERS: dict[str, Answer] = { + "How many Python files": tool_use( + "toolu_count", + "Bash", + command="ls *.py | wc -l", + description="Count the Python files", + ), + "tool_result:toolu_count": "I counted the Python files with ls.", +} + + +async def main() -> None: + install_tracer_provider("claude-agent-sdk-worker") + client = await Client.connect( + os.environ.get("TEMPORAL_ADDRESS", "localhost:7233"), + plugins=[OpenTelemetryPlugin(add_temporal_spans=True)], + ) + + # 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): + os.environ.update(CLAUDE_CODE_TRACING) + async with Worker( + client, + task_queue="claude-agent-sdk-open-telemetry", + workflows=[OpenTelemetryWorkflow], + plugins=[ClaudeAgentSDKPlugin()], + ): + print(f"Worker started. To stop it gracefully: kill {os.getpid()}") + await stop.wait() + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/claude_agent_sdk_plugin/open_telemetry/run_workflow.py b/claude_agent_sdk_plugin/open_telemetry/run_workflow.py new file mode 100644 index 00000000..73bb942b --- /dev/null +++ b/claude_agent_sdk_plugin/open_telemetry/run_workflow.py @@ -0,0 +1,39 @@ +import asyncio +import os +import sys + +import opentelemetry.trace +from temporalio.client import Client +from temporalio.contrib.opentelemetry import OpenTelemetryPlugin + +from claude_agent_sdk_plugin.open_telemetry.telemetry import install_tracer_provider +from claude_agent_sdk_plugin.open_telemetry.workflow import OpenTelemetryWorkflow + +DEFAULT_PROMPT = "How many Python files are in the current directory?" + + +async def main() -> None: + install_tracer_provider("claude-agent-sdk-starter") + client = await Client.connect( + os.environ.get("TEMPORAL_ADDRESS", "localhost:7233"), + plugins=[OpenTelemetryPlugin(add_temporal_spans=True)], + ) + + # The root of the trace. The Workflow, its Activities, and Claude Code's + # spans all join it. + tracer = opentelemetry.trace.get_tracer(__name__) + with tracer.start_as_current_span("ask-agent") as span: + result = await client.execute_workflow( + OpenTelemetryWorkflow.run, + " ".join(sys.argv[1:]) or DEFAULT_PROMPT, + id="claude-agent-sdk-open-telemetry", + task_queue="claude-agent-sdk-open-telemetry", + ) + trace_id = format(span.get_span_context().trace_id, "032x") + + print(f"Result: {result}") + print(f"Trace ID: {trace_id}") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/claude_agent_sdk_plugin/open_telemetry/telemetry.py b/claude_agent_sdk_plugin/open_telemetry/telemetry.py new file mode 100644 index 00000000..2f552762 --- /dev/null +++ b/claude_agent_sdk_plugin/open_telemetry/telemetry.py @@ -0,0 +1,29 @@ +import os + +import opentelemetry.trace +from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter +from opentelemetry.sdk.resources import SERVICE_NAME, Resource +from opentelemetry.sdk.trace.export import BatchSpanProcessor +from temporalio.contrib.opentelemetry import create_tracer_provider + +OTLP_ENDPOINT = os.environ.get("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4317") + +# Turns on Claude Code's tracing. Set in the Worker's environment, not in +# options.env, which is recorded in Workflow history. +CLAUDE_CODE_TRACING = { + "CLAUDE_CODE_ENABLE_TELEMETRY": "1", + "CLAUDE_CODE_ENHANCED_TELEMETRY_BETA": "1", + "OTEL_TRACES_EXPORTER": "otlp", + "OTEL_EXPORTER_OTLP_PROTOCOL": "grpc", + "OTEL_EXPORTER_OTLP_ENDPOINT": OTLP_ENDPOINT, + "OTEL_SERVICE_NAME": "claude-code", +} + + +def install_tracer_provider(service_name: str) -> None: + """Export this process's spans, including replay-safe Workflow spans.""" + provider = create_tracer_provider( + resource=Resource.create({SERVICE_NAME: service_name}) + ) + provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter(OTLP_ENDPOINT))) + opentelemetry.trace.set_tracer_provider(provider) diff --git a/claude_agent_sdk_plugin/open_telemetry/workflow.py b/claude_agent_sdk_plugin/open_telemetry/workflow.py new file mode 100644 index 00000000..a638b3bb --- /dev/null +++ b/claude_agent_sdk_plugin/open_telemetry/workflow.py @@ -0,0 +1,56 @@ +from claude_agent_sdk import ( + ClaudeAgentOptions, + ClaudeSDKClient, + CLIConnectionError, + ResultMessage, + SystemMessage, +) +from temporalio import workflow +from temporalio.claude_agent_sdk import TemporalTransport + + +@workflow.defn +class OpenTelemetryWorkflow: + @workflow.run + async def run(self, prompt: str) -> str: + return await self.ask(prompt) + + def options(self, *, resume: str | None = None) -> ClaudeAgentOptions: + """Options that start a new session, or continue the one given by resume.""" + return ClaudeAgentOptions( + setting_sources=[], + allowed_tools=["Bash"], + env={"CLAUDE_CODE_RESUME_INTERRUPTED_TURN": "1"}, + resume=resume, + ) + + async def ask(self, prompt: str) -> str: + """Answer the prompt, resuming the session if the connection is lost.""" + session_id = None # Claude Code's, once it has the prompt + + while True: + try: + options = self.options(resume=session_id) + transport = TemporalTransport(options) + async with ClaudeSDKClient( + options=options, transport=transport + ) as client: + # Keep sending the prompt until Claude Code reports a session, which + # means it received it. After that, a resume finishes the turn. + if session_id is None: + await client.query(prompt) + + async for message in client.receive_response(): + if ( + isinstance(message, SystemMessage) + and message.subtype == "init" + ): + session_id = message.data["session_id"] + elif isinstance(message, ResultMessage): + return message.result or "" + + raise RuntimeError("Claude Code ended the turn without a result") + + except CLIConnectionError: + # Resume the session if Claude Code started it, or start again. + pass diff --git a/claude_agent_sdk_plugin/plugin_with_custom_transport/README.md b/claude_agent_sdk_plugin/plugin_with_custom_transport/README.md new file mode 100644 index 00000000..2b31c6e9 --- /dev/null +++ b/claude_agent_sdk_plugin/plugin_with_custom_transport/README.md @@ -0,0 +1,91 @@ +# Plugin with Custom Transport + +Give Claude Code your own skill and MCP server by passing it a Claude Code plugin, and run Claude Code inside a [Docker Sandbox](https://docs.docker.com/ai/sandboxes/) instead of on the Worker. + +A Claude Code plugin is a directory that bundles skills, MCP servers, agents, slash commands, and hooks. This sample's `haiku-writer` plugin has two parts: + +- The `haiku` skill, which tells Claude to answer with a haiku signed `~ haiku-writer`, and to check it with `count_syllables` first. +- The `syllables` MCP server, which provides `count_syllables`. It uses only the standard library, so it runs on the sandbox's `python3` without installing anything. + +```text +haiku-writer/ +├── .claude-plugin/plugin.json +├── .mcp.json +├── servers/syllables.py +└── skills/haiku/SKILL.md +``` + +## What this shows + +- **Your plugins work as they do in Claude Code.** Pass the plugin through `ClaudeAgentOptions.plugins`, and Claude picks up its skills and starts its MCP servers. The Workflow only has to allow the MCP tool, which Claude Code names `mcp__plugin_haiku-writer_syllables__count_syllables`. +- **Bring your own runtime.** A connector decides where Claude Code runs. `SandboxConnector` puts it in a Docker Sandbox, and the Workflow code doesn't change: it still uses a plain `TemporalTransport(options)`. +- **The sandbox outlives the Worker.** `SandboxConnector` creates a sandbox for each Workflow and copies the plugin in before Claude Code starts. The Workflow names the plugin's path in the sandbox. You manage the sandbox's lifecycle: it stays until you remove it. +- **The API key stays out of history and out of the sandbox.** The Worker hands the key to the sandbox's proxy, and Claude Code only ever sees a placeholder. + +## Prerequisites + +[Docker Sandboxes](https://docs.docker.com/ai/sandboxes/) (`sbx`) installed on the Worker's machine and signed in (`sbx login`). + +## 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/plugin_with_custom_transport/run_worker.py +``` + +The Worker runs offline: everything runs for real, including the sandbox and the plugin's MCP server, except the model, which follows a script. To use the real model, put `ANTHROPIC_API_KEY=...` in front of the command. This sample needs an API key: the sandbox's proxy doesn't take a `CLAUDE_CODE_OAUTH_TOKEN`. + +Ask for a haiku from a third terminal: + +```bash +temporal workflow execute \ + --type PluginWithCustomTransportWorkflow \ + --task-queue claude-agent-sdk-plugin-with-custom-transport \ + --workflow-id plugin-with-custom-transport \ + --input '"Write a haiku about durable execution."' +``` + +## Expected behavior + +The result is a haiku followed by the `~ haiku-writer` signature. Offline it is always: + +```text +Workers fall and rise +history remembers all +the work carries on + +~ haiku-writer +``` + +Offline, the model's replies are scripted, but the skill and the MCP server run for real in the sandbox. Open the run in the Temporal UI to see them: + +- The `claude-agent-sdk-connect` Activity's input includes `--plugin-dir` and the plugin's path in the sandbox. +- The `claude-agent-sdk-read` Activities carry the skill's text and the MCP server's counts, `5`, `7`, and `5`. + +Live, the wording is Claude's, and the signature shows Claude followed the skill. + +The sandbox outlives the Worker. If the Worker stops mid-turn, Claude Code stops with it, and the Workflow continues the turn in the same sandbox, with its workspace files intact. Output Claude Code wrote after the Worker's last read is lost, so the Workflow first reads the session's transcript in the sandbox with the `find_saved_turn` Activity. If Claude Code had finished the turn, the final answer is there, and no result is lost. + +## Clean up + +The sandbox stays after the Workflow completes, so you can look at its workspace. `sbx ls` lists it as `claude-sandbox-...`. Remove it with: + +```bash +sbx rm --force claude-sandbox-... +``` + +## Files + +| File | Description | +|------|-------------| +| `haiku-writer/` | The plugin: the `haiku` skill and the `syllables` MCP server | +| `connector.py` | `SandboxConnector`, which copies uploads into a sandbox and runs Claude Code there through `sbx`, and the `find_saved_turn` Activity | +| `workflow.py` | `PluginWithCustomTransportWorkflow`: `ClaudeSDKClient` with the plugin, which finishes the turn across Worker restarts | +| `run_worker.py` | Registers `ClaudeAgentSDKPlugin(connector=SandboxConnector(uploads=...))`, the Activities, and the offline answers | diff --git a/claude_agent_sdk_plugin/plugin_with_custom_transport/__init__.py b/claude_agent_sdk_plugin/plugin_with_custom_transport/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/claude_agent_sdk_plugin/plugin_with_custom_transport/connector.py b/claude_agent_sdk_plugin/plugin_with_custom_transport/connector.py new file mode 100644 index 00000000..044217ce --- /dev/null +++ b/claude_agent_sdk_plugin/plugin_with_custom_transport/connector.py @@ -0,0 +1,219 @@ +import asyncio +import io +import json +import os +import re +import tarfile +import uuid +from collections.abc import Mapping +from dataclasses import dataclass + +from claude_agent_sdk import ( + CLIConnectionError, + InMemorySessionStore, + get_session_messages_from_store, + project_key_for_directory, +) +from temporalio import activity +from temporalio.claude_agent_sdk import ( + ClaudeCodeProcess, + ConnectActivityInput, + Output, + SubprocessConnector, +) + +WORKSPACE = "/home/agent/workspace" +PROXY_MANAGED_KEY = "sk-ant-api03-proxy-managed" +FORWARDED_ENV = ("ANTHROPIC_BASE_URL", "CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC") + + +async def sbx(*args: str, input: bytes | None = None) -> bytes: + process = await asyncio.create_subprocess_exec( + "sbx", + *args, + stdin=asyncio.subprocess.DEVNULL if input is None else asyncio.subprocess.PIPE, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + + try: + stdout, stderr = await process.communicate(input) + finally: + if process.returncode is None: + process.kill() + await process.wait() + + if process.returncode: + message = stderr.decode(errors="replace").strip() + raise RuntimeError(f"sbx {args[0]} failed: {message}") + + return stdout + + +def sandbox_name() -> str: + """The sandbox of the Workflow that runs this Activity.""" + workflow_id = uuid.uuid5(uuid.NAMESPACE_URL, str(activity.info().workflow_id)) + return f"claude-sandbox-{workflow_id}" + + +def sandbox_env(request_env: dict[str, str]) -> dict[str, str]: + # A loopback base URL becomes host.docker.internal, so the sandbox can reach + # a gateway on the Worker's host. + forwarded = { + name: re.sub(r"//(127\.0\.0\.1|localhost)\b", "//host.docker.internal", value) + for name in FORWARDED_ENV + if (value := os.environ.get(name)) + } + + # Claude Code needs a key to start, so it gets a placeholder. The sandbox's + # proxy swaps in the real key on requests to Anthropic. + return { + "ANTHROPIC_API_KEY": PROXY_MANAGED_KEY, + **forwarded, + **request_env, + } + + +def tar(directory: str) -> bytes: + archive = io.BytesIO() + with tarfile.open(fileobj=archive, mode="w") as tar: + tar.add(directory, arcname=".") + + return archive.getvalue() + + +class SandboxConnector: + def __init__(self, uploads: Mapping[str, str] | None = None) -> None: + """Run Claude Code in a Docker Sandbox. + + Args: + uploads: Directories on the Worker to copy into each sandbox, by + their path in the sandbox. + """ + self.uploads = dict(uploads or {}) + + async def connect(self, request: ConnectActivityInput) -> ClaudeCodeProcess: + if request.user is not None: + raise CLIConnectionError("Claude Code runs as the sandbox's agent user") + + # One sandbox per Workflow, outliving the Worker and the Workflow, so a retried launch, or + # a client that resumes the session after its Worker went away, finds the + # session and the workspace in it. + # sbx keeps sandboxes on the machine that runs it, so a resume only finds + # it on the same machine. With Workers on several machines, pass + # activity_options={"connect": {"task_queue": ...}} with a task queue + # only one machine polls. + name = sandbox_name() + if name not in (await sbx("ls", "-q")).decode().split(): + await sbx("create", "--name", name, "--no-progress", "claude") + + # "**" also reaches every port on the Worker's host through + # host.docker.internal. Narrow it if that host runs services the agent + # shouldn't reach. + await sbx("policy", "allow", "network", "--sandbox", name, "**") + + # Register the key as the sandbox's "anthropic" service secret. The + # sandbox's egress proxy injects it into outbound requests to Anthropic's + # API hosts, so the key never enters the sandbox. Requests to any other + # host, such as a custom ANTHROPIC_BASE_URL, go out without it. + if key := os.environ.get("ANTHROPIC_API_KEY"): + await sbx( + "secret", + "set", + "anthropic", + "--sandbox", + name, + "--force", + input=key.encode(), + ) + + for path, directory in self.uploads.items(): + script = 'mkdir -p "$1" && tar -x -C "$1"' + await sbx( + "exec", "-i", name, "sh", "-c", script, "sh", path, input=tar(directory) + ) + + env_flags = [ + flag + for key, value in sandbox_env(request.env).items() + for flag in ("-e", f"{key}={value}") + ] + + process = await SubprocessConnector().connect( + ConnectActivityInput( + cli_path="sbx", + args=[ + "exec", + "-i", + *env_flags, + "-w", + request.cwd or WORKSPACE, + name, + request.cli_path or "claude", + *request.args, + ], + env={}, + max_buffer_size=request.max_buffer_size, + ) + ) + return SandboxProcess(name, process) + + +class SandboxProcess: + """Claude Code in a sandbox, reached through ``sbx exec``.""" + + def __init__(self, name: str, process: ClaudeCodeProcess) -> None: + self.name = name + self.process = process + + async def write(self, data: str) -> None: + await self.process.write(data) + + async def end_input(self) -> None: + await self.process.end_input() + + async def read(self) -> list[Output]: + return await self.process.read() + + async def close(self) -> None: + await self.process.close() + + async def terminate(self) -> None: + # The Worker connected to this sandbox is shutting down. Stop Claude Code + # as SIGTERM does, so a resumed session continues its turn. + await sbx("exec", self.name, "pkill", "-TERM", "-x", "claude") + + +@dataclass +class SavedTurn: + started: bool # Claude Code saved the prompt + result: str | None # the final answer, once the turn finished + + +@activity.defn +async def find_saved_turn(session_id: str) -> SavedTurn: + """Read how far the session's turn got from its transcript in the sandbox.""" + script = 'cat "$HOME"/.claude/projects/*/"$1".jsonl 2>/dev/null || true' + transcript = await sbx("exec", sandbox_name(), "sh", "-c", script, "sh", session_id) + + # Give the transcript to the Claude Agent SDK through an in-memory store. + # Nothing is written to the sandbox. + store = InMemorySessionStore() + await store.append( + { + "project_key": project_key_for_directory(WORKSPACE), + "session_id": session_id, + }, + [json.loads(line) for line in transcript.splitlines()], + ) + messages = await get_session_messages_from_store(store, session_id, WORKSPACE) + if not messages: + return SavedTurn(started=False, result=None) + + last = messages[-1].message + result = ( + "".join(block["text"] for block in last["content"] if block["type"] == "text") + if last.get("stop_reason") == "end_turn" + else None + ) + return SavedTurn(started=True, result=result) diff --git a/claude_agent_sdk_plugin/plugin_with_custom_transport/haiku-writer/.claude-plugin/plugin.json b/claude_agent_sdk_plugin/plugin_with_custom_transport/haiku-writer/.claude-plugin/plugin.json new file mode 100644 index 00000000..83e9de8f --- /dev/null +++ b/claude_agent_sdk_plugin/plugin_with_custom_transport/haiku-writer/.claude-plugin/plugin.json @@ -0,0 +1,5 @@ +{ + "name": "haiku-writer", + "description": "Answers in haiku form, with a tool that checks syllable counts", + "version": "1.0.0" +} diff --git a/claude_agent_sdk_plugin/plugin_with_custom_transport/haiku-writer/.mcp.json b/claude_agent_sdk_plugin/plugin_with_custom_transport/haiku-writer/.mcp.json new file mode 100644 index 00000000..d8e889bd --- /dev/null +++ b/claude_agent_sdk_plugin/plugin_with_custom_transport/haiku-writer/.mcp.json @@ -0,0 +1,8 @@ +{ + "mcpServers": { + "syllables": { + "command": "python3", + "args": ["${CLAUDE_PLUGIN_ROOT}/servers/syllables.py"] + } + } +} diff --git a/claude_agent_sdk_plugin/plugin_with_custom_transport/haiku-writer/servers/syllables.py b/claude_agent_sdk_plugin/plugin_with_custom_transport/haiku-writer/servers/syllables.py new file mode 100644 index 00000000..c39d4a3b --- /dev/null +++ b/claude_agent_sdk_plugin/plugin_with_custom_transport/haiku-writer/servers/syllables.py @@ -0,0 +1,58 @@ +"""A stdio MCP server that counts syllables. + +It uses only the standard library, so it runs on any python3, including the +one in the sandbox, without installing anything. +""" + +import json +import re +import sys + +TOOL = { + "name": "count_syllables", + "description": "Count the syllables in each line of a poem.", + "inputSchema": { + "type": "object", + "properties": {"text": {"type": "string"}}, + "required": ["text"], + }, +} + + +def syllables(word: str) -> int: + count = len(re.findall(r"[aeiouy]+", word)) + if word.endswith("e") and not word.endswith("le") and count > 1: + count -= 1 + return max(count, 1) + + +def count_syllables(text: str) -> str: + return "\n".join( + f"{sum(syllables(word) for word in re.findall('[a-z]+', line.lower()))}: {line}" + for line in text.splitlines() + if line.strip() + ) + + +def handle(method: str, params: dict) -> dict: + if method == "initialize": + return { + "protocolVersion": params["protocolVersion"], + "capabilities": {"tools": {}}, + "serverInfo": {"name": "syllables", "version": "1.0.0"}, + } + if method == "tools/list": + return {"tools": [TOOL]} + if method == "tools/call": + text = count_syllables(params["arguments"]["text"]) + return {"content": [{"type": "text", "text": text}]} + return {} + + +for line in sys.stdin: + request = json.loads(line) + # Requests have an id. Notifications don't, and get no response. + if "id" in request: + result = handle(request["method"], request.get("params", {})) + response = {"jsonrpc": "2.0", "id": request["id"], "result": result} + print(json.dumps(response), flush=True) diff --git a/claude_agent_sdk_plugin/plugin_with_custom_transport/haiku-writer/skills/haiku/SKILL.md b/claude_agent_sdk_plugin/plugin_with_custom_transport/haiku-writer/skills/haiku/SKILL.md new file mode 100644 index 00000000..9924635d --- /dev/null +++ b/claude_agent_sdk_plugin/plugin_with_custom_transport/haiku-writer/skills/haiku/SKILL.md @@ -0,0 +1,12 @@ +--- +name: haiku +description: Write the answer as a haiku. Use when the user asks for a haiku or a poem. +--- + +Answer with a single haiku: three lines of five, seven, and five syllables. + +Before answering, check the haiku with the `count_syllables` tool and rewrite any line whose count is off. + +After the haiku, add a blank line and then this exact signature line: + +~ haiku-writer diff --git a/claude_agent_sdk_plugin/plugin_with_custom_transport/run_worker.py b/claude_agent_sdk_plugin/plugin_with_custom_transport/run_worker.py new file mode 100644 index 00000000..f1268202 --- /dev/null +++ b/claude_agent_sdk_plugin/plugin_with_custom_transport/run_worker.py @@ -0,0 +1,58 @@ +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.offline import Answer, offline_api_without_key, tool_use +from claude_agent_sdk_plugin.plugin_with_custom_transport.connector import ( + SandboxConnector, + find_saved_turn, +) +from claude_agent_sdk_plugin.plugin_with_custom_transport.workflow import ( + COUNT_SYLLABLES, + PLUGIN_DIR, + PluginWithCustomTransportWorkflow, +) + +# The plugin on the Worker. The sandbox starts with a clean filesystem, so the +# connector copies the whole directory into each sandbox before Claude Code starts. +LOCAL_PLUGIN_DIR = os.path.join(os.path.dirname(__file__), "haiku-writer") + +HAIKU = "Workers fall and rise\nhistory remembers all\nthe work carries on" + +OFFLINE_ANSWERS: dict[str, Answer] = { + "tool_result:toolu_count": f"{HAIKU}\n\n~ haiku-writer", + "tool_result:toolu_skill": tool_use("toolu_count", COUNT_SYLLABLES, text=HAIKU), + "haiku": tool_use("toolu_skill", "Skill", skill="haiku-writer:haiku"), +} + + +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-plugin-with-custom-transport", + workflows=[PluginWithCustomTransportWorkflow], + activities=[find_saved_turn], + plugins=[ + ClaudeAgentSDKPlugin( + connector=SandboxConnector(uploads={PLUGIN_DIR: LOCAL_PLUGIN_DIR}) + ) + ], + ): + print(f"Worker started. To stop it gracefully: kill {os.getpid()}") + await stop.wait() + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/claude_agent_sdk_plugin/plugin_with_custom_transport/workflow.py b/claude_agent_sdk_plugin/plugin_with_custom_transport/workflow.py new file mode 100644 index 00000000..9d7cd1a8 --- /dev/null +++ b/claude_agent_sdk_plugin/plugin_with_custom_transport/workflow.py @@ -0,0 +1,82 @@ +from datetime import timedelta + +from claude_agent_sdk import ( + ClaudeAgentOptions, + ClaudeSDKClient, + CLIConnectionError, + ResultMessage, +) +from temporalio import workflow +from temporalio.claude_agent_sdk import TemporalTransport + +from claude_agent_sdk_plugin.plugin_with_custom_transport.connector import ( + SavedTurn, + find_saved_turn, +) + +# Where the Worker's connector puts the plugin, in Claude Code's sandbox. The +# path is recorded in history, as part of the connect Activity's input. +PLUGIN_DIR = "/home/agent/plugins/haiku-writer" + +# Claude Code names plugin MCP tools mcp__plugin____. +COUNT_SYLLABLES = "mcp__plugin_haiku-writer_syllables__count_syllables" + + +@workflow.defn +class PluginWithCustomTransportWorkflow: + @workflow.run + async def run(self, prompt: str) -> str: + return await self.ask(prompt) + + def options(self, session_id: str, *, resume: bool) -> ClaudeAgentOptions: + """Options that start the session, or resume it.""" + return ClaudeAgentOptions( + plugins=[{"type": "local", "path": PLUGIN_DIR}], + # Limit Claude Code's built-in tools to just Skill. + tools=["Skill"], + # Claude Code runs skills without asking, but MCP tools need allowing. + allowed_tools=[COUNT_SYLLABLES], + setting_sources=[], + env={"CLAUDE_CODE_RESUME_INTERRUPTED_TURN": "1"}, + session_id=None if resume else session_id, + resume=session_id if resume else None, + ) + + async def ask(self, prompt: str) -> str: + """Answer the prompt, finishing the turn if the connection is lost.""" + session_id = str(workflow.uuid4()) + turn = SavedTurn(started=False, result=None) + + while True: + try: + options = self.options(session_id, resume=turn.started) + transport = TemporalTransport(options) + async with ClaudeSDKClient( + options=options, transport=transport + ) as client: + # Send the prompt until Claude Code saves it. After that, a + # resume finishes the turn. + if not turn.started: + await client.query(prompt) + + async for message in client.receive_response(): + if isinstance(message, ResultMessage): + return message.result or "\n".join(message.errors or []) + + raise RuntimeError("Claude Code ended the turn without a result") + + except CLIConnectionError: + # Claude Code may have finished the turn after the Worker's last + # read. If so, its transcript in the sandbox has the answer. + turn = await workflow.execute_activity( + find_saved_turn, + session_id, + start_to_close_timeout=timedelta(minutes=1), + ) + if turn.result is not None: + return turn.result + + # Claude Code can't continue a session it didn't receive the + # prompt for, so start a new one. + if not turn.started: + session_id = str(workflow.uuid4()) diff --git a/claude_agent_sdk_plugin/streaming/README.md b/claude_agent_sdk_plugin/streaming/README.md new file mode 100644 index 00000000..cf74f911 --- /dev/null +++ b/claude_agent_sdk_plugin/streaming/README.md @@ -0,0 +1,45 @@ +# Streaming + +Watch Claude's reply appear token by token through a [Workflow Stream](https://docs.temporal.io/develop/python/workflows/workflow-streams) (experimental). + +## What this shows + +- **Live output, durable Workflow.** Clients see the text as Claude writes it. The Workflow itself still works with complete messages. +- **Watching is optional.** The Workflow doesn't wait for anyone to subscribe, and returns the same result either way. +## 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/streaming/run_worker.py +``` + +The Worker runs offline: everything runs for real except the model, which answers in five scripted chunks. To use the real model, put `ANTHROPIC_API_KEY=...` in front of the command. + +### Start the Workflow with a live subscriber + +```bash +uv run --group claude-agent-sdk claude_agent_sdk_plugin/streaming/run_workflow.py +``` + +The text prints as it arrives, then `Final result:` repeats it. Offline that's `One. Two. Three. Four. Five.`, and the chunks reach you through Claude Code and the Workflow Stream, the same path live tokens take. Pass your own prompt as arguments. + +The Temporal CLI can't subscribe to a Workflow Stream, which is why this case has its own script. + +### Start the Workflow with no subscriber + +```bash +temporal workflow execute \ + --type StreamingWorkflow \ + --task-queue claude-agent-sdk-streaming \ + --workflow-id claude-agent-sdk-streaming-nosub \ + --input '"Count from one to five, one number per sentence."' +``` + +The Workflow completes with the same result. diff --git a/claude_agent_sdk_plugin/streaming/__init__.py b/claude_agent_sdk_plugin/streaming/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/claude_agent_sdk_plugin/streaming/run_worker.py b/claude_agent_sdk_plugin/streaming/run_worker.py new file mode 100644 index 00000000..2e7eea4e --- /dev/null +++ b/claude_agent_sdk_plugin/streaming/run_worker.py @@ -0,0 +1,37 @@ +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.offline import offline_api_without_key +from claude_agent_sdk_plugin.streaming.workflow import StreamingWorkflow + +OFFLINE_ANSWERS = { + "Count": ["One. ", "Two. ", "Three. ", "Four. ", "Five."], +} + + +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-streaming", + workflows=[StreamingWorkflow], + plugins=[ClaudeAgentSDKPlugin()], + ): + print(f"Worker started. To stop it gracefully: kill {os.getpid()}") + await stop.wait() + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/claude_agent_sdk_plugin/streaming/run_workflow.py b/claude_agent_sdk_plugin/streaming/run_workflow.py new file mode 100644 index 00000000..a61b3c75 --- /dev/null +++ b/claude_agent_sdk_plugin/streaming/run_workflow.py @@ -0,0 +1,34 @@ +import asyncio +import os +import sys + +from temporalio.client import Client +from temporalio.contrib.workflow_streams import WorkflowStreamClient + +from claude_agent_sdk_plugin.streaming.workflow import StreamingWorkflow + +DEFAULT_PROMPT = "Count from one to five, one number per sentence." + + +async def main() -> None: + client = await Client.connect(os.environ.get("TEMPORAL_ADDRESS", "localhost:7233")) + workflow_id = "claude-agent-sdk-streaming" + + handle = await client.start_workflow( + StreamingWorkflow.run, + " ".join(sys.argv[1:]) or DEFAULT_PROMPT, + id=workflow_id, + task_queue="claude-agent-sdk-streaming", + ) + + stream = WorkflowStreamClient.create(client, workflow_id) + async for item in stream.subscribe(["events"]): + delta = item.data["event"].get("delta", {}) + if delta.get("type") == "text_delta": + print(delta["text"], end="", flush=True) + + print(f"\nFinal result: {await handle.result()}") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/claude_agent_sdk_plugin/streaming/workflow.py b/claude_agent_sdk_plugin/streaming/workflow.py new file mode 100644 index 00000000..e4c0114c --- /dev/null +++ b/claude_agent_sdk_plugin/streaming/workflow.py @@ -0,0 +1,68 @@ +from datetime import timedelta + +from claude_agent_sdk import ( + ClaudeAgentOptions, + ClaudeSDKClient, + CLIConnectionError, + ResultMessage, + SystemMessage, +) +from temporalio import workflow +from temporalio.claude_agent_sdk import TemporalTransport +from temporalio.contrib.workflow_streams import WorkflowStream + + +@workflow.defn +class StreamingWorkflow: + @workflow.init + def __init__(self, prompt: str) -> None: + self.events = WorkflowStream().topic("events") + + @workflow.run + async def run(self, prompt: str) -> str: + result = await self.ask(prompt) + + # Give a subscriber's last poll time to read the final events before the + # Workflow, and its in-memory stream, completes. + await workflow.sleep(timedelta(milliseconds=500)) + return result + + def options(self, *, resume: str | None = None) -> ClaudeAgentOptions: + """Options that start a new session, or continue the one given by resume.""" + return ClaudeAgentOptions( + include_partial_messages=True, + setting_sources=[], + env={"CLAUDE_CODE_RESUME_INTERRUPTED_TURN": "1"}, + resume=resume, + ) + + async def ask(self, prompt: str) -> str: + """Answer the prompt, resuming the session if the connection is lost.""" + session_id = None # Claude Code's, once it has the prompt + + while True: + try: + options = self.options(resume=session_id) + transport = TemporalTransport(options, stream=self.events) + async with ClaudeSDKClient( + options=options, transport=transport + ) as client: + # Keep sending the prompt until Claude Code reports a session, which + # means it received it. After that, a resume finishes the turn. + if session_id is None: + await client.query(prompt) + + async for message in client.receive_response(): + if ( + isinstance(message, SystemMessage) + and message.subtype == "init" + ): + session_id = message.data["session_id"] + elif isinstance(message, ResultMessage): + return message.result or "" + + raise RuntimeError("Claude Code ended the turn without a result") + + except CLIConnectionError: + # Resume the session if Claude Code started it, or start again. + pass diff --git a/claude_agent_sdk_plugin/structured_output/README.md b/claude_agent_sdk_plugin/structured_output/README.md new file mode 100644 index 00000000..869306d5 --- /dev/null +++ b/claude_agent_sdk_plugin/structured_output/README.md @@ -0,0 +1,45 @@ +# Structured Output + +Constrain Claude Code's final answer to a JSON schema and validate it as a typed Pydantic model at the Workflow boundary. + +## What this shows + +- **Typed results from an agent.** You define a Pydantic model, and the Workflow returns an instance of it instead of free text. +- **Claude fixes its own mistakes.** Claude Code checks the answer against the schema and sends errors back to the model, so it can try again. +- **A clear failure.** If no valid answer ever comes, the Workflow fails instead of returning bad data. + +## 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/structured_output/run_worker.py +``` + +The Worker runs offline: everything runs for real except the model, which answers with a fixed ticket. To use the real model, put `ANTHROPIC_API_KEY=...` in front of the command. + +Triage a message from a third terminal: + +```bash +temporal workflow execute \ + --type TriageTicketWorkflow \ + --task-queue claude-agent-sdk-structured-output \ + --workflow-id structured-output \ + --input '"The app crashes every time I open the settings page."' +``` + +## Expected behavior + +The result is the typed ticket as JSON, for example `{"category": "bug", "priority": 2, "summary": "..."}`. Offline it is always: + +```json +{"category": "bug", "priority": 2, "summary": "The settings page crashes the app when opened."} +``` + +Offline, the ticket's values are scripted, but Claude Code still checks them against the schema and the Workflow still validates them as a `SupportTicket`. diff --git a/claude_agent_sdk_plugin/structured_output/__init__.py b/claude_agent_sdk_plugin/structured_output/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/claude_agent_sdk_plugin/structured_output/run_worker.py b/claude_agent_sdk_plugin/structured_output/run_worker.py new file mode 100644 index 00000000..aae40c77 --- /dev/null +++ b/claude_agent_sdk_plugin/structured_output/run_worker.py @@ -0,0 +1,48 @@ +import asyncio +import os +import signal + +from temporalio.claude_agent_sdk import ClaudeAgentSDKPlugin +from temporalio.client import Client +from temporalio.contrib.pydantic import pydantic_data_converter +from temporalio.worker import Worker + +from claude_agent_sdk_plugin.offline import offline_api_without_key, tool_use +from claude_agent_sdk_plugin.structured_output.workflow import TriageTicketWorkflow + +OFFLINE_ANSWERS = { + "Triage this customer message": tool_use( + "toolu_ticket", + "StructuredOutput", + category="bug", + priority=2, + summary="The settings page crashes the app when opened.", + ) +} + + +async def main() -> None: + # The Pydantic converter lets the Workflow return a SupportTicket. + client = await Client.connect( + os.environ.get("TEMPORAL_ADDRESS", "localhost:7233"), + data_converter=pydantic_data_converter, + ) + + # 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-structured-output", + workflows=[TriageTicketWorkflow], + plugins=[ClaudeAgentSDKPlugin()], + ): + print(f"Worker started. To stop it gracefully: kill {os.getpid()}") + await stop.wait() + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/claude_agent_sdk_plugin/structured_output/workflow.py b/claude_agent_sdk_plugin/structured_output/workflow.py new file mode 100644 index 00000000..30149a97 --- /dev/null +++ b/claude_agent_sdk_plugin/structured_output/workflow.py @@ -0,0 +1,75 @@ +from typing import Literal + +from claude_agent_sdk import ( + ClaudeAgentOptions, + ClaudeSDKClient, + CLIConnectionError, + ResultMessage, + SystemMessage, +) +from pydantic import BaseModel, Field +from temporalio import workflow +from temporalio.claude_agent_sdk import TemporalTransport +from temporalio.exceptions import ApplicationError + + +class SupportTicket(BaseModel): + category: Literal["billing", "bug", "feature_request"] + priority: int = Field(ge=1, le=5) + summary: str + + +@workflow.defn +class TriageTicketWorkflow: + @workflow.run + async def run(self, text: str) -> SupportTicket: + prompt = f"Triage this customer message:\n{text}" + return await self.ask(prompt) + + def options(self, *, resume: str | None = None) -> ClaudeAgentOptions: + """Options that start a new session, or continue the one given by resume.""" + return ClaudeAgentOptions( + output_format={ + "type": "json_schema", + "schema": SupportTicket.model_json_schema(), + }, + setting_sources=[], + env={"CLAUDE_CODE_RESUME_INTERRUPTED_TURN": "1"}, + resume=resume, + ) + + async def ask(self, prompt: str) -> SupportTicket: + """Answer the prompt, resuming the session if the connection is lost.""" + session_id = None # Claude Code's, once it has the prompt + + while True: + try: + options = self.options(resume=session_id) + transport = TemporalTransport(options) + async with ClaudeSDKClient( + options=options, transport=transport + ) as client: + # Keep sending the prompt until Claude Code reports a session, which + # means it received it. After that, a resume finishes the turn. + if session_id is None: + await client.query(prompt) + + async for message in client.receive_response(): + if ( + isinstance(message, SystemMessage) + and message.subtype == "init" + ): + session_id = message.data["session_id"] + elif isinstance(message, ResultMessage): + if message.structured_output is not None: + return SupportTicket.model_validate( + message.structured_output + ) + + raise ApplicationError( + "Claude Code returned no structured output", non_retryable=True + ) + + except CLIConnectionError: + # Resume the session if Claude Code started it, or start again. + pass diff --git a/claude_agent_sdk_plugin/tools/README.md b/claude_agent_sdk_plugin/tools/README.md new file mode 100644 index 00000000..a91d78b8 --- /dev/null +++ b/claude_agent_sdk_plugin/tools/README.md @@ -0,0 +1,43 @@ +# Tools + +Claude calls two custom Claude Agent SDK `@tool` functions that run in the Workflow. `count_letter` is plain Workflow code, and `get_weather` calls the `fetch_weather` Activity for its I/O. + +## What this shows + +- **Tools are plain `@tool` functions.** You write them as usual with the Claude Agent SDK, and they run in the Workflow. + +## 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/tools/run_worker.py +``` + +The Worker runs offline: everything runs for real except the model, which is scripted to call both tools. To use the real model, put `ANTHROPIC_API_KEY=...` in front of the command. + +### Start the Workflow + +```bash +temporal workflow execute \ + --type ToolsWorkflow \ + --task-queue claude-agent-sdk-tools \ + --workflow-id claude-agent-sdk-tools \ + --input '"Count the r letters in strawberry, then get the weather in Paris."' +``` + +Offline, the final reply is scripted: + +```text +"There are 3 r's in strawberry. It is 21 C and sunny in Paris." +``` + +## See the boundary + +The history is the real proof, live or offline. In the Temporal UI (`http://localhost:8233`) or with `temporal workflow show --workflow-id claude-agent-sdk-tools`, the weather call is one `fetch_weather` Activity. `count_letter` has none; it ran as Workflow code. The other Activities are the plugin's `claude-agent-sdk-*` Activities, which reach Claude Code. diff --git a/claude_agent_sdk_plugin/tools/__init__.py b/claude_agent_sdk_plugin/tools/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/claude_agent_sdk_plugin/tools/run_worker.py b/claude_agent_sdk_plugin/tools/run_worker.py new file mode 100644 index 00000000..cd7dc870 --- /dev/null +++ b/claude_agent_sdk_plugin/tools/run_worker.py @@ -0,0 +1,50 @@ +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.offline import ( + Answer, + offline_api_without_key, + tool_use, +) +from claude_agent_sdk_plugin.tools.workflow import ToolsWorkflow, fetch_weather + +OFFLINE_ANSWERS: dict[str, Answer] = { + "strawberry": tool_use( + "toolu_count", "mcp__demo__count_letter", word="strawberry", letter="r" + ), + "tool_result:toolu_count": tool_use( + "toolu_weather", "mcp__demo__get_weather", city="Paris" + ), + "tool_result:toolu_weather": ( + "There are 3 r's in strawberry. It is 21 C and sunny in Paris." + ), +} + + +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-tools", + workflows=[ToolsWorkflow], + activities=[fetch_weather], + plugins=[ClaudeAgentSDKPlugin()], + ): + print(f"Worker started. To stop it gracefully: kill {os.getpid()}") + await stop.wait() + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/claude_agent_sdk_plugin/tools/workflow.py b/claude_agent_sdk_plugin/tools/workflow.py new file mode 100644 index 00000000..5f19f339 --- /dev/null +++ b/claude_agent_sdk_plugin/tools/workflow.py @@ -0,0 +1,88 @@ +from datetime import timedelta +from typing import Any + +from claude_agent_sdk import ( + ClaudeAgentOptions, + ClaudeSDKClient, + CLIConnectionError, + ResultMessage, + SystemMessage, + create_sdk_mcp_server, + tool, +) +from temporalio import activity, workflow +from temporalio.claude_agent_sdk import TemporalTransport + + +@activity.defn +async def fetch_weather(city: str) -> str: + return f"It is 21 C and sunny in {city}." + + +@tool( + "count_letter", + "Count how often a letter appears in a word", + {"word": str, "letter": str}, +) +async def count_letter(args: dict[str, Any]) -> dict[str, Any]: + count = args["word"].lower().count(args["letter"].lower()) + return {"content": [{"type": "text", "text": str(count)}]} + + +@tool("get_weather", "Get the current weather in a city", {"city": str}) +async def get_weather(args: dict[str, Any]) -> dict[str, Any]: + weather = await workflow.execute_activity( + fetch_weather, args["city"], start_to_close_timeout=timedelta(seconds=30) + ) + return {"content": [{"type": "text", "text": weather}]} + + +@workflow.defn +class ToolsWorkflow: + @workflow.run + async def run(self, prompt: str) -> str: + return await self.ask(prompt) + + def options(self, *, resume: str | None = None) -> ClaudeAgentOptions: + """Options that start a new session, or continue the one given by resume.""" + return ClaudeAgentOptions( + tools=[], + mcp_servers={ + "demo": create_sdk_mcp_server("demo", tools=[count_letter, get_weather]) + }, + allowed_tools=["mcp__demo__count_letter", "mcp__demo__get_weather"], + setting_sources=[], + env={"CLAUDE_CODE_RESUME_INTERRUPTED_TURN": "1"}, + resume=resume, + ) + + async def ask(self, prompt: str) -> str: + """Answer the prompt, resuming the session if the connection is lost.""" + session_id = None # Claude Code's, once it has the prompt + + while True: + try: + options = self.options(resume=session_id) + transport = TemporalTransport(options) + async with ClaudeSDKClient( + options=options, transport=transport + ) as client: + # Keep sending the prompt until Claude Code reports a session, which + # means it received it. After that, a resume finishes the turn. + if session_id is None: + await client.query(prompt) + + async for message in client.receive_response(): + if ( + isinstance(message, SystemMessage) + and message.subtype == "init" + ): + session_id = message.data["session_id"] + elif isinstance(message, ResultMessage): + return message.result or "\n".join(message.errors or []) + + raise RuntimeError("Claude Code ended the turn without a result") + + except CLIConnectionError: + # Resume the session if Claude Code started it, or start again. + pass diff --git a/pyproject.toml b/pyproject.toml index 9fff6239..755409b1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -28,6 +28,11 @@ dev = [ "poethepoet>=0.36.0", ] bedrock = ["boto3>=1.34.92,<2"] +claude-agent-sdk = [ + "temporalio-claude-agent-sdk>=0.0.0", + "temporalio[opentelemetry,pydantic]>=1.35.0,<2", + "opentelemetry-exporter-otlp-proto-grpc>=1.30.0,<2", +] deepagents = [ "temporalio-deepagents>=0.0.1 ; python_version >= '3.11'", "temporalio-langsmith>=0.1.0 ; python_version >= '3.11'", @@ -101,6 +106,9 @@ cloud-export-to-parquet = [ "pyarrow>=19.0.1", ] +[tool.uv.sources] +temporalio-claude-agent-sdk = { git = "https://github.com/temporalio/ai-integrations", subdirectory = "python/claude_agent_sdk", rev = "177fb623f874b85db519401c4be496eeb19da8f5" } + [tool.uv] constraint-dependencies = [ # yarl 1.24.0 was published without an sdist and only has cp310 wheels, so it cannot diff --git a/tests/claude_agent_sdk_plugin/__init__.py b/tests/claude_agent_sdk_plugin/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/claude_agent_sdk_plugin/agent_chat_test.py b/tests/claude_agent_sdk_plugin/agent_chat_test.py new file mode 100644 index 00000000..551164dd --- /dev/null +++ b/tests/claude_agent_sdk_plugin/agent_chat_test.py @@ -0,0 +1,81 @@ +import json +import uuid + +import pytest +from temporalio.api.enums.v1 import EventType +from temporalio.claude_agent_sdk import ClaudeAgentSDKPlugin +from temporalio.client import Client, WorkflowContinuedAsNewError +from temporalio.worker import Replayer, Worker + +from claude_agent_sdk_plugin.agent_chat.workflow import ChatInput, ChatWorkflow +from claude_agent_sdk_plugin.offline import prompt_of, scripted_api + +ANSWERS = { + "Ada": "Nice to meet you, Ada.", + "Lovelace": "Ada Lovelace, noted.", + "my name": "Your name is Ada Lovelace.", +} + + +async def test_chat_resumes_claude_session_after_continue_as_new( + client: Client, +) -> None: + task_queue = f"claude-agent-sdk-agent-chat-{uuid.uuid4()}" + workflow_id = f"claude-agent-sdk-agent-chat-{uuid.uuid4()}" + with scripted_api(ANSWERS) as api: + async with Worker( + client, + task_queue=task_queue, + workflows=[ChatWorkflow], + plugins=[ClaudeAgentSDKPlugin()], + ): + first = await client.start_workflow( + ChatWorkflow.run, + ChatInput(turns_per_run=2), + id=workflow_id, + task_queue=task_queue, + ) + assert ( + await first.execute_update(ChatWorkflow.send_message, "I'm Ada.") + == "Nice to meet you, Ada." + ) + assert ( + await first.execute_update( + ChatWorkflow.send_message, "My surname is Lovelace." + ) + == "Ada Lovelace, noted." + ) + with pytest.raises(WorkflowContinuedAsNewError): + await first.result(follow_runs=False) + + second = client.get_workflow_handle(workflow_id) + assert ( + await second.execute_update( + ChatWorkflow.send_message, "What is my name?" + ) + == "Your name is Ada Lovelace." + ) + await second.signal(ChatWorkflow.end_conversation) + await second.result() + + after_resume = next(r for r in api.requests if "What is my name?" in prompt_of(r)) + conversation = json.dumps(after_resume["messages"]) + assert "I'm Ada." in conversation + assert "Ada Lovelace, noted." in conversation + + first_run = client.get_workflow_handle(workflow_id, run_id=first.result_run_id) + first_history = await first_run.fetch_history() + continued = first_history.events[-1] + assert ( + continued.event_type == EventType.EVENT_TYPE_WORKFLOW_EXECUTION_CONTINUED_AS_NEW + ) + [carried] = await client.data_converter.decode( + continued.workflow_execution_continued_as_new_event_attributes.input.payloads, + [ChatInput], + ) + assert carried.session_id + assert carried.turns_per_run == 2 + + replayer = Replayer(workflows=[ChatWorkflow], plugins=[ClaudeAgentSDKPlugin()]) + await replayer.replay_workflow(first_history) + await replayer.replay_workflow(await second.fetch_history()) diff --git a/tests/claude_agent_sdk_plugin/conftest.py b/tests/claude_agent_sdk_plugin/conftest.py new file mode 100644 index 00000000..44e81424 --- /dev/null +++ b/tests/claude_agent_sdk_plugin/conftest.py @@ -0,0 +1,30 @@ +import sys +from pathlib import Path + +import pytest +import pytest_asyncio +from temporalio.client import Client +from temporalio.testing import WorkflowEnvironment + +collect_ignore_glob: list[str] = [] + +if sys.platform == "win32": + collect_ignore_glob = ["*_test.py"] +else: + try: + import temporalio.claude_agent_sdk # noqa: F401 + except ImportError: + collect_ignore_glob = ["*_test.py"] + + +@pytest_asyncio.fixture +async def client(env: WorkflowEnvironment) -> Client: + if env.supports_time_skipping: + pytest.skip("Claude Code runs in real time") + return env.client + + +@pytest.fixture(autouse=True) +def workspace(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + monkeypatch.chdir(tmp_path) + return tmp_path diff --git a/tests/claude_agent_sdk_plugin/fork_session_test.py b/tests/claude_agent_sdk_plugin/fork_session_test.py new file mode 100644 index 00000000..1ff73d6a --- /dev/null +++ b/tests/claude_agent_sdk_plugin/fork_session_test.py @@ -0,0 +1,49 @@ +import json +import uuid +from typing import Any + +from temporalio.claude_agent_sdk import ClaudeAgentSDKPlugin +from temporalio.client import Client +from temporalio.worker import Replayer, Worker + +from claude_agent_sdk_plugin.fork_session.run_worker import OFFLINE_ANSWERS +from claude_agent_sdk_plugin.fork_session.workflow import ForkSessionWorkflow +from claude_agent_sdk_plugin.offline import prompt_of, scripted_api + +PROMPT = "Plan a weekend in Lisbon" +FOLLOW_UPS = ["Redo it on a tight budget.", "Redo it with no budget limit."] + + +async def test_each_follow_up_runs_in_its_own_fork_of_the_conversation( + client: Client, +) -> None: + task_queue = f"claude-agent-sdk-fork-session-{uuid.uuid4()}" + with scripted_api(OFFLINE_ANSWERS) as api: + async with Worker( + client, + task_queue=task_queue, + workflows=[ForkSessionWorkflow], + plugins=[ClaudeAgentSDKPlugin()], + ): + handle = await client.start_workflow( + ForkSessionWorkflow.run, + args=[PROMPT, FOLLOW_UPS], + id=f"fork-session-{uuid.uuid4()}", + task_queue=task_queue, + ) + budget, luxury = await handle.result() + + assert budget.answer == OFFLINE_ANSWERS["on a tight budget"] + assert luxury.answer == OFFLINE_ANSWERS["with no budget limit"] + assert budget.session_id != luxury.session_id + + # Each fork's request has the base conversation, and no other fork's turn. + conversations: dict[str, Any] = {prompt_of(r): r["messages"] for r in api.requests} + for branch, other in [(budget, luxury), (luxury, budget)]: + conversation = json.dumps(conversations[branch.prompt]) + assert PROMPT in conversation + assert other.prompt not in conversation + + await Replayer( + workflows=[ForkSessionWorkflow], plugins=[ClaudeAgentSDKPlugin()] + ).replay_workflow(await handle.fetch_history()) diff --git a/tests/claude_agent_sdk_plugin/human_in_the_loop_test.py b/tests/claude_agent_sdk_plugin/human_in_the_loop_test.py new file mode 100644 index 00000000..f4f6a019 --- /dev/null +++ b/tests/claude_agent_sdk_plugin/human_in_the_loop_test.py @@ -0,0 +1,143 @@ +import asyncio +import uuid +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager +from pathlib import Path + +import pytest +from temporalio.api.enums.v1 import EventType +from temporalio.claude_agent_sdk import ClaudeAgentSDKPlugin +from temporalio.client import ( + Client, + WorkflowFailureError, + WorkflowHandle, + WorkflowUpdateFailedError, +) +from temporalio.exceptions import CancelledError +from temporalio.worker import Replayer, Worker + +from claude_agent_sdk_plugin.human_in_the_loop.run_worker import ( + OFFLINE_ANSWERS, + TOOL_USE_ID, + tool_result, +) +from claude_agent_sdk_plugin.human_in_the_loop.workflow import ( + Decision, + HumanInTheLoopWorkflow, + ToolRequest, +) +from claude_agent_sdk_plugin.offline import ScriptedApi, scripted_api + +TASK = "Create hello.txt with a greeting." + + +@asynccontextmanager +async def started( + client: Client, +) -> AsyncIterator[tuple[WorkflowHandle[HumanInTheLoopWorkflow, str], ScriptedApi]]: + task_queue = f"claude-agent-sdk-human-in-the-loop-{uuid.uuid4()}" + with scripted_api(OFFLINE_ANSWERS) as api: + async with Worker( + client, + task_queue=task_queue, + workflows=[HumanInTheLoopWorkflow], + plugins=[ClaudeAgentSDKPlugin()], + ): + handle = await client.start_workflow( + HumanInTheLoopWorkflow.run, + TASK, + id=f"human-in-the-loop-{uuid.uuid4()}", + task_queue=task_queue, + ) + yield handle, api + + +async def wait_for_request( + handle: WorkflowHandle[HumanInTheLoopWorkflow, str], +) -> list[ToolRequest]: + for _ in range(300): + pending = await handle.query(HumanInTheLoopWorkflow.pending_tool_calls) + if pending: + return pending + await asyncio.sleep(0.2) + raise AssertionError("No tool call is waiting for a decision") + + +async def assert_replays(handle: WorkflowHandle[HumanInTheLoopWorkflow, str]) -> None: + await Replayer( + workflows=[HumanInTheLoopWorkflow], plugins=[ClaudeAgentSDKPlugin()] + ).replay_workflow(await handle.fetch_history()) + + +async def test_approved_tool_call_runs(client: Client, workspace: Path) -> None: + async with started(client) as (handle, _): + [request] = await wait_for_request(handle) + assert request == ToolRequest( + TOOL_USE_ID, + "Bash", + {"command": "echo 'Hello from Claude Code' > hello.txt"}, + ) + assert not (workspace / "hello.txt").exists() + + await handle.execute_update( + HumanInTheLoopWorkflow.decide, Decision(TOOL_USE_ID, approved=True) + ) + + assert await handle.result() == "Created hello.txt." + assert (workspace / "hello.txt").read_text() == "Hello from Claude Code\n" + await assert_replays(handle) + + +async def test_denied_tool_call_reaches_the_model( + client: Client, workspace: Path +) -> None: + async with started(client) as (handle, api): + await wait_for_request(handle) + + await handle.execute_update( + HumanInTheLoopWorkflow.decide, Decision(TOOL_USE_ID, approved=False) + ) + + assert await handle.result() == "Understood. I did not create hello.txt." + assert not (workspace / "hello.txt").exists() + result = tool_result(api.requests[-1]) + assert result["is_error"] is True + assert "A person denied this tool call" in str(result["content"]) + await assert_replays(handle) + + +async def test_decision_for_unknown_tool_call_is_rejected(client: Client) -> None: + async with started(client) as (handle, _): + await wait_for_request(handle) + + with pytest.raises(WorkflowUpdateFailedError) as rejected: + await handle.execute_update( + HumanInTheLoopWorkflow.decide, Decision("toolu_other", approved=True) + ) + assert "No pending tool call toolu_other" in str(rejected.value.cause) + + await handle.execute_update( + HumanInTheLoopWorkflow.decide, Decision(TOOL_USE_ID, approved=False) + ) + await handle.result() + + +async def test_cancel_while_waiting_ends_the_workflow( + client: Client, workspace: Path +) -> None: + async with started(client) as (handle, _): + await wait_for_request(handle) + + await handle.cancel() + + with pytest.raises(WorkflowFailureError) as failure: + await handle.result() + assert isinstance(failure.value.cause, CancelledError) + + history = await handle.fetch_history() + assert ( + history.events[-1].event_type + == EventType.EVENT_TYPE_WORKFLOW_EXECUTION_CANCELED + ) + assert not (workspace / "hello.txt").exists() + await assert_replays(handle) diff --git a/tests/claude_agent_sdk_plugin/multi_agent_test.py b/tests/claude_agent_sdk_plugin/multi_agent_test.py new file mode 100644 index 00000000..d790c7c1 --- /dev/null +++ b/tests/claude_agent_sdk_plugin/multi_agent_test.py @@ -0,0 +1,54 @@ +import uuid +from typing import Any + +from temporalio.claude_agent_sdk import ClaudeAgentSDKPlugin +from temporalio.client import Client +from temporalio.worker import Replayer, Worker + +from claude_agent_sdk_plugin.multi_agent.run_worker import ( + OFFLINE_ANSWERS, + BRIEFING, + RESEARCH_NOTES, +) +from claude_agent_sdk_plugin.multi_agent.workflow import MultiAgentWorkflow +from claude_agent_sdk_plugin.offline import scripted_api + + +def system_of(request: dict[str, Any]) -> str: + return " ".join(block["text"] for block in request["system"]) + + +async def test_coordinator_delegates_to_both_subagents(client: Client) -> None: + with scripted_api(OFFLINE_ANSWERS) as api: + async with Worker( + client, + task_queue=f"claude-agent-sdk-multi-agent-{uuid.uuid4()}", + workflows=[MultiAgentWorkflow], + plugins=[ClaudeAgentSDKPlugin()], + ) as worker: + handle = await client.start_workflow( + MultiAgentWorkflow.run, + "durable execution", + id=f"claude-agent-sdk-multi-agent-{uuid.uuid4()}", + task_queue=worker.task_queue, + ) + briefing = await handle.result() + + assert briefing.result == f"Final briefing: {BRIEFING}" + assert briefing.messages_by_agent["researcher"] == [RESEARCH_NOTES] + assert briefing.messages_by_agent["writer"] == [BRIEFING] + assert briefing.messages_by_agent["coordinator"][-1] == briefing.result + + assert api.asked == [ + "short briefing on this topic", + "Collect three key facts", + "tool_result:toolu_research", + "Turn these notes", + "tool_result:toolu_write", + ] + assert len([r for r in api.requests if "You are a researcher" in system_of(r)]) == 1 + assert len([r for r in api.requests if "You are a writer" in system_of(r)]) == 1 + + await Replayer( + workflows=[MultiAgentWorkflow], plugins=[ClaudeAgentSDKPlugin()] + ).replay_workflow(await handle.fetch_history()) diff --git a/tests/claude_agent_sdk_plugin/open_telemetry_test.py b/tests/claude_agent_sdk_plugin/open_telemetry_test.py new file mode 100644 index 00000000..a35783bf --- /dev/null +++ b/tests/claude_agent_sdk_plugin/open_telemetry_test.py @@ -0,0 +1,122 @@ +import asyncio +import json +import os +import threading +import uuid +from collections.abc import Iterator +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import Any + +import opentelemetry.trace +import pytest +from opentelemetry.sdk.trace.export import SimpleSpanProcessor +from opentelemetry.sdk.trace.export.in_memory_span_exporter import ( + InMemorySpanExporter, +) +from temporalio.claude_agent_sdk import ClaudeAgentSDKPlugin +from temporalio.client import Client +from temporalio.contrib.opentelemetry import OpenTelemetryPlugin, create_tracer_provider +from temporalio.worker import Worker + +from claude_agent_sdk_plugin.offline import scripted_api +from claude_agent_sdk_plugin.open_telemetry.run_worker import OFFLINE_ANSWERS +from claude_agent_sdk_plugin.open_telemetry.telemetry import CLAUDE_CODE_TRACING +from claude_agent_sdk_plugin.open_telemetry.workflow import OpenTelemetryWorkflow + + +@pytest.fixture +def otlp_collector() -> Iterator[tuple[str, list[dict[str, Any]]]]: + """Claude Code's spans, as it exports them over OTLP/HTTP JSON.""" + spans: list[dict[str, Any]] = [] + + class Handler(BaseHTTPRequestHandler): + def do_POST(self) -> None: + body = json.loads(self.rfile.read(int(self.headers["Content-Length"]))) + spans.extend( + span + for resource in body.get("resourceSpans", []) + for scope in resource["scopeSpans"] + for span in scope["spans"] + ) + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.end_headers() + self.wfile.write(b"{}") + + def log_message(self, format: str, *args: Any) -> None: + pass + + server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + threading.Thread(target=server.serve_forever, daemon=True).start() + yield f"http://127.0.0.1:{server.server_address[1]}", spans + server.shutdown() + server.server_close() + + +async def test_claude_code_spans_join_the_workflow_trace( + client: Client, otlp_collector: tuple[str, list[dict[str, Any]]] +) -> None: + endpoint, claude_code_spans = otlp_collector + exporter = InMemorySpanExporter() + provider = create_tracer_provider() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + opentelemetry.trace.set_tracer_provider(provider) + + config = client.config() + config["plugins"] = [OpenTelemetryPlugin(add_temporal_spans=True)] + client = Client(**config) + task_queue = f"claude-agent-sdk-open-telemetry-{uuid.uuid4()}" + + with scripted_api(OFFLINE_ANSWERS): + os.environ.update( + { + **CLAUDE_CODE_TRACING, + "OTEL_EXPORTER_OTLP_PROTOCOL": "http/json", + "OTEL_EXPORTER_OTLP_ENDPOINT": endpoint, + "OTEL_TRACES_EXPORT_INTERVAL": "100", + } + ) + try: + async with Worker( + client, + task_queue=task_queue, + workflows=[OpenTelemetryWorkflow], + plugins=[ClaudeAgentSDKPlugin()], + ): + tracer = opentelemetry.trace.get_tracer(__name__) + with tracer.start_as_current_span("ask-agent") as root: + result = await client.execute_workflow( + OpenTelemetryWorkflow.run, + "How many Python files are in the current directory?", + id=f"open-telemetry-{uuid.uuid4()}", + task_queue=task_queue, + ) + finally: + for name in CLAUDE_CODE_TRACING: + os.environ.pop(name, None) + os.environ.pop("OTEL_TRACES_EXPORT_INTERVAL", None) + + assert result == "I counted the Python files with ls." + + # Claude Code exports on its own schedule, and on exit. + for _ in range(50): + if any(s["name"] == "claude_code.interaction" for s in claude_code_spans): + break + await asyncio.sleep(0.1) + + trace_id = format(root.get_span_context().trace_id, "032x") + [connect] = [ + span + for span in exporter.get_finished_spans() + if span.name == "RunActivity:claude-agent-sdk-connect" + ] + [interaction] = [ + span for span in claude_code_spans if span["name"] == "claude_code.interaction" + ] + assert interaction["traceId"] == trace_id + assert interaction["parentSpanId"] == format(connect.context.span_id, "016x") + assert {span["name"] for span in claude_code_spans} >= { + "claude_code.llm_request", + "claude_code.tool", + } + assert {span["traceId"] for span in claude_code_spans} == {trace_id} diff --git a/tests/claude_agent_sdk_plugin/plugin_with_custom_transport_test.py b/tests/claude_agent_sdk_plugin/plugin_with_custom_transport_test.py new file mode 100644 index 00000000..97477f99 --- /dev/null +++ b/tests/claude_agent_sdk_plugin/plugin_with_custom_transport_test.py @@ -0,0 +1,197 @@ +import json +import os +import sys +import uuid +from pathlib import Path +from typing import Any + +import claude_agent_sdk +import pytest +from temporalio.claude_agent_sdk import ClaudeAgentSDKPlugin +from temporalio.client import Client +from temporalio.worker import Replayer, Worker + +from claude_agent_sdk_plugin.offline import scripted_api +from claude_agent_sdk_plugin.plugin_with_custom_transport.connector import ( + PROXY_MANAGED_KEY, + SandboxConnector, + find_saved_turn, +) +from claude_agent_sdk_plugin.plugin_with_custom_transport.run_worker import ( + HAIKU, + LOCAL_PLUGIN_DIR, + OFFLINE_ANSWERS, +) +from claude_agent_sdk_plugin.plugin_with_custom_transport.workflow import ( + COUNT_SYLLABLES, + PLUGIN_DIR, + PluginWithCustomTransportWorkflow, +) + +SECRET = "sk-ant-test-secret-never-in-history" + +FAKE_SBX = """#!{python} +import json, os, sys + +ROOT = {root!r} +BIN = {bin!r} +args = sys.argv[1:] +with open(os.path.join(ROOT, "log.jsonl"), "a") as log: + log.write(json.dumps(args) + "\\n") +sandboxes = os.path.join(ROOT, "sandboxes") +command = args[0] +if command == "ls": + print("\\n".join(sorted(os.listdir(sandboxes)))) +elif command == "create": + name = args[args.index("--name") + 1] + os.makedirs(os.path.join(sandboxes, name, "home/agent/workspace")) +elif command == "secret": + name = args[args.index("--sandbox") + 1] + with open(os.path.join(ROOT, "secrets.jsonl"), "a") as secrets: + secrets.write(json.dumps([name, sys.stdin.read().strip()]) + "\\n") +elif command == "exec": + rest = args[1:] + env = {{"PATH": BIN + ":/usr/bin:/bin"}} + workdir = "/home/agent/workspace" + interactive = False + while rest[0].startswith("-"): + flag = rest.pop(0) + if flag == "-i": + interactive = True + elif flag == "-e": + key, _, value = rest.pop(0).partition("=") + env[key] = value.replace("host.docker.internal", "127.0.0.1") + elif flag == "-w": + workdir = rest.pop(0) + name, *argv = rest + sandbox = os.path.join(sandboxes, name) + if not os.path.isdir(sandbox): + sys.exit("sandbox not found: " + name) + home = os.path.join(sandbox, "home/agent") + env["HOME"] = home + argv = [home + arg[len("/home/agent"):] if arg.startswith("/home/agent") else arg for arg in argv] + os.setsid() + with open(os.path.join(sandbox, "groups"), "a") as groups: + groups.write(str(os.getpid()) + "\\n") + os.chdir(os.path.join(sandbox, workdir.lstrip("/"))) + if not interactive: + os.dup2(os.open(os.devnull, os.O_RDONLY), 0) + os.execvpe(argv[0], argv, env) +""" + + +@pytest.fixture +def fake_sbx(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + """An `sbx` that runs each sandbox's commands in a temporary directory. + + Like the real sandbox, it has `claude` and `python3`: the Claude Code bundled + with the Claude Agent SDK and the test's own Python. + """ + root = tmp_path / "sbx" + bin = root / "bin" + bin.mkdir(parents=True) + (root / "sandboxes").mkdir() + (bin / "claude").symlink_to( + Path(claude_agent_sdk.__file__).parent / "_bundled" / "claude" + ) + (bin / "python3").symlink_to(sys.executable) + script = bin / "sbx" + script.write_text( + FAKE_SBX.format(python=sys.executable, root=str(root), bin=str(bin)) + ) + script.chmod(0o755) + monkeypatch.setenv("PATH", f"{bin}{os.pathsep}{os.environ['PATH']}") + return root + + +def sbx_calls(root: Path) -> list[list[str]]: + return [json.loads(line) for line in (root / "log.jsonl").read_text().splitlines()] + + +def user_blocks(request: dict[str, Any]) -> list[dict[str, Any]]: + return [ + block + for message in request["messages"] + if message["role"] == "user" and isinstance(message["content"], list) + for block in message["content"] + ] + + +def tool_result(request: dict[str, Any], tool_use_id: str) -> dict[str, Any]: + return next( + block + for block in user_blocks(request) + if block.get("tool_use_id") == tool_use_id + ) + + +async def test_plugin_skill_and_mcp_server_run_from_the_copy_in_the_sandbox( + client: Client, fake_sbx: Path +) -> None: + task_queue = f"claude-agent-sdk-plugin-with-custom-transport-{uuid.uuid4()}" + with scripted_api(OFFLINE_ANSWERS) as api: + os.environ["ANTHROPIC_API_KEY"] = SECRET + async with Worker( + client, + task_queue=task_queue, + workflows=[PluginWithCustomTransportWorkflow], + activities=[find_saved_turn], + plugins=[ + ClaudeAgentSDKPlugin( + connector=SandboxConnector(uploads={PLUGIN_DIR: LOCAL_PLUGIN_DIR}) + ) + ], + ): + handle = await client.start_workflow( + PluginWithCustomTransportWorkflow.run, + "Write a haiku about durable execution.", + id=f"claude-agent-sdk-plugin-with-custom-transport-{uuid.uuid4()}", + task_queue=task_queue, + ) + result = await handle.result() + + assert result == f"{HAIKU}\n\n~ haiku-writer" + assert api.asked == ["haiku", "tool_result:toolu_skill", "tool_result:toolu_count"] + + first, second, third = api.requests + assert sorted(t["name"] for t in first["tools"]) == ["Skill", COUNT_SYLLABLES] + assert "- haiku-writer:haiku: Write the answer as a haiku." in json.dumps(first) + + # The skill's text reached Claude. + assert tool_result(second, "toolu_skill")["content"] == ( + "Launching skill: haiku-writer:haiku" + ) + skill_text = "\n".join( + b["text"] for b in user_blocks(second) if b.get("type") == "text" + ) + assert "count_syllables" in skill_text + + # The plugin's MCP server ran in the sandbox and answered. + counts = json.dumps(tool_result(third, "toolu_count")["content"]) + assert "5: Workers fall and rise" in counts + assert "7: history remembers all" in counts + + # Claude Code got the plugin's path in the sandbox, which only the upload filled. + calls = sbx_calls(fake_sbx) + [create] = [call for call in calls if call[0] == "create"] + name = create[create.index("--name") + 1] + upload = next(call for call in calls if call[0] == "exec" and "tar" in call[-3]) + launch = next(call for call in calls if call[0] == "exec" and "claude" in call) + assert upload[-1] == PLUGIN_DIR + assert calls.index(upload) < calls.index(launch) + assert launch[launch.index("--plugin-dir") + 1] == PLUGIN_DIR + + # The key went to the sandbox's proxy, and Claude Code got a placeholder. + assert f"ANTHROPIC_API_KEY={PROXY_MANAGED_KEY}" in launch + secrets = [json.loads(line) for line in (fake_sbx / "secrets.jsonl").open()] + assert [name, SECRET] in secrets + + # The sandbox outlives the Workflow, for the user to remove. + assert (fake_sbx / "sandboxes" / name).exists() + assert not any(call[0] == "rm" for call in calls) + + history = await handle.fetch_history() + assert SECRET not in history.to_json() + await Replayer( + workflows=[PluginWithCustomTransportWorkflow], plugins=[ClaudeAgentSDKPlugin()] + ).replay_workflow(history) diff --git a/tests/claude_agent_sdk_plugin/run_worker_test.py b/tests/claude_agent_sdk_plugin/run_worker_test.py new file mode 100644 index 00000000..96fbc7b6 --- /dev/null +++ b/tests/claude_agent_sdk_plugin/run_worker_test.py @@ -0,0 +1,295 @@ +"""Drives each scenario's documented entry point. + +Each test runs the real ``run_worker.main()`` offline against the test server, +then does what the scenario README does with the Temporal CLI, through the +Python client. +""" + +import asyncio +import contextlib +import importlib +import sys +import uuid +from collections.abc import AsyncIterator, Awaitable, Callable +from pathlib import Path +from typing import Any, TypeVar + +import pytest +from temporalio.client import Client, WorkflowContinuedAsNewError, WorkflowHandle +from temporalio.contrib.pydantic import pydantic_data_converter + +from claude_agent_sdk_plugin.agent_chat.workflow import ChatWorkflow +from claude_agent_sdk_plugin.human_in_the_loop.workflow import ( + Decision, + HumanInTheLoopWorkflow, +) +from claude_agent_sdk_plugin.multi_agent.run_worker import ( + BRIEFING, + RESEARCH_NOTES, +) +from claude_agent_sdk_plugin.multi_agent.workflow import MultiAgentWorkflow +from claude_agent_sdk_plugin.plugin_with_custom_transport.workflow import ( + PluginWithCustomTransportWorkflow, +) +from claude_agent_sdk_plugin.streaming import run_workflow +from claude_agent_sdk_plugin.streaming.workflow import StreamingWorkflow +from claude_agent_sdk_plugin.structured_output.workflow import ( + SupportTicket, + TriageTicketWorkflow, +) +from claude_agent_sdk_plugin.tools.workflow import ToolsWorkflow + +# Reused so the plugin_with_custom_transport entry point runs against a fake `sbx` on PATH. +from tests.claude_agent_sdk_plugin.plugin_with_custom_transport_test import ( # noqa: F401 + fake_sbx, +) + +T = TypeVar("T") + +TIMEOUT = 120 +STREAMING_PROMPT = "Count from one to five, one number per sentence." +HAIKU = ( + "Workers fall and rise\n" + "history remembers all\n" + "the work carries on\n" + "\n" + "~ haiku-writer" +) + + +def workflow_id(scenario: str) -> str: + return f"{scenario}-{uuid.uuid4()}" + + +def task_queue(scenario: str) -> str: + return f"claude-agent-sdk-{scenario.replace('_', '-')}" + + +@contextlib.asynccontextmanager +async def running_main( + scenario: str, client: Client, monkeypatch: pytest.MonkeyPatch +) -> AsyncIterator[Callable[[Awaitable[T]], Awaitable[T]]]: + """Runs the scenario's `run_worker.main()` offline in the background. + + Yields `until`, which awaits an awaitable but fails fast, with the Worker's + error, if `main()` stops first. On exit, cancels `main()` and waits for its + Worker to finish shutting down. + """ + monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False) + monkeypatch.delenv("CLAUDE_CODE_OAUTH_TOKEN", raising=False) + monkeypatch.setenv("TEMPORAL_ADDRESS", client.service_client.config.target_host) + run_worker = importlib.import_module( + f"claude_agent_sdk_plugin.{scenario}.run_worker" + ) + main = asyncio.create_task(run_worker.main()) + + async def until(awaitable: Awaitable[T]) -> T: + step = asyncio.ensure_future(awaitable) + await asyncio.wait( + {main, step}, timeout=TIMEOUT, return_when=asyncio.FIRST_COMPLETED + ) + if main.done(): + step.cancel() + main.result() + raise AssertionError("run_worker.main() returned") + if not step.done(): + step.cancel() + raise AssertionError("Timed out") + return step.result() + + try: + yield until + finally: + main.cancel() + with contextlib.suppress(asyncio.CancelledError): + await main + + +async def pending_call( + until: Callable[[Awaitable[Any]], Awaitable[Any]], + handle: WorkflowHandle, + query: Any, + tool_use_id: str, +) -> Any: + async def poll() -> Any: + while True: + for call in await handle.query(query): + if call.tool_use_id == tool_use_id: + return call + await asyncio.sleep(0.2) + + return await until(poll()) + + +@pytest.mark.parametrize( + ("scenario", "run", "argument", "expected"), + [ + ( + "tools", + ToolsWorkflow.run, + "Count the r letters in strawberry, then get the weather in Paris.", + "There are 3 r's in strawberry. It is 21 C and sunny in Paris.", + ), + ( + "structured_output", + TriageTicketWorkflow.run, + "The app crashes every time I open the settings page.", + SupportTicket( + category="bug", + priority=2, + summary="The settings page crashes the app when opened.", + ), + ), + ( + "streaming", + StreamingWorkflow.run, + STREAMING_PROMPT, + "One. Two. Three. Four. Five.", + ), + ], +) +async def test_execute_workflow( + scenario: str, + run: Any, + argument: str, + expected: Any, + client: Client, + monkeypatch: pytest.MonkeyPatch, +) -> None: + if scenario == "structured_output": + client = await Client.connect( + client.service_client.config.target_host, + data_converter=pydantic_data_converter, + ) + async with running_main(scenario, client, monkeypatch) as until: + result = await until( + client.execute_workflow( + run, + argument, + id=workflow_id(scenario), + task_queue=task_queue(scenario), + ) + ) + assert result == expected + + +async def test_multi_agent(client: Client, monkeypatch: pytest.MonkeyPatch) -> None: + async with running_main("multi_agent", client, monkeypatch) as until: + briefing = await until( + client.execute_workflow( + MultiAgentWorkflow.run, + "durable execution", + id=workflow_id("multi-agent"), + task_queue=task_queue("multi_agent"), + ) + ) + assert briefing.result == f"Final briefing: {BRIEFING}" + assert briefing.messages_by_agent["researcher"] == [RESEARCH_NOTES] + assert briefing.messages_by_agent["writer"] == [BRIEFING] + assert briefing.messages_by_agent["coordinator"][-1] == briefing.result + + +async def test_agent_chat(client: Client, monkeypatch: pytest.MonkeyPatch) -> None: + async with running_main("agent_chat", client, monkeypatch) as until: + chat_id = workflow_id("agent-chat") + handle = await client.start_workflow( + "ChatWorkflow", + {}, + id=chat_id, + task_queue=task_queue("agent_chat"), + ) + replies = [ + await until( + handle.execute_update( + ChatWorkflow.send_message, f"Message {number}, my name is Ada." + ) + ) + for number in range(1, 4) + ] + # The third message fills the run, so the fourth is answered by a new + # run that resumes Claude Code's session. + with pytest.raises(WorkflowContinuedAsNewError): + await until(handle.result(follow_runs=False)) + replies.append( + await until( + client.get_workflow_handle(chat_id).execute_update( + ChatWorkflow.send_message, "What is my name?" + ) + ) + ) + await until( + client.get_workflow_handle(chat_id).signal(ChatWorkflow.end_conversation) + ) + await until(client.get_workflow_handle(chat_id).result()) + assert replies == [f"Offline reply {n} in this conversation." for n in range(1, 5)] + + +@pytest.mark.parametrize( + ("approved", "expected", "creates_file"), + [ + (True, "Created hello.txt.", True), + (False, "Understood. I did not create hello.txt.", False), + ], +) +async def test_human_in_the_loop( + approved: bool, + expected: str, + creates_file: bool, + client: Client, + workspace: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + async with running_main("human_in_the_loop", client, monkeypatch) as until: + handle = await client.start_workflow( + HumanInTheLoopWorkflow.run, + "Create hello.txt with a greeting.", + id=workflow_id("human-in-the-loop"), + task_queue=task_queue("human_in_the_loop"), + ) + request = await pending_call( + until, handle, HumanInTheLoopWorkflow.pending_tool_calls, "toolu_hello" + ) + assert request.tool_name == "Bash" + assert request.tool_input == { + "command": "echo 'Hello from Claude Code' > hello.txt" + } + await until( + handle.execute_update( + HumanInTheLoopWorkflow.decide, Decision("toolu_hello", approved) + ) + ) + result = await until(handle.result()) + assert result == expected + assert (workspace / "hello.txt").exists() is creates_file + + +async def test_plugin_with_custom_transport( + client: Client, + fake_sbx: Path, # noqa: F811 + monkeypatch: pytest.MonkeyPatch, +) -> None: + async with running_main( + "plugin_with_custom_transport", client, monkeypatch + ) as until: + result = await until( + client.execute_workflow( + PluginWithCustomTransportWorkflow.run, + "Write a haiku about durable execution.", + id=workflow_id("plugin-with-custom-transport"), + task_queue=task_queue("plugin_with_custom_transport"), + ) + ) + assert result == HAIKU + + +async def test_streaming_run_workflow_prints_deltas_and_result( + client: Client, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + monkeypatch.setattr(sys, "argv", ["run_workflow.py"]) + async with running_main("streaming", client, monkeypatch) as until: + await until(run_workflow.main()) + assert capsys.readouterr().out.endswith( + "One. Two. Three. Four. Five.\nFinal result: One. Two. Three. Four. Five.\n" + ) diff --git a/tests/claude_agent_sdk_plugin/streaming_test.py b/tests/claude_agent_sdk_plugin/streaming_test.py new file mode 100644 index 00000000..c862b2fc --- /dev/null +++ b/tests/claude_agent_sdk_plugin/streaming_test.py @@ -0,0 +1,63 @@ +import asyncio +import uuid + +from temporalio.claude_agent_sdk import ClaudeAgentSDKPlugin +from temporalio.client import Client, WorkflowHandle +from temporalio.contrib.workflow_streams import WorkflowStreamClient +from temporalio.worker import Replayer, Worker + +from claude_agent_sdk_plugin.offline import scripted_api +from claude_agent_sdk_plugin.streaming.workflow import StreamingWorkflow + +CHUNKS = ["One. ", "Two. ", "Three. ", "Four. ", "Five."] + + +async def start( + client: Client, task_queue: str +) -> WorkflowHandle[StreamingWorkflow, str]: + return await client.start_workflow( + StreamingWorkflow.run, + "Count to five", + id=f"claude-agent-sdk-streaming-{uuid.uuid4()}", + task_queue=task_queue, + ) + + +async def test_subscriber_receives_deltas_in_order(client: Client) -> None: + task_queue = f"claude-agent-sdk-streaming-{uuid.uuid4()}" + with scripted_api({"Count": CHUNKS}): + async with Worker( + client, + task_queue=task_queue, + workflows=[StreamingWorkflow], + plugins=[ClaudeAgentSDKPlugin()], + ): + handle = await start(client, task_queue) + stream = WorkflowStreamClient.create(client, handle.id) + deltas = [ + item.data["event"]["delta"]["text"] + async for item in stream.subscribe(["events"]) + if item.data["event"].get("delta", {}).get("type") == "text_delta" + ] + result = await handle.result() + + assert deltas == CHUNKS + assert "".join(deltas) == result + await Replayer( + workflows=[StreamingWorkflow], plugins=[ClaudeAgentSDKPlugin()] + ).replay_workflow(await handle.fetch_history()) + + +async def test_completes_without_subscriber(client: Client) -> None: + task_queue = f"claude-agent-sdk-streaming-{uuid.uuid4()}" + with scripted_api({"Count": CHUNKS}): + async with Worker( + client, + task_queue=task_queue, + workflows=[StreamingWorkflow], + plugins=[ClaudeAgentSDKPlugin()], + ): + handle = await start(client, task_queue) + result = await asyncio.wait_for(handle.result(), timeout=60) + + assert result == "".join(CHUNKS) diff --git a/tests/claude_agent_sdk_plugin/structured_output_test.py b/tests/claude_agent_sdk_plugin/structured_output_test.py new file mode 100644 index 00000000..32d4b9f1 --- /dev/null +++ b/tests/claude_agent_sdk_plugin/structured_output_test.py @@ -0,0 +1,116 @@ +import uuid +from typing import Any + +import pytest +import pytest_asyncio +from temporalio.claude_agent_sdk import ClaudeAgentSDKPlugin +from temporalio.client import Client, WorkflowFailureError, WorkflowHandle +from temporalio.contrib.pydantic import pydantic_data_converter +from temporalio.exceptions import ApplicationError +from temporalio.worker import Replayer, Worker + +from claude_agent_sdk_plugin.offline import ScriptedApi, scripted_api, tool_use +from claude_agent_sdk_plugin.structured_output.workflow import ( + SupportTicket, + TriageTicketWorkflow, +) + +MESSAGE = "The app crashes every time I open the settings page." +TICKET = SupportTicket(category="bug", priority=2, summary="Settings page crashes") + + +@pytest_asyncio.fixture +async def pydantic_client(client: Client) -> Client: + config = client.config() + config["data_converter"] = pydantic_data_converter + return Client(**config) + + +async def triage( + client: Client, answers: dict[str, Any] +) -> tuple[ScriptedApi, WorkflowHandle[Any, SupportTicket]]: + task_queue = f"claude-agent-sdk-structured-output-{uuid.uuid4()}" + with scripted_api(answers) as api: + async with Worker( + client, + task_queue=task_queue, + workflows=[TriageTicketWorkflow], + plugins=[ClaudeAgentSDKPlugin()], + ): + handle = await client.start_workflow( + TriageTicketWorkflow.run, + MESSAGE, + id=f"structured-output-{uuid.uuid4()}", + task_queue=task_queue, + ) + await handle.result() + return api, handle + + +async def test_returns_validated_model(pydantic_client: Client) -> None: + api, handle = await triage( + pydantic_client, + { + "Triage this customer message": tool_use( + "toolu_1", "StructuredOutput", **TICKET.model_dump() + ) + }, + ) + + assert await handle.result() == TICKET + + schema_tool = next( + tool for tool in api.requests[0]["tools"] if tool["name"] == "StructuredOutput" + ) + assert schema_tool["input_schema"] == SupportTicket.model_json_schema() + + await Replayer( + workflows=[TriageTicketWorkflow], plugins=[ClaudeAgentSDKPlugin()] + ).replay_workflow(await handle.fetch_history()) + + +async def test_claude_code_returns_schema_errors_to_the_model( + pydantic_client: Client, +) -> None: + api, handle = await triage( + pydantic_client, + { + "Triage this customer message": tool_use( + "toolu_1", "StructuredOutput", **{**TICKET.model_dump(), "priority": 9} + ), + "tool_result:toolu_1": tool_use( + "toolu_2", "StructuredOutput", **TICKET.model_dump() + ), + }, + ) + + assert await handle.result() == TICKET + assert api.asked == ["Triage this customer message", "tool_result:toolu_1"] + schema_error = next( + block + for request in api.requests + for message in request["messages"] + if message["role"] == "user" and not isinstance(message["content"], str) + for block in message["content"] + if block["type"] == "tool_result" + ) + assert schema_error["is_error"] + assert "priority: must be <= 5" in schema_error["content"] + + +async def test_workflow_fails_without_structured_output( + pydantic_client: Client, +) -> None: + with pytest.raises(WorkflowFailureError) as failure: + await triage( + pydantic_client, + { + "Triage this customer message": "I cannot triage this.", + "structured-output-enforce": "I still cannot triage this.", + }, + ) + + cause = failure.value.cause + assert isinstance(cause, ApplicationError) + assert cause.message == "Claude Code returned no structured output" + assert cause.non_retryable diff --git a/tests/claude_agent_sdk_plugin/tools_test.py b/tests/claude_agent_sdk_plugin/tools_test.py new file mode 100644 index 00000000..b74455a8 --- /dev/null +++ b/tests/claude_agent_sdk_plugin/tools_test.py @@ -0,0 +1,70 @@ +import uuid + +from temporalio.api.enums.v1 import EventType +from temporalio.claude_agent_sdk import ClaudeAgentSDKPlugin +from temporalio.client import Client +from temporalio.worker import Replayer, Worker + +from claude_agent_sdk_plugin.offline import Answer, scripted_api, tool_use +from claude_agent_sdk_plugin.tools.workflow import ToolsWorkflow, fetch_weather + + +async def test_tools(client: Client) -> None: + answers: dict[str, Answer] = { + "strawberry": tool_use( + "toolu_count", "mcp__demo__count_letter", word="strawberry", letter="r" + ), + "tool_result:toolu_count": tool_use( + "toolu_weather", "mcp__demo__get_weather", city="Paris" + ), + "tool_result:toolu_weather": "3 r's in strawberry. 21 C and sunny in Paris.", + } + task_queue = f"claude-agent-sdk-tools-{uuid.uuid4()}" + + with scripted_api(answers) as api: + async with Worker( + client, + task_queue=task_queue, + workflows=[ToolsWorkflow], + activities=[fetch_weather], + plugins=[ClaudeAgentSDKPlugin()], + ): + handle = await client.start_workflow( + ToolsWorkflow.run, + "Count the r's in strawberry, then get the weather in Paris.", + id=f"claude-agent-sdk-tools-{uuid.uuid4()}", + task_queue=task_queue, + ) + result = await handle.result() + + assert result == "3 r's in strawberry. 21 C and sunny in Paris." + assert api.asked == [ + "strawberry", + "tool_result:toolu_count", + "tool_result:toolu_weather", + ] + tool_results = [ + block["content"][0]["text"] + for message in api.requests[-1]["messages"] + if message["role"] == "user" and isinstance(message["content"], list) + for block in message["content"] + if block.get("type") == "tool_result" + ] + assert tool_results == ["3", "It is 21 C and sunny in Paris."] + assert [t["name"] for t in api.requests[0]["tools"]] == [ + "mcp__demo__count_letter", + "mcp__demo__get_weather", + ] + + history = await handle.fetch_history() + scheduled = [ + event.activity_task_scheduled_event_attributes.activity_type.name + for event in history.events + if event.event_type == EventType.EVENT_TYPE_ACTIVITY_TASK_SCHEDULED + ] + assert scheduled.count("fetch_weather") == 1 + assert "count_letter" not in scheduled + + await Replayer( + workflows=[ToolsWorkflow], plugins=[ClaudeAgentSDKPlugin()] + ).replay_workflow(history) diff --git a/uv.lock b/uv.lock index 74d3488d..c46b2012 100644 --- a/uv.lock +++ b/uv.lock @@ -652,6 +652,26 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/db/8f/61959034484a4a7c527811f4721e75d02d653a35afb0b6054474d8185d4c/charset_normalizer-3.4.7-py3-none-any.whl", hash = "sha256:3dce51d0f5e7951f8bb4900c257dad282f49190fdbebecd4ba99bcc41fef404d", size = 61958, upload-time = "2026-04-02T09:28:37.794Z" }, ] +[[package]] +name = "claude-agent-sdk" +version = "0.2.165" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "anyio" }, + { name = "jsonschema" }, + { name = "mcp" }, + { name = "sniffio" }, + { name = "typing-extensions", marker = "python_full_version < '3.11'" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/7f/45/10f00a408b480926b5e340e3507562ede1e30d4ce5b73eea9ccc3468a9e6/claude_agent_sdk-0.2.165.tar.gz", hash = "sha256:1bfa8e7a6bb36e82de9a12324c9bcfa6eebd4f96ee5890f0c656d50445240ff8", size = 365169, upload-time = "2026-10-08T18:18:49.528Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/e4/a7/a7226ae5e7fa3f225d99a0acf12ee354c4af9c953661f66bbcd424e48853/claude_agent_sdk-0.2.165-py3-none-macosx_11_0_arm64.whl", hash = "sha256:7f7017bb59eaf77b9a7c7ce3b9ed2c8a397b8201c8630f011bb1b955290f06e4", size = 97973182, upload-time = "2026-10-08T18:18:56.039Z" }, + { url = "https://files.pythonhosted.org/packages/5c/9d/1eaa82f8dbafa63a3fb26aabc088be307c4977438269cc6cb10056a32e99/claude_agent_sdk-0.2.165-py3-none-macosx_11_0_x86_64.whl", hash = "sha256:f4b5c6f536062e3af72357b1235f05ad4513230a32a3c639341a91a607d2df1e", size = 102479136, upload-time = "2026-10-08T18:19:01.908Z" }, + { url = "https://files.pythonhosted.org/packages/40/4a/93fe172811bcd7704c4c4776d5eed1bb1881ba440d86e0f8e5cbbec0dca9/claude_agent_sdk-0.2.165-py3-none-manylinux_2_17_aarch64.whl", hash = "sha256:46a47e1e1075a8f2b7976f6bd5aa06c5610cb322c47cdc5b728fb25d02691164", size = 107447420, upload-time = "2026-10-08T18:19:09.571Z" }, + { url = "https://files.pythonhosted.org/packages/57/81/a0d3ff04ae7045218566b36d1c78729c3fc7550877e154697f263ac5122e/claude_agent_sdk-0.2.165-py3-none-manylinux_2_17_x86_64.whl", hash = "sha256:9dbee4bfc69f0bb27ee455afdc19d78f958540a99217d2e28831fb9b7d496f08", size = 107800988, upload-time = "2026-10-08T18:19:16.24Z" }, + { url = "https://files.pythonhosted.org/packages/7f/d8/1685e8fc14bae5b9cbcfd6f4f6ca8b7ae8fd497a8bbf9c108978dd98620b/claude_agent_sdk-0.2.165-py3-none-win_amd64.whl", hash = "sha256:cf41dc1b1019bc7321b695d621fc94e6d8982c095cfb7d433d5c293e356a2404", size = 110401968, upload-time = "2026-10-08T18:19:24.136Z" }, +] + [[package]] name = "click" version = "8.4.0" @@ -4593,7 +4613,7 @@ wheels = [ [[package]] name = "temporalio" -version = "1.34.0" +version = "1.35.0" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "nexus-rpc" }, @@ -4602,15 +4622,15 @@ dependencies = [ { name = "types-protobuf" }, { name = "typing-extensions" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/13/85/9ece6f400552ed21951c8c5c534056e7d015cb065e3384470ec23ac24f28/temporalio-1.34.0.tar.gz", hash = "sha256:6453cb20e18df485e16578b22c82a9c4bcb1cf7eedd94147dfd373551d80f5b6", size = 3163890, upload-time = "2026-09-30T20:22:56.056Z" } +sdist = { url = "https://files.pythonhosted.org/packages/69/38/03641a02eb18e2b16f58c1c136deafdffeb30828dcd83ebd8ac41eb39fd7/temporalio-1.35.0.tar.gz", hash = "sha256:c7a14f29909f8d847f78de49305af89c0ec5aea39272cd4050124386ac3ca295", size = 2879693, upload-time = "2026-10-28T20:13:38.726Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/3c/71/e55380e7820819357afada375d4e5c33d141bb94a08fda35aa0551b4369b/temporalio-1.34.0-cp310-abi3-macosx_10_12_x86_64.whl", hash = "sha256:87118447ad13e1062b79bfc8b44b1695e8328ac9c6be1776e426e87b501b135a", size = 14064102, upload-time = "2026-09-30T20:22:35.601Z" }, - { url = "https://files.pythonhosted.org/packages/db/78/356c8e2f0ec1678b757d4c5e8a1b0b1c65b15b8afb2335ad2944c158ea10/temporalio-1.34.0-cp310-abi3-macosx_11_0_arm64.whl", hash = "sha256:023fff9cd9dd21860061e003880dcf95250700f5afab96c030c9cb258504dab8", size = 13763219, upload-time = "2026-09-30T20:22:38.774Z" }, - { url = "https://files.pythonhosted.org/packages/e9/00/4a4b4e018c4c12b4691211aca1d97db8707ed4c9e31f40dcf09784c9913d/temporalio-1.34.0-cp310-abi3-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:cdb6e2fb525ea5635afa4a3f1ae4c992c16dbe29dc92939a85153baeb69d95a2", size = 14121901, upload-time = "2026-09-30T20:22:41.428Z" }, - { url = "https://files.pythonhosted.org/packages/c1/20/b032d4a0df51d466fd929d8205e1288e83d97a890f021479f43af9bd6f9b/temporalio-1.34.0-cp310-abi3-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:540761f738bdfe5cb5bd7240b659e116a0b5094b94282aef09b8d8c2d66e9c52", size = 14457183, upload-time = "2026-09-30T20:22:44.215Z" }, - { url = "https://files.pythonhosted.org/packages/b6/0d/b08ffeca93bbc29200a5bd85a45cca3c6be0a123de605ecd1798b617df4f/temporalio-1.34.0-cp310-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:87647f87f42ecd45efb675642e2ec8ca34aa42359f5968f31f5391f1db5ebcbe", size = 14179836, upload-time = "2026-09-30T20:22:47.474Z" }, - { url = "https://files.pythonhosted.org/packages/81/55/acde4d1b7c9f23434e263752ef0a3ef58adeb3b95ef8472607414798dfaa/temporalio-1.34.0-cp310-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:71caa4b9061628b22a457c87c3b16da40f7ac6ac1a26ece9aa1ee6b4934404d7", size = 14570248, upload-time = "2026-09-30T20:22:50.569Z" }, - { url = "https://files.pythonhosted.org/packages/cc/c5/6bdc3b02ec483093a71e9f887e8a50ae20fcc79e0578ed7435a81d4271b1/temporalio-1.34.0-cp310-abi3-win_amd64.whl", hash = "sha256:03bd86561188c18d88425178bc690fe0791104b78566dcebd98f59cbdbce0952", size = 15415684, upload-time = "2026-09-30T20:22:53.342Z" }, + { url = "https://files.pythonhosted.org/packages/9e/d9/8bda464beda3adb897d8d9f0e5fa26b28f2bf63c831c6800cbd73f7728cb/temporalio-1.35.0-cp310-abi3-macosx_10_12_x86_64.whl", hash = "sha256:ac3ab920cc05e24a1d420688a49aabed7b2be8f25dd275200dfa839be30980fd", size = 13989719, upload-time = "2026-10-28T20:13:16.897Z" }, + { url = "https://files.pythonhosted.org/packages/26/c1/94861d72cbe5e1778fcd4685db63231af01d40ad99f0375b4df6d4e60d06/temporalio-1.35.0-cp310-abi3-macosx_11_0_arm64.whl", hash = "sha256:51bae017d484450084daa19afafad12fa0af1a00af351ba304104d7b97b44f9a", size = 13552491, upload-time = "2026-10-28T20:13:19.725Z" }, + { url = "https://files.pythonhosted.org/packages/3c/ed/a776afec1033f749ec8f4c0c5264167aedc56a108a5adebfa4084af8aa82/temporalio-1.35.0-cp310-abi3-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:b93845a2200d23b8edc9a8b0bf78b58bdb5e78721602b3f64072867e07e28504", size = 13927445, upload-time = "2026-10-28T20:13:21.984Z" }, + { url = "https://files.pythonhosted.org/packages/ca/0b/bf0ed1f1b623e1ffd5c21c9e65a331a5c1c4f77f3829dc99ceae99b69f39/temporalio-1.35.0-cp310-abi3-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:45b5438f41ab154eda4eea40069df2fa8779cb8f5c27f266b917a700bfba8bb8", size = 14386743, upload-time = "2026-10-28T20:13:28.201Z" }, + { url = "https://files.pythonhosted.org/packages/76/d6/373d5ffe28263fc56ea0013bdd45715b4ff84f97d1052519b7c0d9ecd437/temporalio-1.35.0-cp310-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:8a5ce8b1e9c9e2df5fb9f17a345c0ef0a132c787cef8cf5a6593b5d707648253", size = 13988280, upload-time = "2026-10-28T20:13:31.667Z" }, + { url = "https://files.pythonhosted.org/packages/9c/41/47ad6f26002f84d8293d99098c414f75e9e8ff7955d034b7121bd211649b/temporalio-1.35.0-cp310-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:96310ae91b098c9fe2194f0892cef25381e8f870e48d5c069230312bbd1c6262", size = 14494297, upload-time = "2026-10-28T20:13:33.988Z" }, + { url = "https://files.pythonhosted.org/packages/db/33/30942b2d568df2c50089e381ee63b2d141f2ebfb7b4787bac3540f5cf617/temporalio-1.35.0-cp310-abi3-win_amd64.whl", hash = "sha256:6f434c0d3b6d049f4473f9f5d3e80b5e80c986440fd0490f2ee70cc631cbcddd", size = 15317073, upload-time = "2026-10-28T20:13:36.195Z" }, ] [package.optional-dependencies] @@ -4622,6 +4642,15 @@ pydantic = [ { name = "pydantic" }, ] +[[package]] +name = "temporalio-claude-agent-sdk" +version = "0.0.0" +source = { git = "https://github.com/temporalio/ai-integrations?subdirectory=python%2Fclaude_agent_sdk&rev=177fb623f874b85db519401c4be496eeb19da8f5#177fb623f874b85db519401c4be496eeb19da8f5" } +dependencies = [ + { name = "claude-agent-sdk" }, + { name = "temporalio" }, +] + [[package]] name = "temporalio-deepagents" version = "0.0.1" @@ -4745,6 +4774,11 @@ dependencies = [ bedrock = [ { name = "boto3" }, ] +claude-agent-sdk = [ + { name = "opentelemetry-exporter-otlp-proto-grpc" }, + { name = "temporalio", extra = ["opentelemetry", "pydantic"] }, + { name = "temporalio-claude-agent-sdk" }, +] cloud-export-to-parquet = [ { name = "boto3" }, { name = "numpy", marker = "python_full_version < '3.13'" }, @@ -4867,6 +4901,11 @@ requires-dist = [ [package.metadata.requires-dev] bedrock = [{ name = "boto3", specifier = ">=1.34.92,<2" }] +claude-agent-sdk = [ + { name = "opentelemetry-exporter-otlp-proto-grpc", specifier = ">=1.30.0,<2" }, + { name = "temporalio", extras = ["opentelemetry", "pydantic"], specifier = ">=1.35.0,<2" }, + { name = "temporalio-claude-agent-sdk", git = "https://github.com/temporalio/ai-integrations?subdirectory=python%2Fclaude_agent_sdk&rev=177fb623f874b85db519401c4be496eeb19da8f5" }, +] cloud-export-to-parquet = [ { name = "boto3", specifier = ">=1.34.89,<2" }, { name = "numpy", marker = "python_full_version >= '3.10' and python_full_version < '3.13'", specifier = ">=1.26.0,<2" },