跳到主要内容
Super Durable
菜单

把 Dex 加入你的 Agent

查看 Markdown

把 Agent 快速开始 中的示例 Agent 移到你自己的代码中：安装框架的适配器，复制完整文件，再在 Web 应用中对外提供 Session。

你的应用会有哪些变化

  • Agent 定义保持原样。 你把它包装一层，并为每个 tool 指定一个策略。
  • 请求处理函数（handler）的形态会变。 它把用户消息发给 Session 后就返回。回复出现在 Session 的事件流上，或者在它的快照中。
  • 你要多运行两样东西。 Worker 进程运行 Session 的 Step，Dex Server 存储它们。Dex Server 可以运行在 Dex Cloud 上，也可以运行在你自己的基础设施上；选择 Dex 的运行方式 对比了两者。Dex 处于 Beta 阶段，Dex Server 的版本是 0.1，一个集群由一个活动节点提供服务。
  • 每个 turn 都要等待 Dex 提交。 一个带三个并行 Tool Call 的 turn 大约多花 0.4 秒；参见 延迟与成本。

在生产环境运行 Agent 列出了需要准备的事项。性能 给出具体数字，限制 给出许可证说明。

你的代码有哪些变化

示例先按框架本身的写法写出 Agent，再把它包进 Dex Agent Session，并为每个 tool 指定 policy。对这两个 function tool 来说，改动只有这些：

claude_agent_sdk_quick_start.pyPython
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

Claude Agent SDK 在 options 旁边以列表形式接收 tool，因此由 Dex 自己运行每个 tool。

由框架自己运行的 tool，例如 MCP server 和 hosted tool，需要按你所用框架的 限制 做相应修改。Session 在 Worker 中运行，由 Client 驱动；把 Dex 加入你的 Agent 展示了这部分宿主代码。

在你自己的 Agent 中使用

把适配器和它对应的框架一起安装。每个适配器都会引入 Dex Agents core。然后复制下面的 完整文件。

pip install dex-agents-claude-agent-sdk

例子: 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 和你的框架。在仓库中阅读它们：

文件PythonTypeScript
你的 Agent 模块，包含普通版和持久化版，以及全部 importclaude_agent_sdk · google_adk · langchain · openai_agents · pydantic_aiclaude-agent-sdk · mastra · openai-agents · vercel-ai
Runner：Registry、Worker、Client，以及一个带审批的 turnsession.pyquick-start-session.ts
在命令行提示中请求审批的入口__main__.pyquick-start-main.ts

把 Session 接入你的应用

对 Agent 本身的改动很小：就是 Agent 快速开始第 2 步 中的那层包装。其余是每个 Dex 应用都有的宿主代码：一个运行 Step 的 Worker，和一个调用它们的 Client。下面这个函数是最小的宿主。它在你的进程中运行一个 Worker 和一个 Client，启动一个 Session，发送一条消息，并在 turn 结束后返回 Session：

embed.pyPython
"""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

它接收你的模块中 build_durable_agent 或 buildDurableAgent 返回的 Session Flow。需要审批的 turn 会一直等待，直到你的应用处理它；做法参见 在你的应用中审批。

Web 应用把同样的部件分到各个路由中：

  1. 一个包含 Session Flow 的 Registry、一个运行其 Step 的 Worker，以及一个调用它的 Client，和上面的函数一样。Worker 主动向外连接 Dex，所以不需要开放端口。两者都会打开一个 blob cache，即存放较大值的本地缓存；Registry 与 blob cache 对它做了说明。
  2. 一个启动 Session 的路由。Session ID 就是 Flow ID。
  3. 一个发送用户消息的路由，以及一个读取 Session 快照（其中包含 Agent 回复）的路由。要在回复生成的同时流式返回，请读取 AgentEvents Stream，如 下一节 所示。
  4. 一个批准或拒绝等待中 Tool Call 的路由。参见 在你的应用中审批。

订单客服 Agent 示例为每个框架都提供了这些 HTTP 路由。

Web handler 中由什么代替框架调用

Handler 把消息排入 Session 的队列，然后返回。

它代替了原来等待回复的 query 调用。

回复出现在 Session 的 AgentEvents Stream（英文）上，UI 以 server-sent events 的形式读取它。上面的函数轮询 Session 快照，适合脚本和测试。Chat UI 则读取这个 stream。

这个完整的应用托管 Worker 和 Client，并提供 chat UI 需要的所有路由：启动 Session、发送消息、读取 Session、流式读取它的 event，以及答复审批。完整模块 quick_start/web.py：

web.pyPython
"""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] = []

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

下一步

Session