跳到主要内容
Super Durable
简体中文
English简体中文
GitHub登录注册 Cloud
菜单

Claude Agent SDK

查看 Markdown

Claude Agent SDK 把 Claude Code 作为 Agent 运行。它的 agent loop 运行在 Claude Code CLI 中，CLI 是 Worker 的子进程。Dex adapter 把这个 loop 放进 durable session 中运行。它支持 Python 和 TypeScript。

何时使用​

需要 Claude Code 的 agent loop、上下文管理和它的 tool 时，使用它。Agent 已经用 Claude Agent SDK 编写时，也使用它。

如果需要每个 Step 只调用一次模型、并行的 tool Step 或 Dex compaction，请使用 内置 agent loop。

Session 如何运行​

Claude Code 一次运行整个 turn，因此 adapter 把一个 turn 拆成多个 segment。一个 RunModel Step 为一个 segment 运行 CLI。模型调用在 segment 内运行。Read tool 也在 segment 内运行：把 read 移出 segment 会让 CLI 重新启动。Connector tool call 是例外：它总是离开 segment，读操作也一样。参见 连接 Connector。

写操作的 tool call 会结束当前 segment。需要审批时，它在 AwaitToolApproval 中等待。然后它在自己的 Step 中运行：idempotent write 用 CallTool，其他写操作用 CallSingleAttemptTool。之后 ResumeTurn 回到 RunModel，由它恢复同一个 Claude session。CLI 拿到记录下来的结果，不会再次运行这个 tool。

失败的 segment 会重试，最多三次，并且整体重跑。其中的模型调用和 read 会再次运行。已经在自己的 Step 中运行过的写操作不会再次运行。

查询订单的 turn 只运行一个 RunModel Step。取消订单的 turn 依次运行 RunModel、AwaitToolApproval、CallSingleAttemptTool、ResumeTurn 和 RunModel。

认证​

Worker 用 ANTHROPIC_API_KEY 为 Claude Code CLI 认证。把它设置在 Worker 的环境中。它永远不会进入 Flow state。

Example 为 CLI 构建的环境中，ANTHROPIC_API_KEY 是唯一的 credential。没有这个 key 时，example 不会注册 Claude Agent，但仍会提供其他 harness。

安装​

把 adapter 与它的 framework 一起安装。每个 adapter 都会带上 Dex Agents core。

pip install dex-agents-claude-agent-sdk

例子: agents/python/claude-agent-sdk/README.md

Example​

这个 example 就是每个 framework 页面共用的订单客服 Agent。

Flow type 是 ClaudeAgentSDKOrderSupport（Python），ClaudeAgentSdkOrderSupportAgent（TypeScript）。完整 example 位于 examples/python/agent-harnesses 和 examples/typescript/src/products/agent-harnesses。

Agent 与 tool​

Tool 是 Claude Agent SDK tool。Options 提供模型、system prompt 和 CLI 环境。Tool policy 用每个 tool 自己的名称指定 tool。

claude_agent_sdk_order_support.pyPython
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),
        },
    )

例子: examples/python/agent-harnesses/agent_harness_examples/claude_agent_sdk_order_support.py

Example 关闭了所有 Claude Code 内置 tool，因此 Agent 只使用订单 tool。它也不加载任何 settings 文件，因此不会有 settings 文件在 Dex 请求审批之前就批准某个 tool。

Worker​

一个 Worker 为 example 中所有 harness 的 session Flow 提供服务。Agent session 的 Step 是 async 的，因此 Python Registry 允许 async handler。每个启动 session 或调用其 RPC 的进程，都要构建同一个 session Flow。

app.pyPython
        # 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)

例子: examples/python/agent-harnesses/agent_harness_examples/app.py

Session route​

每个 harness 都提供这同一组 route。这些 route 和 session token 的说明见 Session route。

http_routes.pyPython
    @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())

例子: examples/python/agent-harnesses/agent_harness_examples/http_routes.py

在本地运行​

启动 Dex 与 example​

在第一个终端中，先渲染 session graph，再启动 Dex。Dex Web 启动时读取这个 graph，用它显示 session 和审批控件。

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

例子: examples/python/agent-harnesses/README.md

dexcli dev 会一直在前台运行。在第二个终端中启动 example：

cd examples/python/agent-harnesses
ANTHROPIC_API_KEY=... uv run --frozen python -m agent_harness_examples

例子: examples/python/agent-harnesses/README.md

Python 应用监听 127.0.0.1:8081。TypeScript 应用监听 127.0.0.1:8080。Dex Web 位于 127.0.0.1:8802。

