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

内置 Agent Loop

查看 Markdown

内置 agent loop 不需要任何 agent framework 就能运行 Agent。会话由 Dex 管理，每次模型输入都从 durable state 构建。因此 Agent 不依赖 LLM provider 的 conversation API。

何时使用​

新的 Agent 使用内置 loop。它提供 session 的全部功能：提问、durable wait、plan、steering、compaction 和 MCP tool。它还允许每个 session 从 Worker 允许的模型列表中选择自己的模型。

如果已经有用 framework 写好的 Agent，请使用 framework harness。

Session 如何运行​

每次模型调用都是一个 RunModel Step。每次 tool call 都是独立的 Step，因此崩溃最多只会重复当时正在运行的那一次调用。

Step、Attribute、Channel、RPC 与 event 都与其他 Dex Agents session 相同：

StartAgentSession -> AwaitUserMessage -> RunModel
AwaitUserMessage -> steer or cancel_turn -> ApplyTurnControl
RunModel -> final answer -> AwaitUserMessage
RunModel -> failed after three attempts -> RecordTurnFailure -> AwaitUserMessage
RunModel -> CompactContext -> RunModel
RunModel -> tool calls -> AwaitToolApproval -> CallTool or CallSingleAttemptTool
                          CallTool or CallSingleAttemptTool
                          AwaitQuestionAnswer      for ask_user
                          AwaitTimer               for durable_wait
                          ResumeTurn -> RunModel, or ApplyTurnControl when steered
ApplyTurnControl -> RunModel with the steered messages, else AwaitUserMessage

这里是简化视图。Dex Agents README 列出了每个 Step 及其重试与 failure route。

失败的 RunModel 会重试。第三次尝试失败后，这个 turn 失败，session 等待下一条消息。Read 或 idempotent write 在 CallTool 中重试。其他写操作在 CallSingleAttemptTool 中只运行一次；如果 policy 要求审批，会在审批之后运行。

安装​

在 TypeScript 中，Dex SDK 是 peer package，因此也要一起安装。

pip install "dex-agents[litellm,mcp]"

例子: agents/python/core/README.md

litellm extra 和 ai package 提供 model client。mcp extra 和 @modelcontextprotocol/sdk 提供 MCP tool。不用的可以不装。

Example​

AI Agent 是一个长期运行的通用 Agent，有 Python 与 TypeScript 两个版本。它与用户聊天并调用受信任的 MCP tool。它在修改外部系统前等待审批、向用户提问、通过 durable Timer 暂停，并在执行前先制定计划供用户审阅。

Session Flow​

Application 用 Dex 的概念声明 Agent，并构建它的 Flow。Flow type 是 AIAgentSession。一次 Flow run 就是一个 session，Flow ID 就是 session ID。

ai_agent_session.pyPython
def build_ai_agent_session_flow(
    models: Sequence[str],
    system_prompt: str,
    context_window_tokens: int,
    mcp_servers: Sequence[McpServerConfig],
    slack_connection: slack.Connection | None,
) -> AgentSessionFlow:
    """Builds the agent; ``models`` are the allowed model names, and the first is the default.

    ``slack_connection`` None leaves out the Slack connector tools.
    """
    clients = {model: _model_client(model) for model in models}
    mcp_tool_sources = (
        [McpToolSource(mcp_servers, resource_and_prompt_tools_enabled=True)] if mcp_servers else []
    )
    definition = AgentDefinition(
        name="ai-agent",
        instructions=system_prompt,
        model=clients[models[0]],
        # Settings hold only a model name; an unknown one gets the default, never a new client.
        resolve_model=lambda settings: clients.get(str(settings.get("model")), clients[models[0]]),
        tool_sources=[*mcp_tool_sources, *build_slack_tool_sources(slack_connection)],
        built_in_tools=frozenset(BuiltInTool),
        plan_mode_enabled=True,
        compaction=CompactionPolicy(context_window_tokens=context_window_tokens),
    )
    return stepwise_agent_session_flow(definition, flow_type=AI_AGENT_SESSION_FLOW_TYPE)

例子: examples/python/dex_examples/products/ai-agent/ai_agent_session.py

允许的 model、system prompt、context window 与 MCP server 都是 Worker 配置，由所有 session 共享。每个 session 在启动时从允许的 model 中选择一个。它的 session settings 只保存 model 名称，由 Worker 把名称映射到 model client，因此 credential 和 server command 永远不会来自 browser 或 Flow input。修改后的 prompt 或 model client 会在每个 session 的下一次模型调用时生效，包括正在等待的 turn，因为 Dex 不会重放 Flow。

