Claude Agent SDK
View as MarkdownThe Claude Agent SDK runs Claude Code as an agent. Its agent loop runs in the Claude Code CLI, a subprocess of the Worker. The Dex adapter runs that loop inside a durable session. It is available for Python and TypeScript.
When to use it
Use it when you want the Claude Code agent loop, its context management, and its tools. Use it when your agent already uses the Claude Agent SDK.
Use the built-in agent loop when you want one model call per Step, parallel tool Steps, or Dex compaction.
How a session runs
Claude Code runs a whole turn, so the adapter splits a turn into segments. One RunModel Step runs the CLI for one segment. Model calls run inside the segment. So do read tools: a deferred read would restart the CLI. A Connector tool call is the exception: it always leaves the segment, even a read. See Connect Connectors.
A write tool call ends the segment. It waits in AwaitToolApproval when it needs approval. Then it runs in its own Step: CallTool for an idempotent write, CallSingleAttemptTool for any other write. ResumeTurn then returns to RunModel, which resumes the same Claude session. The CLI gets the recorded result and does not run the tool again.
A failed segment retries, up to three attempts, and reruns whole. Model calls and reads inside it run again. A write that ran in its own Step never runs again.
A turn that looks up an order runs one RunModel Step. A turn that cancels an order runs RunModel, AwaitToolApproval, CallSingleAttemptTool, ResumeTurn, and RunModel.
Authentication
The Worker authenticates the Claude Code CLI with ANTHROPIC_API_KEY. Set it in the Worker's environment. It never enters Flow state.
The example builds the CLI's environment so that ANTHROPIC_API_KEY is its only credential. Without the key, the example does not register the Claude agent. It serves the other harnesses.
Install
Install the adapter with its framework. Each adapter brings in the Dex Agents core.
pip install dex-agents-claude-agent-sdk
The example
The example is the order support agent that every framework page uses.
The Flow type is ClaudeAgentSDKOrderSupport in Python and ClaudeAgentSdkOrderSupportAgent in TypeScript. The full example is under examples/python/agent-harnesses and examples/typescript/src/products/agent-harnesses.
Agent and tools
The tools are Claude Agent SDK tools. The options give the model, the system prompt, and the CLI environment. The tool policies name each tool by its own name.
def build_claude_agent_sdk_order_support_flow(
order_desk: OrderDesk, model: str, slack_connection: slack.Connection | None
) -> AgentSessionFlow:
"""Builds the session Flow; ``model`` is a Claude model name, such as "claude-opus-5-5"."""
# The tools are closures, so they call the OrderDesk passed in here.
@tool(
"look_up_order",
"Look up one order by its order ID and return its items and status.",
{"order_id": str},
)
async def look_up_order(arguments: dict[str, Any]) -> dict[str, Any]:
order = order_desk.describe_order(arguments["order_id"])
return {"content": [{"type": "text", "text": order}]}
@tool(
"cancel_order",
"Cancel one order that has not shipped yet.",
{"order_id": str, "reason": str},
)
async def cancel_order(arguments: dict[str, Any]) -> dict[str, Any]:
result = order_desk.cancel_order(arguments["order_id"], arguments["reason"])
return {"content": [{"type": "text", "text": result}]}
# Every session gets the same options, so the factory does not read its context.
def build_options(context: ClaudeAgentOptionsContext) -> ClaudeAgentOptions:
return ClaudeAgentOptions(
model=model,
env=build_claude_cli_environment(os.environ),
system_prompt=ORDER_SUPPORT_INSTRUCTIONS,
# No Claude Code built-in tools: the agent uses only the order tools.
tools=[],
# Settings files could approve a tool before Dex asks for approval.
setting_sources=[],
)
return claude_agent_session_flow(
build_options,
[look_up_order, cancel_order],
flow_type=CLAUDE_AGENT_SDK_ORDER_SUPPORT_FLOW_TYPE,
tool_sources=build_slack_tool_sources(slack_connection),
tool_policies={
"look_up_order": ToolPolicy(ToolEffect.READ),
"cancel_order": ToolPolicy(ToolEffect.NON_IDEMPOTENT_WRITE),
},
)
Example: examples/python/agent-harnesses/agent_harness_examples/claude_agent_sdk_order_support.py
The example turns off every built-in Claude Code tool, so the agent uses only the order tools. It also loads no settings files, so no settings file can approve a tool before Dex asks.
Worker
One Worker serves the session Flows of every harness in the example. Agent session Steps are async, so the Python Registry allows async handlers. Build the same session Flow in every process that starts sessions or calls their RPCs.
# Agent session Steps are async, so the Registry allows async handlers.
self.registry = Registry(
tuple(example.session_flow for example in self.examples), allow_async_handlers=True
)
config.blob_cache_dir.mkdir(parents=True, exist_ok=True)
self.blob_cache = open_blob_cache(BlobCacheConfig(str(config.blob_cache_dir), 1 << 30))
self.worker = AsyncWorker(
self.registry,
self.blob_cache,
WorkerOptions(
server_address=config.server_address,
pull=PullWorkerOptions(max_concurrent_tasks=MAX_CONCURRENT_TASKS),
),
)
self.client = AsyncClient(
self.registry, self.blob_cache, ClientOptions(server_address=config.server_address)
)
# ...
def _new_claude_agent_sdk_example(
config: AgentHarnessExamplesConfig, slack_connection: slack.Connection | None
) -> AgentHarnessExample:
order_desk = OrderDesk()
flow = build_claude_agent_sdk_order_support_flow(
order_desk, config.claude_agent_sdk_model, slack_connection
)
return AgentHarnessExample("claude-agent-sdk", flow, order_desk)
Example: examples/python/agent-harnesses/agent_harness_examples/app.py
Session routes
Every harness serves these same routes. Session routes explains them and the session token.
@blueprint.post("/sessions")
async def start_session() -> Response:
# The server picks the session ID, so a caller cannot claim an existing session.
session_id = f"{harness_name}-order-support-{uuid.uuid4()}"
await sessions.start_session(session_id)
session_token = sign_session_id(app_state.config.session_key, session_id)
return jsonify(session_id=session_id, session_token=session_token)
@blueprint.post("/messages")
async def send_user_message() -> Response:
body, session_id = await session_request()
accepted = await sessions.send_user_message(session_id, _required_string(body, "text"))
return jsonify(accepted=accepted)
@blueprint.post("/interrupts/resolve")
async def resolve_interrupt() -> Response:
body, session_id = await session_request()
interrupt_id = _required_string(body, "interrupt_id")
if not isinstance(approved := body.get("approved"), bool):
raise BadRequest("approved must be a boolean")
rejection_message = body.get("rejection_message")
if rejection_message is not None and not isinstance(rejection_message, str):
raise BadRequest("rejection_message must be a string")
accepted = await sessions.resolve_interrupt(
session_id, interrupt_id, approved=approved, rejection_message=rejection_message
)
return jsonify(accepted=accepted)
@blueprint.get("/snapshot")
async def get_agent_snapshot() -> Response:
session_id = request.args.get("session_id", "")
if not session_id:
raise BadRequest("session_id is required")
authorize_session(session_id)
try:
snapshot = await read_agent_event_stream_snapshot(
app_state.client, example.session_flow, session_id
)
except FlowNotActiveOrNotFoundError as error:
raise NotFound("the agent session is missing or no longer retained") from error
return jsonify(snapshot.to_json())
Example: examples/python/agent-harnesses/agent_harness_examples/http_routes.py
Run it locally
Start Dex and the example
In a first terminal, render the session graph, then start Dex. Dex Web reads the graph when it starts, to show the session and its approval controls.
cd examples/python/agent-harnesses
uv sync --locked
mkdir -p definitions
uv run --frozen dex-agents-session-graph --flow-type ClaudeAgentSDKOrderSupport \
--permission order-support.operate --out definitions/claude-agent-sdk.json
dexcli dev --flow-rendering-dir ./definitions
dexcli dev stays in the foreground. In a second terminal, start the example:
cd examples/python/agent-harnesses
ANTHROPIC_API_KEY=... uv run --frozen python -m agent_harness_examples
The Python app serves on 127.0.0.1:8081. The TypeScript app serves on 127.0.0.1:8080. Dex Web is at http://127.0.0.1:8802.
| Variable | Default | Meaning |
|---|---|---|
| ANTHROPIC_API_KEY | none; required | The Anthropic API key of the Worker. |
| DEX_CLAUDE_AGENT_SDK_MODEL | claude-opus-5-5 | The Claude model name. |
| DEX_AGENT_SESSION_KEY | a random key | The key that signs session tokens. A random key ends every token when the process restarts. |
Call the session
Start a session, ask the agent to cancel order 1002, and approve the call:
base=http://127.0.0.1:8081/products/agent-harnesses/claude-agent-sdk
session=$(curl -s -X POST "$base/sessions" -H 'Content-Type: application/json' -d '{}')
id=$(echo "$session" | jq -r .session_id)
token=$(echo "$session" | jq -r .session_token)
curl -s -X POST "$base/messages" -H 'Content-Type: application/json' \
-H "X-Agent-Session-Token: $token" -d "{\"session_id\": \"$id\", \"text\": \"Cancel order 1002\"}"
until interrupt=$(curl -s "$base/snapshot?session_id=$id" -H "X-Agent-Session-Token: $token" \
| jq -re '.snapshot.pending_interrupts[0].interrupt_id'); do sleep 1; done
curl -s -X POST "$base/interrupts/resolve" -H 'Content-Type: application/json' \
-H "X-Agent-Session-Token: $token" \
-d "{\"session_id\": \"$id\", \"interrupt_id\": \"$interrupt\", \"approved\": true}"
The loop reads the snapshot until the agent asks to run cancel_order. That call waits for approval, then runs once. Read the snapshot again to see the agent's answer.
Approve in Dex Web
Instead of the last command, you can approve the call in Dex Web. Follow the steps in Approve in Dex Web.
Connect Connectors
The agent can call Connector operations as tools, as Connectors as tools describes. The example adds the Slack tool slack_post_thread_reply, which replies in a Slack thread.
Offer the tool
In Python, install the connectors extra of dex-agents and the Connector package, such as dex-connector-slack. In TypeScript, install @superdurable/connectors-sdk and the Connector package, such as @superdurable/connector-slack. The install commands are in Connector tools.
The Worker opens the Slack connection support-slack from the project configuration. connector_tool in Python and connectorTool in TypeScript turns the operation into a tool, and ConnectorToolSource holds it. Without the connection, the agent gets no Slack tool.
def build_slack_tool_sources(connection: slack.Connection | None) -> list[ToolSource]:
"""Returns the Slack tool source, or no source when Slack is not configured."""
if connection is None:
return []
return [
ConnectorToolSource([
# The manifest does not ask for approval; this agent asks before every post.
connector_tool(
connection.post_thread_reply(), connection.reference, requires_approval=True
),
]),
]
# ...
async def open_slack_project_connection(
environment: Mapping[str, str] = os.environ,
) -> SlackProjectConnection | None:
# ...
# An incomplete or invalid configuration raises here, so the Worker fails at startup.
try:
project = await load_from_environment(environment)
except FileNotFoundError:
_log_slack_not_set_up("the connector configuration file does not exist")
return None
# ...
try:
connection = slack.new_project_connection(project, SLACK_CONNECTION_NAME)
except BaseException:
await project.close()
raise
return SlackProjectConnection(project, connection)
Example: examples/python/agent-harnesses/agent_harness_examples/slack_connector_tools.py
The Worker opens the connection once, at startup, and every agent uses it:
async def main() -> None:
config = AgentHarnessExamplesConfig.from_env()
slack_setup = await open_slack_project_connection()
try:
app_state = AgentHarnessExamplesApp(config, slack_setup.connection if slack_setup else None)
# ...
finally:
if slack_setup is not None:
await slack_setup.aclose()
Example: examples/python/agent-harnesses/agent_harness_examples/__main__.py
The agent passes the tool source next to its own tools:
return claude_agent_session_flow(
build_options,
[look_up_order, cancel_order],
flow_type=CLAUDE_AGENT_SDK_ORDER_SUPPORT_FLOW_TYPE,
tool_sources=build_slack_tool_sources(slack_connection),
# ...
)
Example: examples/python/agent-harnesses/agent_harness_examples/claude_agent_sdk_order_support.py
How a call runs
Each call of slack_post_thread_reply runs once, in its own CallSingleAttemptTool Step, after a person approves it. The Slack Connector tool explains why.
In the Claude Agent SDK, a Connector tool call never runs inside the CLI, not even a read. Every call ends the segment and runs in its own tool Step, because the Connector needs that Step for its Call ID. The resumed CLI then gets the recorded result. The CLI sees the tool as mcp__tools__slack_post_thread_reply; tool policies, approvals, and records use slack_post_thread_reply.
A later write of the same reply is refused inside the turn and leaves no record; see Write order. Do not list a Connector tool in allowed_tools (allowedTools): its calls must leave the CLI, so such an entry fails the RunModel attempt.
Configure the connection
Connections explains where the Worker finds the support-slack connection and what the Slack app needs.
To try it, start the example with DEX_CONNECTOR_CONFIG_FILE set. Then ask the agent to reply in a Slack thread, with the channel ID and the thread timestamp:
Reply in Slack thread C0123456789 1710000000.000100: Order 1002 ships on Thursday.
The post waits for approval, like cancel_order.
Limits
- One write call per segment leaves the turn. A reply with two writes costs one more segment and model call. The writes of one reply never run as parallel Steps.
- Tools that run inside the CLI, such as Bash or Edit, are at least once. They run again when a segment reruns. Files the CLI changes are not recorded.
- A segment must finish within the RunModel method timeout, 15 minutes by default. Raise it with run_model_limit in Python or runModelLimit in TypeScript.
- Dex compaction is not available. Claude Code compacts its own transcript. A long session ends at the state limit, not the context window.
- A recorded tool result keeps text only.
- A write call inside a subagent is denied. The model is asked to call it from the main conversation.
The adapter README lists every limit. See Limits for long sessions for the server limits.