变量默认值含义
ANTHROPIC_API_KEY无；必填Worker 的 Anthropic API key。
DEX_CLAUDE_AGENT_SDK_MODELclaude-opus-5-5Claude 模型名称。
DEX_AGENT_SESSION_KEY随机 key签发 session token 的 key。使用随机 key 时，进程重启后所有 token 都会失效。

调用 session​

启动一个 session，让 Agent 取消订单 1002，然后批准这次调用：

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}"

例子: examples/python/agent-harnesses/README.md

循环会反复读取 snapshot，直到 Agent 请求运行 cancel_order。这次调用先等待审批，批准后只运行一次。再读一次 snapshot 即可看到 Agent 的回答。

在 Dex Web 中审批​

也可以不执行最后一条命令，改在 Dex Web 中审批，步骤见 在 Dex Web 中审批。

连接 Connector​

Agent 可以把 Connector 的 operation 当作 tool 调用，见 把 Connector 用作 tool。这个 example 加入了 Slack tool slack_post_thread_reply，用来在 Slack thread 中回复。

提供 tool​

在 Python 中，安装 dex-agents 的 connectors extra，以及 Connector 包，例如 dex-connector-slack。在 TypeScript 中，安装 @superdurable/connectors-sdk 以及 Connector 包，例如 @superdurable/connector-slack。安装命令见 Connector tools。

Worker 从项目配置中打开 Slack connection support-slack。Python 的 connector_tool 或 TypeScript 的 connectorTool 把 operation 变成 tool，由 ConnectorToolSource 持有。没有这个 connection 时，agent 不会得到 Slack tool。

slack_connector_tools.pyPython
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)

例子: examples/python/agent-harnesses/agent_harness_examples/slack_connector_tools.py

Worker 在启动时打开一次 connection，每个 agent 都使用它：

__main__.pyPython
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()

例子: examples/python/agent-harnesses/agent_harness_examples/__main__.py

Agent 把 tool source 和自己的 tool 一起传入：

claude_agent_sdk_order_support.pyPython
    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),
        # ...
    )

例子: examples/python/agent-harnesses/agent_harness_examples/claude_agent_sdk_order_support.py

调用如何运行​

每次调用 slack_post_thread_reply 都要先经人批准，然后在自己的 CallSingleAttemptTool Step 中只运行一次。原因见 Slack Connector tool。

在 Claude Agent SDK 中，Connector tool call 从不在 CLI 内运行，读操作也一样。每次调用都会结束当前 segment，并在自己的 tool Step 中运行，因为 Connector 需要这个 Step 来得到它的 Call ID。之后恢复的 CLI 拿到记录下来的结果。CLI 看到的 tool 名是 mcp__tools__slack_post_thread_reply；tool policy、审批和记录使用 slack_post_thread_reply。

同一次回复中后面的写操作会在 turn 内被拒绝，不留下记录，见 写操作顺序。不要把 Connector tool 列进 allowed_tools（allowedTools）：它的调用必须离开 CLI，所以这样的条目会让 RunModel 这次尝试失败。

配置 connection​

Worker 从哪里找到 connection support-slack、Slack app 需要什么，见 Connection。

试用时，先设置 DEX_CONNECTOR_CONFIG_FILE 再启动 example。然后让 agent 在 Slack thread 中回复，并给出 channel ID 和 thread timestamp：

Reply in Slack thread C0123456789 1710000000.000100: Order 1002 ships on Thursday.

这次发帖和 cancel_order 一样，会先等待审批。

限制​

  • 每个 segment 只有一个写操作调用会离开 turn。一次回复中有两个写操作时，会多花一个 segment 和一次模型调用。同一次回复中的写操作永远不会作为并行 Step 运行。
  • 在 CLI 内运行的 tool，例如 Bash 或 Edit，至少运行一次。segment 重跑时它们会再次运行。CLI 修改的文件不会被记录。
  • 一个 segment 必须在 RunModel 的 method timeout 内完成，默认 15 分钟。在 Python 中用 run_model_limit、在 TypeScript 中用 runModelLimit 调大它。
  • 不支持 Dex compaction。Claude Code 自己压缩它的 transcript。长 session 会在达到 state 上限时结束，而不是在达到 context window 时结束。
  • 记录下来的 tool result 只保留文本。
  • Subagent 中的写操作调用会被拒绝。模型会被要求从主对话中调用它。

Adapter README 列出了全部限制。Server 的限制参见 Limits for long sessions。