Worker​

Worker 把 session Flow 与应用的其他 Flow 一起注册。Agent session 的 Step 是 async 的，因此 Python Registry 允许 async handler。

app.pyPython
        self.ai_agent_mcp_servers = load_mcp_servers(config.agent_mcp_config)
        self.ai_agent = build_ai_agent_session_flow(
            config.agent_models,
            config.agent_system_prompt,
            config.agent_context_tokens,
            self.ai_agent_mcp_servers,
            ai_agent_slack_connection,
        )
        # ...
        self.registry = Registry(tuple(flows), allow_async_handlers=True)

例子: examples/python/dex_examples/app.py

Durable context​

Session 会把每条 user message、模型输出 item 与 tool result 分别保存为一条 message。最新的 message 是 AgentMessages AttributeMap instance，较早的 message 会移入不可变的 AgentMessageChunks instance。模型输出 item 会按 provider 返回的原样保存，包括加密的 reasoning item 与 thinking signature，并原样重放。

当模型输入接近配置的 context window 时，CompactContext Step 会在下一次模型调用前摘要较早的 message。之后模型收到 summary 与最近的 message。Compaction 不会把 tool call 与它的 result 分开。保留策略之后可以删除较早的已摘要 message，因此长 session 的 state 保持有界。触发阈值、保留上限以及如何保留完整对话，参见 Compaction 与存储上限。

排队消息与 Steer​

Turn 运行期间发送的 message 会在 durable QueuedUserMessages queue 中等待。Message 处于 pending 时，UI 会显示 Steer 与 Delete 控件。

Steer 会在 turn 的下一个 safe Step boundary 应用 message。它不会停止已经开始的模型调用或 tool call。尚未开始的 tool call，以及仍在等待的 approval、question 与 durable wait，会记录结构化的 cancellation result。随后 turn 结束，新的 turn 以所有 steered message 开始。Stop 调用 cancel_turn，以同样方式结束 turn，但不会开始新的 turn。

提问与 durable wait​

内置 ask_user tool 会向用户提问，并在 AwaitQuestionAnswer 中等待。UI 会用选择按钮或文本框显示问题，答案会在同一个 turn 中成为该 tool 的 result。

内置 durable_wait tool 会在 AwaitTimer 中等待一个 Timer。等待期间不占用模型调用或 Worker，并且可以跨 Worker 与 engine 重启。Steer 或 Stop 会中断等待，queued message 不会。

Durable plan​

为一条 message 开启 Plan mode，即可让 Agent 制定计划。Planning turn 只能调用内置 write_todos tool，它会替换 AgentPlan Attribute 并增加 revision。UI 会展示 draft，Execute plan 会用审阅过的 revision 调用 execute_plan。过期的 revision 会被拒绝。如果 plan 在仍有未完成 task 时停止，它会保持 active，并可以继续执行。Plan 处于 active 时，其他 turn 可以调用所有 tool；只有 draft plan 会把 turn 限制为只能调用 write_todos。

MCP tool 与审批​

Worker 从本地配置加载受信任的 stdio 或 Streamable HTTP MCP server。Agent 把 Dex Agents 的 McpToolSource 作为 tool source，因此模型能看到每个 server 的 tool。Worker 会拒绝它不认识的配置 key，包括这个 example 早期版本中按 tool 设置的配置。每次 MCP 调用都在自己的 tool Step 中运行。无法列出 tool 的 server 会在这次模型调用中被略过，并在下一次调用时重试，因此所有 server 都不可用时，Agent 仍会继续回答。

Tool 的 effect 决定它如何运行。当 server 的 annotation 受信任时，read-only tool 会立即运行，并在崩溃后重试。其他 tool 会在 AwaitToolApproval 中等待用户批准，然后最多运行一次。被崩溃打断的写操作会记录为结果未知，并且不会重复执行。每次 MCP 调用都带有 idempotency key，server 可以据此丢弃重复请求。

开启 resource_and_prompt_tools_enabled 后，tool source 还会提供三个 read tool，用于列出并读取 server 的 resource 与 prompt。

实时 event​

文本、reasoning summary、tool activity、approval 与 plan 更新会出现在 session 的 AgentEvents Stream 上。Python example 使用库提供的 browser stream helper，把它以 Server-Sent Events 的形式提供给浏览器。TypeScript example 把它作为供 useChat 使用的 AI SDK UI message stream：一个 route 发送用户消息或 tool 审批，并把这一轮的内容流式写入浏览器中的 chat。

