--- title: "把 Dex 加入你的 Agent" description: "安装你所用框架的适配器，复制完整的 Agent 文件与宿主文件，再把持久化的 Agent Session 接入一个流式返回回复的 Web 应用。" canonical_url: "https://docs.superdurable.io/zh-Hans/agent-frameworks/add-to-your-agent/" source: "docs/i18n/zh-Hans/docusaurus-plugin-content-docs/current/agent-frameworks/add-to-your-agent.mdx" locale: "zh-Hans" framework: "claude-agent-sdk" language: "python" --- > 供 AI Agent 使用的全部页面索引：https://docs.superdurable.io/zh-Hans/llms.txt # 把 Dex 加入你的 Agent 本文件为 Claude Agent SDK 版本。其他框架：[Google ADK](https://docs.superdurable.io/zh-Hans/agent-frameworks/add-to-your-agent.google-adk.md) · [LangChain and LangGraph](https://docs.superdurable.io/zh-Hans/agent-frameworks/add-to-your-agent.langchain-langgraph.md) · [Mastra](https://docs.superdurable.io/zh-Hans/agent-frameworks/add-to-your-agent.mastra.md) · [OpenAI Agents SDK](https://docs.superdurable.io/zh-Hans/agent-frameworks/add-to-your-agent.openai-agents-sdk.md) · [Pydantic AI](https://docs.superdurable.io/zh-Hans/agent-frameworks/add-to-your-agent.pydantic-ai.md) · [Vercel AI SDK](https://docs.superdurable.io/zh-Hans/agent-frameworks/add-to-your-agent.vercel-ai-sdk.md) 本页代码为 Python。其他语言：[TypeScript](https://docs.superdurable.io/zh-Hans/agent-frameworks/add-to-your-agent.typescript.md) 把 [Agent 快速开始](https://docs.superdurable.io/zh-Hans/agent-frameworks/quick-start.md) 中的示例 Agent 移到你自己的代码中：安装框架的适配器，复制完整文件，再在 Web 应用中对外提供 Session。 > **用编程 Agent 来做** > > 先安装一次 Dex Skills： > > Claude Code: > > ```bash > claude plugin marketplace add superdurable/dex-skills > claude plugin install superdurable-dex@superdurable > ``` > > Codex: > > ```bash > codex plugin marketplace add superdurable/dex-skills > codex plugin add superdurable-dex@superdurable > ``` > > 然后粘贴这段提示词： > > ```text > 用 Dex 让 中的 Agent 变得持久。 > > 1. 阅读 https://docs.superdurable.io/zh-Hans/agent-frameworks/add-to-your-agent.md 和 https://docs.superdurable.io/zh-Hans/agent-frameworks/framework-support.md，把它们当作唯一依据。 > 2. 用对应框架的 Dex Agents 适配器包装这个 Agent，并给每个 tool 指定策略：read、idempotent_write 或 non_idempotent_write。 > 3. 把 Session 接入我的 Web 应用：启动 Session、发送消息、流式返回回复。 > 4. 在 dexcli dev 上运行，向我展示一个带 Tool Call 的 turn。 > > 规则：无法判断类别的 tool 先问我。不要编造 API。 > ``` ## 你的应用会有哪些变化 - **Agent 定义保持原样。** 你把它包装一层，并为每个 tool 指定一个策略。 - **请求处理函数（handler）的形态会变。** 它把用户消息发给 Session 后就返回。回复出现在 Session 的事件流上，或者在它的快照中。 - **你要多运行两样东西。** Worker 进程运行 Session 的 Step，Dex Server 存储它们。Dex Server 可以运行在 Dex Cloud 上，也可以运行在你自己的基础设施上；[选择 Dex 的运行方式](https://docs.superdurable.io/zh-Hans/components/choose-how-to-run.md) 对比了两者。Dex 处于 Beta 阶段，Dex Server 的版本是 0.1，一个集群由一个活动节点提供服务。 - **每个 turn 都要等待 Dex 提交。** 一个带三个并行 Tool Call 的 turn 大约多花 0.4 秒；参见 [延迟与成本](https://docs.superdurable.io/zh-Hans/agent-frameworks/sessions.md#latency-and-cost)。 [在生产环境运行 Agent](https://docs.superdurable.io/zh-Hans/agent-frameworks/production.md) 列出了需要准备的事项。[性能](https://docs.superdurable.io/zh-Hans/intro/performance.md) 给出具体数字，[限制](https://docs.superdurable.io/zh-Hans/intro/limits.md#license) 给出许可证说明。 ## 你的代码有哪些变化 示例先按框架本身的写法写出 Agent，再把它包进 Dex Agent Session，并为每个 tool 指定 policy。对这两个 function tool 来说，改动只有这些： **Python** ```python def build_plain_agent(order_desk: OrderDesk, model: str) -> ClaudeAgentOptions: """Returns the agent's options as the Claude Agent SDK alone writes them.""" orders = create_sdk_mcp_server("orders", tools=build_order_tools(order_desk)) return ClaudeAgentOptions( model=model, system_prompt=ORDER_SUPPORT_INSTRUCTIONS, mcp_servers={"orders": orders}, allowed_tools=["mcp__orders__look_up_order", "mcp__orders__cancel_order"], ) def build_durable_agent(order_desk: OrderDesk, model: str) -> AgentSessionFlow: """Returns the same agent as a durable Dex agent session.""" # Dex passes the tools to the CLI itself and decides when each one runs. def build_options(context: ClaudeAgentOptionsContext) -> ClaudeAgentOptions: return ClaudeAgentOptions( model=model, env=build_claude_cli_environment(os.environ), system_prompt=ORDER_SUPPORT_INSTRUCTIONS, tools=[], setting_sources=[], ) return claude_agent_session_flow( build_options, build_order_tools(order_desk), flow_type="ClaudeAgentSDKQuickStart", tool_policies={ "look_up_order": ToolPolicy(ToolEffect.READ), "cancel_order": ToolPolicy(ToolEffect.NON_IDEMPOTENT_WRITE), }, ) ``` 例子: [examples/python/agent-harnesses/agent_harness_examples/quick_start/claude_agent_sdk_quick_start.py](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses/agent_harness_examples/quick_start/claude_agent_sdk_quick_start.py) Claude Agent SDK 在 options 旁边以列表形式接收 tool，因此由 Dex 自己运行每个 tool。 由框架自己运行的 tool，例如 MCP server 和 hosted tool，需要按你所用框架的 [限制](https://docs.superdurable.io/zh-Hans/agent-frameworks/framework-support.md#limits) 做相应修改。Session 在 Worker 中运行，由 Client 驱动；[把 Dex 加入你的 Agent](https://docs.superdurable.io/zh-Hans/agent-frameworks/add-to-your-agent.md#wire-it-into-your-app) 展示了这部分宿主代码。 ## 在你自己的 Agent 中使用 把适配器和它对应的框架一起安装。每个适配器都会引入 Dex Agents core。然后复制下面的 [完整文件](https://docs.superdurable.io/zh-Hans/agent-frameworks/add-to-your-agent.md#complete-files)。 **Python** ```bash pip install dex-agents-claude-agent-sdk ``` 例子: [agents/python/claude-agent-sdk](https://github.com/superdurable/dex/tree/main/agents/python/claude-agent-sdk) **认证** Worker 用 `ANTHROPIC_API_KEY` 为 Claude Code CLI 认证。把它设置在 Worker 的环境中。它永远不会进入 Flow 状态。 示例为 CLI 构建的环境中，`ANTHROPIC_API_KEY` 是唯一的凭据。缺少这个 key 时，示例不会注册 Claude Agent，只提供其他 harness。 ### 完整文件 Agent 快速开始程序每种语言有三个文件。除了示例自带的 order desk 及其 tool，以及入口读取的设置之外，它们只用到 Dex、Dex Agents 和你的框架。在仓库中阅读它们： | 文件 | Python | TypeScript | | ------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | 你的 Agent 模块，包含普通版和持久化版，以及全部 import | [claude_agent_sdk](https://github.com/superdurable/dex/blob/main/examples/python/agent-harnesses/agent_harness_examples/quick_start/claude_agent_sdk_quick_start.py) · [google_adk](https://github.com/superdurable/dex/blob/main/examples/python/agent-harnesses/agent_harness_examples/quick_start/google_adk_quick_start.py) · [langchain](https://github.com/superdurable/dex/blob/main/examples/python/agent-harnesses/agent_harness_examples/quick_start/langchain_quick_start.py) · [openai_agents](https://github.com/superdurable/dex/blob/main/examples/python/agent-harnesses/agent_harness_examples/quick_start/openai_agents_quick_start.py) · [pydantic_ai](https://github.com/superdurable/dex/blob/main/examples/python/agent-harnesses/agent_harness_examples/quick_start/pydantic_ai_quick_start.py) | [claude-agent-sdk](https://github.com/superdurable/dex/blob/main/examples/typescript/src/products/agent-harnesses/quick-start/claude-agent-sdk-quick-start.ts) · [mastra](https://github.com/superdurable/dex/blob/main/examples/typescript/src/products/agent-harnesses/quick-start/mastra-quick-start.ts) · [openai-agents](https://github.com/superdurable/dex/blob/main/examples/typescript/src/products/agent-harnesses/quick-start/openai-agents-quick-start.ts) · [vercel-ai](https://github.com/superdurable/dex/blob/main/examples/typescript/src/products/agent-harnesses/quick-start/vercel-ai-quick-start.ts) | | Runner：Registry、Worker、Client，以及一个带审批的 turn | [session.py](https://github.com/superdurable/dex/blob/main/examples/python/agent-harnesses/agent_harness_examples/quick_start/session.py) | [quick-start-session.ts](https://github.com/superdurable/dex/blob/main/examples/typescript/src/products/agent-harnesses/quick-start/quick-start-session.ts) | | 在命令行提示中请求审批的入口 | [\_\_main\_\_.py](https://github.com/superdurable/dex/blob/main/examples/python/agent-harnesses/agent_harness_examples/quick_start/__main__.py) | [quick-start-main.ts](https://github.com/superdurable/dex/blob/main/examples/typescript/src/products/agent-harnesses/quick-start/quick-start-main.ts) | ### 把 Session 接入你的应用 对 Agent 本身的改动很小：就是 [Agent 快速开始第 2 步](https://docs.superdurable.io/zh-Hans/agent-frameworks/add-to-your-agent.md#what-changes-in-your-code) 中的那层包装。其余是每个 Dex 应用都有的宿主代码：一个运行 Step 的 Worker，和一个调用它们的 Client。下面这个函数是最小的宿主。它在你的进程中运行一个 Worker 和一个 Client，启动一个 Session，发送一条消息，并在 turn 结束后返回 Session： **Python** ```python """The smallest host for a durable agent: one Worker, one Client, one turn.""" from __future__ import annotations import asyncio from dex import ( AsyncClient, AsyncWorker, BlobCacheConfig, ClientOptions, Registry, WorkerOptions, open_blob_cache, ) from dex_agents import AgentSessionClient, AgentSessionFlow, AgentSessionStatus, AgentSnapshot async def ask_agent( session_flow: AgentSessionFlow, session_id: str, message: str, server_address: str = "127.0.0.1:8801", blob_cache_dir: str = ".dex-blob-cache", ) -> AgentSnapshot: """Starts a session, sends one message, and returns the session after the turn.""" registry = Registry((session_flow,), allow_async_handlers=True) cache = open_blob_cache(BlobCacheConfig(blob_cache_dir, 1 << 30)) worker = AsyncWorker(registry, cache, WorkerOptions(server_address=server_address)) client = AsyncClient(registry, cache, ClientOptions(server_address=server_address)) worker_task = asyncio.create_task(worker.start()) try: await worker.wait_pull_registered(timeout=30) sessions = AgentSessionClient(client, session_flow) await sessions.start_session(session_id) await sessions.send_user_message(session_id, message) while True: snapshot = await sessions.get_agent_snapshot(session_id) state = snapshot.state if state and state.turn_count >= 1: if state.status == AgentSessionStatus.WAITING_FOR_USER_MESSAGE: return snapshot if state and state.status == AgentSessionStatus.FAILED: raise RuntimeError(f"the session failed: {state.last_turn_error}") await asyncio.sleep(0.5) finally: await client.close() await worker.close() await worker_task cache.close() ``` 例子: [examples/python/agent-harnesses/agent_harness_examples/quick_start/embed.py](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses/agent_harness_examples/quick_start/embed.py) 它接收你的模块中 `build_durable_agent` 或 `buildDurableAgent` 返回的 Session Flow。需要审批的 turn 会一直等待，直到你的应用处理它；做法参见 [在你的应用中审批](https://docs.superdurable.io/zh-Hans/agent-frameworks/approvals.md#approve-from-your-application)。 Web 应用把同样的部件分到各个路由中： 1. 一个包含 Session Flow 的 **Registry**、一个运行其 Step 的 **Worker**，以及一个调用它的 **Client**，和上面的函数一样。Worker 主动向外连接 Dex，所以不需要开放端口。两者都会打开一个 blob cache，即存放较大值的本地缓存；[Registry 与 blob cache](https://docs.superdurable.io/zh-Hans/components/application.md#blob-cache) 对它做了说明。 2. 一个启动 Session 的路由。Session ID 就是 Flow ID。 3. 一个发送用户消息的路由，以及一个读取 Session 快照（其中包含 Agent 回复）的路由。要在回复生成的同时流式返回，请读取 AgentEvents Stream，如 [下一节](https://docs.superdurable.io/zh-Hans/agent-frameworks/add-to-your-agent.md#web-handler) 所示。 4. 一个批准或拒绝等待中 Tool Call 的路由。参见 [在你的应用中审批](https://docs.superdurable.io/zh-Hans/agent-frameworks/approvals.md#approve-from-your-application)。 [订单客服 Agent](https://docs.superdurable.io/zh-Hans/use-cases/order-support-agent.md) 示例为每个框架都提供了这些 HTTP 路由。 #### Web handler 中由什么代替框架调用 Handler 把消息排入 Session 的队列，然后返回。 它代替了原来等待回复的 `query` 调用。 回复出现在 Session 的 [AgentEvents Stream](https://docs.superdurable.io/zh-Hans/reference/agents/event-stream.md#read-the-stream)（英文）上，UI 以 server-sent events 的形式读取它。上面的函数轮询 Session 快照，适合脚本和测试。Chat UI 则读取这个 stream。 **Python** 这个完整的应用托管 Worker 和 Client，并提供 chat UI 需要的所有路由：启动 Session、发送消息、读取 Session、流式读取它的 event，以及答复审批。完整模块 `quick_start/web.py`： ```python """A complete web app for a durable agent: start, message, read, stream, approve. It hosts the Worker and the Client in the web process. Add your own authentication before you expose these routes: anyone who knows a session ID can read and change that session here. """ from __future__ import annotations import uuid from pathlib import Path from typing import Any from dex_agents import AgentSessionFlow from dex_agents.ui_stream import ( AGENT_EVENT_STREAM_HEADERS, agent_snapshot_to_json, stream_agent_event_frames, ) from quart import Quart, Response, jsonify, make_response, request from agent_harness_examples.quick_start.safe_send import send_user_message_once from agent_harness_examples.quick_start.session import QuickStartApp def create_web_app( session_flow: AgentSessionFlow, server_address: str, blob_cache_dir: Path ) -> Quart: """Returns a Quart app whose routes run one durable agent's sessions.""" web = Quart(__name__) apps: list[QuickStartApp] = [] def dex() -> QuickStartApp: return apps[0] # The Worker and the Client bind to the event loop that serves the app. @web.before_serving async def start_worker() -> None: apps.append(QuickStartApp(session_flow, server_address, blob_cache_dir)) await dex().start_worker() @web.after_serving async def stop_worker() -> None: await apps.pop().close() @web.post("/sessions") async def start_session() -> Response: session_id = f"web-{uuid.uuid4().hex}" await dex().sessions.start_session(session_id) return jsonify(session_id=session_id) @web.post("/sessions//messages") async def send_message(session_id: str) -> Response: body: dict[str, Any] = await request.get_json() await send_user_message_once(dex().sessions, session_id, str(body["text"])) return jsonify(accepted=True) @web.get("/sessions/") async def read_session(session_id: str) -> Response: return jsonify(agent_snapshot_to_json(await dex().sessions.get_agent_snapshot(session_id))) @web.get("/sessions//events") async def stream_events(session_id: str) -> Response: resume_token = request.headers.get("Last-Event-ID", "") frames = await stream_agent_event_frames( dex().client, session_flow, session_id, resume_token=resume_token ) response = await make_response(frames, dict(AGENT_EVENT_STREAM_HEADERS)) assert isinstance(response, Response) response.timeout = None return response @web.post("/sessions//interrupts/") async def resolve(session_id: str, interrupt_id: str) -> Response: body: dict[str, Any] = await request.get_json() accepted = await dex().sessions.resolve_interrupt( session_id, interrupt_id, approved=body.get("approved") is True ) return jsonify(accepted=accepted) return web ``` 例子: [examples/python/agent-harnesses/agent_harness_examples/quick_start/web.py](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses/agent_harness_examples/quick_start/web.py) ## 下一步 [Session](https://docs.superdurable.io/zh-Hans/agent-frameworks/sessions.md)