--- title: "内置 Agent Loop" description: "内置 agent loop 不需要任何 agent framework 就能运行 Agent。会话由 Dex 管理，每次模型输入都从 durable state 构建。因此 Agent 不依赖 LLM provider 的 conversation API。" canonical_url: "https://docs.superdurable.io/zh-Hans/agent-frameworks/built-in-agent-loop/" source: "docs/i18n/zh-Hans/docusaurus-plugin-content-docs/current/agent-frameworks/built-in-agent-loop.mdx" locale: "zh-Hans" --- # 内置 Agent Loop 内置 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](https://docs.superdurable.io/zh-Hans/agent-frameworks.md#choose-a-harness)。 ## Session 如何运行 每次模型调用都是一个 **RunModel** Step。每次 tool call 都是独立的 Step，因此崩溃最多只会重复当时正在运行的那一次调用。 Step、Attribute、Channel、RPC 与 event 都与其他 Dex Agents session 相同： ```text 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](https://github.com/superdurable/dex/tree/main/agents/python/core) 列出了每个 Step 及其重试与 failure route。 失败的 **RunModel** 会重试。第三次尝试失败后，这个 turn 失败，session 等待下一条消息。Read 或 idempotent write 在 **CallTool** 中重试。其他写操作在 **CallSingleAttemptTool** 中只运行一次；如果 policy 要求审批，会在审批之后运行。 ## 安装 在 TypeScript 中，Dex SDK 是 peer package，因此也要一起安装。 **Python** ```bash pip install "dex-agents[litellm,mcp]" ``` 例子: [agents/python/core/README.md](https://github.com/superdurable/dex/tree/main/agents/python/core/README.md) **TypeScript** ```bash npm i @superdurable/dex-agents @superdurable/dex ai @modelcontextprotocol/sdk ``` 例子: [agents/typescript/core/README.md](https://github.com/superdurable/dex/tree/main/agents/typescript/core/README.md) **litellm** extra 和 **ai** package 提供 model client。**mcp** extra 和 **@modelcontextprotocol/sdk** 提供 MCP tool。不用的可以不装。 ## Example [AI Agent](https://docs.superdurable.io/zh-Hans/use-cases/ai-agent.md) 是一个长期运行的通用 Agent，有 Python 与 TypeScript 两个版本。它与用户聊天并调用受信任的 MCP tool。它在修改外部系统前等待审批、向用户提问、通过 durable Timer 暂停，并在执行前先制定计划供用户审阅。 ### Session Flow Application 用 Dex 的概念声明 Agent，并构建它的 Flow。Flow type 是 **AIAgentSession**。一次 Flow run 就是一个 session，Flow ID 就是 session ID。 **Python** ```python 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](https://github.com/superdurable/dex/tree/main/examples/python/dex_examples/products/ai-agent/ai_agent_session.py) **TypeScript** ```typescript export function createAiAgentFlow(config: AiAgentConfig, slack: Connection | undefined): AiAgentFlow { const clients = new Map(config.models.map((model) => [model, createModelClient(model)])); const defaultClient = clients.get(config.models[0]!)!; return createStepwiseAgentSessionFlow( { name: "ai-agent", instructions: config.systemPrompt, model: defaultClient, // Settings hold only a model name; an unknown one gets the default, never a new client. resolveModel: (settings) => clients.get(String(settings.model)) ?? defaultClient, // Each MCP tool's annotations pick its effect, and so whether it waits for approval. toolSources: [ new McpToolSource(config.mcpServers, { resourceAndPromptToolsEnabled: true }), ...createSlackConnectorToolSources(slack), ], builtInTools: ["ask_user", "durable_wait", "write_todos"], planModeEnabled: true, compaction: { contextWindowTokens: config.contextWindowTokens }, }, "AIAgentSession", ); } ``` 例子: [examples/typescript/src/products/ai-agent/ai-agent-flow.ts](https://github.com/superdurable/dex/tree/main/examples/typescript/src/products/ai-agent/ai-agent-flow.ts) 允许的 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。 **Python** ```python 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](https://github.com/superdurable/dex/tree/main/examples/python/dex_examples/app.py) **TypeScript** ```typescript // One Slack connection for the agent examples; without it they offer no Slack tools. const slack = await openSlackConnectionFromProject(); const aiAgentConfig = loadAiAgentConfig(); const aiAgentFlow = createAiAgentFlow(aiAgentConfig, slack); ``` 例子: [examples/typescript/src/registry.ts](https://github.com/superdurable/dex/tree/main/examples/typescript/src/registry.ts) ## 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 与存储上限](https://docs.superdurable.io/zh-Hans/reference/agents/session-limits.md#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](https://docs.superdurable.io/zh-Hans/concepts/primitives/timer.md)。等待期间不占用模型调用或 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](https://docs.superdurable.io/zh-Hans/concepts/primitives/stream.md) 上。Python example 使用库提供的 browser stream helper，把它以 Server-Sent Events 的形式提供给浏览器。TypeScript example 把它作为供 **useChat** 使用的 AI SDK UI message stream：一个 route 发送用户消息或 tool 审批，并把这一轮的内容流式写入浏览器中的 chat。 **Python** ```python @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](https://github.com/superdurable/dex/tree/main/examples/python/dex_examples/products/ai-agent/http_routes.py) **TypeScript** ```typescript router.post("/api/chat", async (request, response) => { const body = requireBody(request); const flowId = authorizeSession(request, requireString(body, "id")); const lastMessage = Array.isArray(body.messages) ? (body.messages.at(-1) as JsonBody | undefined) : undefined; const signal = abortSignalOf(response); if (lastMessage?.role === "assistant") { const approvals = readAgentToolApprovalResponses(lastMessage); await sendFetchResponse(response, resolveAgentToolApprovalsWithUIMessageStream({ ...session, flowId, approvals, signal })); return; } const parts = (Array.isArray(lastMessage?.parts) ? lastMessage.parts : []) as JsonBody[]; const text = parts.map((part) => (part.type === "text" ? String(part.text) : "")).join(""); await sendFetchResponse(response, sendAgentUserMessageWithUIMessageStream({ ...session, flowId, text, signal })); }); ``` 例子: [examples/typescript/src/products/ai-agent/controller.ts](https://github.com/superdurable/dex/tree/main/examples/typescript/src/products/ai-agent/controller.ts) 页面先加载 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。 **Python** ```python 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](https://github.com/superdurable/dex/tree/main/examples/python/dex_examples/products/ai-agent/http_routes.py) **TypeScript** ```typescript const issueSessionToken = (sessionId: string) => createHmac("sha256", sessionKey).update(sessionId).digest("hex"); const authorizeSession = (request: Request, sessionId: string): string => { const expected = Buffer.from(issueSessionToken(sessionId)); const presented = Buffer.from(request.header(sessionTokenHeaderName) ?? ""); if (presented.length !== expected.length || !timingSafeEqual(presented, expected)) { throw new HttpError(403, "this browser has no token for the AI Agent session"); } return sessionId; }; const sessionIdOf = (request: Request) => authorizeSession(request, String(request.params.id)); router.use("/", express.static(fileURLToPath(new URL("./ui/", import.meta.url)))); router.get("/settings", (_request, response) => { response.json({ model: models[0], models }); }); router.post("/sessions", async (request, response) => { const model = requireBody(request).model ?? models[0]; if (typeof model !== "string" || !models.includes(model)) { throw new HttpError(400, "model must be one of the models in /settings"); } // The server picks the session ID, and settings carry only the model name the Worker resolves. const sessionId = `ai-agent-${randomUUID()}`; const start = { initialMessage: null, sessionSettings: { model } }; await client.startFlow(flow, sessionId, start, { timeoutMs: 30 * DAY_MS }); response.json({ session_id: sessionId, session_token: issueSessionToken(sessionId) }); }); ``` 例子: [examples/typescript/src/products/ai-agent/controller.ts](https://github.com/superdurable/dex/tree/main/examples/typescript/src/products/ai-agent/controller.ts) 两个页面都把每个 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 页面，因此一次只运行一个。 **Python** ```bash 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](https://github.com/superdurable/dex/tree/main/examples/python/ai-agent/README.md) **TypeScript** ```bash cd examples/typescript npm install npm start ``` 例子: [examples/typescript/src/products/ai-agent/README.md](https://github.com/superdurable/dex/tree/main/examples/typescript/src/products/ai-agent/README.md) Python 版的 Flow 与 HTTP route 位于 [examples/python/dex_examples/products/ai-agent](https://github.com/superdurable/dex/tree/main/examples/python/dex_examples/products/ai-agent)。它的 React UI、MCP 配置示例与运行说明位于 [examples/python/ai-agent](https://github.com/superdurable/dex/tree/main/examples/python/ai-agent)。TypeScript 版的 Flow、route 与 React UI 位于 [examples/typescript/src/products/ai-agent](https://github.com/superdurable/dex/tree/main/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_MODELS** | Session 还可以选择的其他模型，用逗号分隔。 | | **DEX_AGENT_SYSTEM_PROMPT** | System prompt。 | | **DEX_AGENT_CONTEXT_TOKENS** | Compaction 所依据的 context window。 | | **DEX_AGENT_MCP_CONFIG** | MCP server 配置文件。 | | **DEX_AGENT_SESSION_KEY** | 签发 session token 的 key。 | 把每个 provider key 放在 Worker 的环境中，Python 可以放在 **examples/.env**。Worker 从那里读取它，它永远不会进入 Flow state。 Example 页面有自己的审批按钮。如果想在 Dex Web 中审批，请按 [在 Dex Web 中审批](https://docs.superdurable.io/zh-Hans/agent-frameworks/built-in-agent-loop.md#approve-in-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 不需要重启就能显示： ```bash 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](https://docs.superdurable.io/zh-Hans/reference/agents/dex-web.md)。 ## 连接 Connector Agent 可以把 [Connector](https://docs.superdurable.io/zh-Hans/concepts/connector.md) 的 operation 当作 tool 调用，见 [把 Connector 用作 tool](https://docs.superdurable.io/zh-Hans/agent-frameworks.md#connectors-as-tools)。这个 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](https://docs.superdurable.io/zh-Hans/reference/agents/connector-tools.md#install)。 Worker 从项目配置中打开 Slack connection **support-slack**。Python 的 **connector_tool** 或 TypeScript 的 **connectorTool** 把 operation 变成 tool，由 **ConnectorToolSource** 持有。没有这个 connection 时，agent 不会得到 Slack tool。 **Python** ```python 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](https://github.com/superdurable/dex/tree/main/examples/python/dex_examples/products/ai-agent/slack_connector_tools.py) **TypeScript** ```typescript export async function openSlackConnectionFromProject( environment: NodeJS.ProcessEnv = process.env, ): Promise { // ... const project = await loadFromEnvironment(environment); // ... return newProjectConnection(project, slackConnectionName); } // ... export function createSlackConnectorToolSources(slack: Connection | undefined): ToolSource[] { if (slack === undefined) { return []; } return [ new ConnectorToolSource([ connectorTool(slack.operations.postThreadReply(), slack.reference, { manifest, inputValidator: PostThreadReplyInputSchema, // The manifest does not ask for approval; this agent asks before every post. requiresApproval: true, }), ]), ]; } ``` 例子: [examples/typescript/src/shared/slack-connector-tools.ts](https://github.com/superdurable/dex/tree/main/examples/typescript/src/shared/slack-connector-tools.ts) Worker 在启动时打开一次 connection，每个 agent 都使用它： **Python** ```python 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](https://github.com/superdurable/dex/tree/main/examples/python/main.py) **TypeScript** ```typescript // One Slack connection for the agent examples; without it they offer no Slack tools. const slack = await openSlackConnectionFromProject(); ``` 例子: [examples/typescript/src/registry.ts](https://github.com/superdurable/dex/tree/main/examples/typescript/src/registry.ts) Agent 把 Slack tool source 和 MCP tool source 放在一起： **Python** ```python 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](https://github.com/superdurable/dex/tree/main/examples/python/dex_examples/products/ai-agent/ai_agent_session.py) **TypeScript** ```typescript // Each MCP tool's annotations pick its effect, and so whether it waits for approval. toolSources: [ new McpToolSource(config.mcpServers, { resourceAndPromptToolsEnabled: true }), ...createSlackConnectorToolSources(slack), ], ``` 例子: [examples/typescript/src/products/ai-agent/ai-agent-flow.ts](https://github.com/superdurable/dex/tree/main/examples/typescript/src/products/ai-agent/ai-agent-flow.ts) ### 调用如何运行 每次调用 **slack_post_thread_reply** 都要先经人批准，然后在自己的 **CallSingleAttemptTool** Step 中只运行一次。原因见 [Slack Connector tool](https://docs.superdurable.io/zh-Hans/agent-frameworks.md#slack-connector-tool)。 ### 配置 connection Worker 从哪里找到 connection **support-slack**、Slack app 需要什么，见 [Connection](https://docs.superdurable.io/zh-Hans/reference/agents/connector-tools.md#connector-tools-connection)。 试用时，先设置 **DEX_CONNECTOR_CONFIG_FILE** 再启动 example。Demo 模型可以直接调用这个 tool： ```text /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](https://docs.superdurable.io/zh-Hans/reference/agents/session-limits.md)。Single-attempt tool 的例外参见 [最多一次有一个例外](https://docs.superdurable.io/zh-Hans/agent-frameworks.md#tool-runs-twice)。 ## 相关内容 - [Agent Framework](https://docs.superdurable.io/zh-Hans/agent-frameworks.md) - [AI Agent](https://docs.superdurable.io/zh-Hans/use-cases/ai-agent.md) - [Agents reference](https://docs.superdurable.io/zh-Hans/reference/agents.md) - [Attribute](https://docs.superdurable.io/zh-Hans/concepts/primitives/attribute.md) - [Channel](https://docs.superdurable.io/zh-Hans/concepts/primitives/channel.md) - [Stream](https://docs.superdurable.io/zh-Hans/concepts/primitives/stream.md) - [Timer](https://docs.superdurable.io/zh-Hans/concepts/primitives/timer.md)