http_routes.pyPython
    @blueprint.get("/events")
    async def stream_agent_events() -> Response:
        session_id = authorize_session(required_query("session_id"))
        try:
            await sessions().get_agent_snapshot(session_id)
        except FlowNotActiveOrNotFoundError as error:
            raise NotFound("the AI Agent session is missing or no longer retained") from error
        resume_token = request.headers.get("Last-Event-ID") or request.args.get("resume_token", "")
        try:
            frames = await stream_agent_event_frames(
                app_state.client, app_state.ai_agent, session_id, resume_token=resume_token
            )
        except InvalidAgentEventResumeTokenError as error:
            raise BadRequest("the resume token is not valid; reload the snapshot") from error
        response = await make_response(frames, dict(AGENT_EVENT_STREAM_HEADERS))
        response.timeout = None
        return response

例子: examples/python/dex_examples/products/ai-agent/http_routes.py

页面先加载 session snapshot，以及打开 Stream 所需的 resume token。重新连接的浏览器会从它看到的最后一个 event 继续。Stream delivery 是 best effort，因此页面会在每个 turn 结束后以及每八秒重新加载 snapshot。Conversation 始终来自 snapshot，其中包含最新的 200 条 message。Load older messages 会通过 list_agent_messages 向前翻页。

Session 访问​

Session ID 不是秘密：它会出现在页面 URL 中。因此，每个指定 session 的 route 在调用 Dex 之前，都会先检查该 session 的 token。POST /sessions 会检查所请求的 model 是否在 Worker 允许的 model 之中，选择新的 session ID，并返回它的 token。Token 是用 DEX_AGENT_SESSION_KEY 对 session ID 计算的 HMAC-SHA256，因此 example 不存储任何 token，使用同一个 key 的每个进程都接受它。如果请求缺少 X-Agent-Session-Token header，或 header 不匹配，route 返回 403。

http_routes.pyPython
    def authorize_session(session_id: str) -> str:
        # Every route that names a session checks its token before any Dex call.
        expected = sign_session_id(app_state.config.agent_session_key, session_id)
        if not hmac.compare_digest(request.headers.get(SESSION_TOKEN_HEADER, ""), expected):
            raise Forbidden("this browser has no token for the AI Agent session")
        return session_id

    @blueprint.post("/sessions")
    async def start_session() -> Response:
        body = await request.get_json(silent=True)
        if not isinstance(body, dict):
            raise BadRequest("the request body must be a JSON object")
        model = body.get("model", app_state.config.agent_models[0])
        if model not in app_state.config.agent_models:
            raise BadRequest("model must be one of the models in /settings")
        # The server picks the session ID, so a caller cannot claim an existing session.
        session_id = f"ai-agent-{uuid.uuid4()}"
        # Settings carry only the model name; the Worker resolves it, and keys stay on the Worker.
        await sessions().start_session(session_id, session_settings={"model": model})
        session_token = sign_session_id(app_state.config.agent_session_key, session_id)
        return jsonify(session_id=session_id, session_token=session_token)

例子: examples/python/dex_examples/products/ai-agent/http_routes.py

两个页面都把每个 token 按 session ID 保存在 local storage 中，页面 URL 中只放 session ID，因此复制的 URL 不会授予访问权限。两个页面都在每个请求中发送这个 header。Python 页面用 fetch 读取 event stream，而不是 EventSource，因为 EventSource 无法发送 header。TypeScript 页面把这个 header 交给 useChat transport，用于它的请求和重新连接。AI Agent 的 route 不发送 CORS header，因此其他 origin 的页面无法调用它们。

如果没有设置 DEX_AGENT_SESSION_KEY，进程会在启动时选择一个随机 key，重启之后，之前启动的 session 就无法再访问。真实的应用会用自己的登录替换这个 token，并检查用户是否可以使用该 session。

在本地运行​

在第一个终端中，用 dexcli dev 启动 Dex。它会一直在前台运行。在第二个终端中启动 example。两个 example 都在 127.0.0.1:8080/products/ai-agent/ 提供 AI Agent 页面，因此一次只运行一个。

cd examples/python
uv sync --locked
cd ai-agent
npm ci
npm run build
cd ..
export DEX_AGENT_MCP_CONFIG="$PWD/ai-agent/mcp-servers.local.yaml"
export DEX_AGENT_SESSION_KEY="$(openssl rand -hex 32)"
uv run --frozen python main.py

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

