--- title: "订单客服 Agent" description: "一个带两个 tool 的订单客服 Agent，为每个 Agent 框架各实现了一次。选择你的框架，下面每一节都会按这个选择显示。" canonical_url: "https://docs.superdurable.io/zh-Hans/use-cases/order-support-agent/" source: "docs/i18n/zh-Hans/docusaurus-plugin-content-docs/current/use-cases/order-support-agent.mdx" locale: "zh-Hans" framework: "claude-agent-sdk" language: "python" --- > 供 AI Agent 使用的全部页面索引：https://docs.superdurable.io/zh-Hans/llms.txt # 订单客服 Agent 本文件为 Claude Agent SDK 版本。其他框架：[Google ADK](https://docs.superdurable.io/zh-Hans/use-cases/order-support-agent.google-adk.md) · [LangChain and LangGraph](https://docs.superdurable.io/zh-Hans/use-cases/order-support-agent.langchain-langgraph.md) · [Mastra](https://docs.superdurable.io/zh-Hans/use-cases/order-support-agent.mastra.md) · [OpenAI Agents SDK](https://docs.superdurable.io/zh-Hans/use-cases/order-support-agent.openai-agents-sdk.md) · [Pydantic AI](https://docs.superdurable.io/zh-Hans/use-cases/order-support-agent.pydantic-ai.md) · [Vercel AI SDK](https://docs.superdurable.io/zh-Hans/use-cases/order-support-agent.vercel-ai-sdk.md) 本页代码为 Python。其他语言：[TypeScript](https://docs.superdurable.io/zh-Hans/use-cases/order-support-agent.typescript.md) 一个带两个 tool 的订单客服 Agent，为每个 Agent 框架各实现了一次。选择你的框架，下面每一节都会按这个选择显示。 **SDK：** Python · TypeScript 每个框架使用同一个订单客服 Agent。它有两个 tool，作用于一个内存中的 order desk： | Tool | Policy | Dex 做什么 | | --------------- | ------ | ----------------------------------- | | `look_up_order` | 读 | 无需审批即可运行，失败时重试。 | | `cancel_order` | 非幂等写 | 在 `AwaitToolApproval` 中等待，然后最多运行一次。 | `cancel_order` 只在订单仍处于处理中时才能取消它。 Python Agent 位于 [examples/python/agent-harnesses](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses)。TypeScript Agent 位于 [examples/typescript/src/products/agent-harnesses](https://github.com/superdurable/dex/tree/main/examples/typescript/src/products/agent-harnesses)。 ## Agent 和它的 tool Python 中的 Flow type 是 `ClaudeAgentSDKOrderSupport`，TypeScript 中是 `ClaudeAgentSdkOrderSupportAgent`。完整示例位于 [examples/python/agent-harnesses](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses) 和 [examples/typescript/src/products/agent-harnesses](https://github.com/superdurable/dex/tree/main/examples/typescript/src/products/agent-harnesses)。 这些 tool 是 Claude Agent SDK tool。options 给出模型、system prompt 和 CLI 环境。tool policy 用每个 tool 自己的名字指定它。 **Python** ```python 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](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses/agent_harness_examples/claude_agent_sdk_order_support.py) 这个示例关闭了 Claude Code 的所有内置 tool，所以 Agent 只使用订单 tool。它也不加载任何 settings 文件，所以 settings 文件无法抢在 Dex 请求审批之前批准 tool。 ## Worker 一个 Worker 为示例中所有 harness 的 Session Flow 提供服务。在 Python 中，Agent Session 的 Step 是 async 的，所以 Registry 允许 async handler。每个启动 Session 或调用其 RPC 的进程，都要构建同一个 Session Flow。Registry、Worker 和 Client 的设置对每个框架都一样；每个标签页只展示该框架的 Session Flow 如何构建和注册。 **Python** ```python # 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) ) ``` 例子: [examples/python/agent-harnesses/agent_harness_examples/app.py](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses/agent_harness_examples/app.py) **Python** ```python 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](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses/agent_harness_examples/app.py) [把 Session 接入你的应用](https://docs.superdurable.io/zh-Hans/agent-frameworks/add-to-your-agent.md#wire-it-into-your-app)列出了你自己的应用需要的各个部分，每个部分一行说明。这段代码中的 blob cache 是存放大型 Flow 值的本地缓存；[Registry 和 blob cache](https://docs.superdurable.io/zh-Hans/components/application.md#blob-cache)介绍了它。 ## Session 路由 每个 harness 提供相同的 HTTP 路由，你可以把它们放在一起对比试用。每个路由对应 Session Flow 上的一次调用：启动 Session、发送消息，或处理一次审批。 路由内部的调用是 Dex Agents API，在你的应用中也是同样的调用： | 路由做什么 | Python 中的 Dex Agents 调用 | TypeScript 中的 Dex Agents 调用 | | --------------- | ---------------------------------------------------------------------------- | ---------------------------------------------- | | 启动 Session | `AgentSessionClient.start_session` | `startAgentSession` | | 发送用户消息 | `AgentSessionClient.send_user_message` | `sendUserMessage` RPC，或 `sendAgentUserMessage` | | 读取 Session | `read_agent_event_stream_snapshot` 或 `AgentSessionClient.get_agent_snapshot` | `getAgentSnapshot` RPC，或 `readAgentSnapshot` | | 批准或拒绝 Tool Call | `AgentSessionClient.resolve_interrupt` | `resolveInterrupt` RPC | 其余部分都属于示例本身：URL 路径、JSON 结构、Session token 和错误码。Python 和 TypeScript 示例在这些地方的设计略有不同。请围绕上面的调用设计你自己的路由。 启动 Session 会返回一个 Session token：用 `DEX_AGENT_SESSION_KEY` 对 Session ID 计算的 HMAC。其他每个路由都会检查 **X-Agent-Session-Token** header 中的 token，缺少 token 时返回 403。`DEX_AGENT_SESSION_KEY` 未设置时，示例会随机选一个 key，所以重启后所有 token 都会失效。真实的应用应改用自己的登录机制。 每个 harness 都提供这些相同的路由。[Session 路由](https://docs.superdurable.io/zh-Hans/use-cases/order-support-agent.md#session-routes)介绍了这些路由和 Session token。 **Python** ```python @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](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses/agent_harness_examples/http_routes.py) ## 在本地运行 在第一个终端里渲染 Session 图，然后启动 Dex。Dex Web 在启动时读取这张图，用来显示 Session 和它的审批控件。 **Python** ```bash 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](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses) **dexcli dev** 会一直在前台运行。在第二个终端里启动示例： **Python** ```bash cd examples/python/agent-harnesses ANTHROPIC_API_KEY=... uv run --frozen python -m agent_harness_examples ``` 例子: [examples/python/agent-harnesses](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses) Python 应用监听 **127.0.0.1:8081**，TypeScript 应用监听 **127.0.0.1:8080**。Dex Web 位于 `http://127.0.0.1:8802`。 | 变量 | 默认值 | 含义 | | ---------------------------- | ------------------- | ------------------------------------------------------- | | `ANTHROPIC_API_KEY` | 无；必填 | Worker 使用的 Anthropic API key。 | | `DEX_CLAUDE_AGENT_SDK_MODEL` | **claude-opus-5-5** | Claude 模型名称。 | | `DEX_AGENT_SESSION_KEY` | 随机 key | 用于签名 Session token 的 key。使用随机 key 时，进程重启后所有 token 都会失效。 | 启动一个 Session，让 Agent 取消订单 1002，然后批准这次调用： **Python** ```bash 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](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses) 循环会一直读取 snapshot，直到 Agent 请求运行 `cancel_order`。这次调用先等待审批，然后运行一次。再读一次 snapshot，就能看到 Agent 的回答。 ## 下一步 - [Agent 快速开始](https://docs.superdurable.io/zh-Hans/agent-frameworks/quick-start.md) - [审批](https://docs.superdurable.io/zh-Hans/agent-frameworks/approvals.md) - [把 Connector 用作 tool](https://docs.superdurable.io/zh-Hans/agent-frameworks/connectors-as-tools.md)