Python 版的 Flow 与 HTTP route 位于 examples/python/dex_examples/products/ai-agent。它的 React UI、MCP 配置示例与运行说明位于 examples/python/ai-agent。TypeScript 版的 Flow、route 与 React UI 位于 examples/typescript/src/products/ai-agent。

第一页会显示 Worker 的 model、prompt、context window 与 MCP tool。启动表单从 Worker 允许的 model 中为 session 选择 model。

默认的 demo model 不需要 credential。它支持 /ask、/choose、/wait 与 /tool 等斜杠命令，因此无需 key 就能体验所有功能。在 TypeScript 中，demo model 是 mock/dex 与 mock/dex-mini，/model 会说出 session 的模型。

要使用真实模型，设置下面的变量并重启 Worker：

变量含义
DEX_AGENT_MODEL默认模型。Python 中是 LiteLLM model 名称。TypeScript 中是 AI SDK model，例如 openai/gpt-5-mini。
DEX_AGENT_MODELSSession 还可以选择的其他模型，用逗号分隔。
DEX_AGENT_SYSTEM_PROMPTSystem prompt。
DEX_AGENT_CONTEXT_TOKENSCompaction 所依据的 context window。
DEX_AGENT_MCP_CONFIGMCP server 配置文件。
DEX_AGENT_SESSION_KEY签发 session token 的 key。

把每个 provider key 放在 Worker 的环境中，Python 可以放在 examples/.env。Worker 从那里读取它，它永远不会进入 Flow state。

Example 页面有自己的审批按钮。如果想在 Dex Web 中审批，请按 在 Dex Web 中审批 操作，Flow type 用 AIAgentSession。

在 Dex Web 中审批​

Dex Web 需要 session graph 才能显示 Agent session tab 及其控件。为 Flow type 把 graph 渲染一次到一个目录，再用 --flow-rendering-dir 启动 dexcli dev。如果 dexcli dev 已经在没有这个参数的情况下运行，先停止它。Dex Web 每次加载页面都会重新读取这个目录，所以重新渲染的 graph 不需要重启就能显示：

mkdir -p definitions
dex-agents-session-graph --flow-type AIAgentSession --permission ai-agent.operate \
  --out definitions/ai-agent.json
dexcli dev --flow-rendering-dir ./definitions

Python 与 TypeScript package 都会安装 dex-agents-session-graph。在 examples/python 中，用 uv run --frozen 运行它。在 examples/typescript 中，用 npx 运行它。

在 Dex Web 中打开这个 session 的 run，选择 Agent session tab。把 Working as 设为渲染时指定的 permission，例如 ai-agent.operate。然后在 tool call 卡片上选择 Approve 或 Reject。生产环境的做法参见 在 Dex Web 中显示 session。

连接 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/dex_examples/products/ai-agent/slack_connector_tools.py

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

main.pyPython
async def main() -> None:
    load_dotenv(Path(__file__).resolve().parents[1] / ".env")
    config = ExamplesConfig.from_env()
    slack_setup = await open_slack_project_connection()
    try:
        app_state = ExampleApp(config, slack_setup.connection if slack_setup else None)
        # ...
    finally:
        if slack_setup is not None:
            await slack_setup.aclose()

例子: examples/python/main.py

Agent 把 Slack tool source 和 MCP tool source 放在一起：

ai_agent_session.pyPython
    mcp_tool_sources = (
        [McpToolSource(mcp_servers, resource_and_prompt_tools_enabled=True)] if mcp_servers else []
    )
    definition = AgentDefinition(
        # ...
        tool_sources=[*mcp_tool_sources, *build_slack_tool_sources(slack_connection)],
        # ...
    )

例子: examples/python/dex_examples/products/ai-agent/ai_agent_session.py

调用如何运行​

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

配置 connection​

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

试用时，先设置 DEX_CONNECTOR_CONFIG_FILE 再启动 example。Demo 模型可以直接调用这个 tool：

/tool slack_post_thread_reply {"channelId":"C0123456789","threadTimestamp":"1710000000.000100","text":"Hello from Dex"}

这次发帖会先等待 Approve，然后才运行。

限制​

  • 每次模型调用和每次 tool call 都会增加一次 Step 往返。
  • AgentEvents 是 best effort。对话内容请读取 session snapshot。
  • 失败的 RunModel 会重复它的模型调用。
  • Server 的限制参见 Limits for long sessions。Single-attempt tool 的例外参见 最多一次有一个例外。