跳到主要内容
Super Durable
菜单

把 Connector 用作 tool

查看 Markdown

Agent 可以把官方 Connector 的 operation 当作 tool 调用。每次调用都在自己的 tool Step 中运行，使用 Connector 的 credential 和幂等键。

工作方式

ConnectorToolSource 提供你指定的 operation，Agent 把它作为 tool source 接收（Python 中是 tool_sources，TypeScript 中是 toolSources）。每次调用都在自己的 tool Step 中运行一个 operation，因此 Call ID、幂等键和 credential 的行为与在 Connector Step 中相同。每次尝试的 Call ID 都保持不变。

“连接 Connector”一节展示了每个框架的接法；内置 Agent Loop 展示了不用框架的接法。Connector tools reference（英文）介绍 package、tool 名称、结果大小和 connection。

提供哪些 operation

应用逐个指定 Agent 可以使用的 operation，而且 Connector 的 manifest dex-connector.json 必须把它标记为对 Agent 开放。定义 Agent 时，Dex 会拒绝其他任何 operation。

Dex 先按 operation 的 input schema 检查模型给出的参数。字段错误或缺失时，调用在 operation 运行之前就会失败，并返回一条模型可以据此修正的消息。

Effect class

Manifest 决定一次调用能否再次运行，从而决定由哪个 Step 运行它：

OperationEffect class运行于
QueryreadCallTool，会重试
Manifest 标记为可安全重试（agent.retry: safe）的 Mutationidempotent_writeCallTool，会重试
其他 Mutationnon_idempotent_writeCallSingleAttemptTool，只运行一次

可安全重试的意思是：用同一个 Call ID 重复执行，不会产生第二次效果。尝试次数用完的调用，Query 记录为 failed，Mutation 记录为 outcome_unknown。

Session 对 Connector tool 设置的 tool policy 只改变它的审批方式。Effect class 始终来自 manifest。

审批

默认情况下，manifest 建议审批（agent.approval: recommended）的 Connector tool 会等待审批，例如退款；其他 Mutation 直接运行。应用可以为每个 tool 覆盖这个设置，或者传入一个 predicate，根据参数逐次决定，例如超过某个金额的退款。Predicate 抛出异常时，调用会请求审批。

结果

Operation 以它的某个分支结束，由分支决定 Tool Call 的记录：succeeded、failed，或者在写操作可能已经生效时记录为 outcome_unknown。Outcomes（英文）列出了每个分支及其记录。

outcome_unknown 的核对方式与其他 tool 相同；参见 核对 outcome_unknown。

写操作顺序

同一个模型响应中的 Connector 写操作按模型给出的顺序逐个运行。读操作和其他 tool 的调用并行运行。需要审批的写操作要等轮到它时才请求审批。

当一个写操作失败、以 outcome_unknown 结束或被拒绝时，同一响应中后面的 Connector 写操作都会被跳过。每个被跳过的写操作记录为 failed 并设置 skipped，错误信息以“the tool did not run, so it is safe to call again”开头，Dex Web 把它显示为已跳过。

崩溃

Connector 写操作期间 Worker 崩溃，这个写操作最多只发送一次。可安全重试的写操作会用同一个幂等键重试，或者先在 provider 那里查找之前的写入；单次尝试写操作记录为 outcome_unknown。Crashes（英文）逐一列出了每种情况。

连接 Connector

这个示例加入了 Slack tool slack_post_thread_reply，用来在 Slack thread 中回复。

示例中的 Slack tool

配置好 Slack connection support-slack 后，每个示例 Agent 都会得到 slack_post_thread_reply，也就是 Slack operation postThreadReply。Worker 在本地从 DEX_CONNECTOR_CONFIG_FILE 打开这个 connection，部署后则从 DEX_PROJECT_* 设置中打开；Connections（英文）介绍这两种方式，the Slack tool in the examples（英文）介绍 Slack app。

postThreadReply 是单次尝试 Mutation，所以每次发帖只在 CallSingleAttemptTool 中运行一次，Worker 崩溃也不会让它发送两次。示例还设置了 requires_approval=True（TypeScript 中是 requiresApproval: true），所以每次发帖都会在 AwaitToolApproval 中等待人审批。

提供 tool

安装 dex-agents 的 connectors extra，以及 Connector 包，例如 dex-connector-slack。安装命令见 Connector tools（英文）。

Worker 从项目配置中打开 Slack connection support-slack。connector_tool 把 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/agent-harnesses/agent_harness_examples/slack_connector_tools.py

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

__main__.pyPython
async def main() -> None:
    config = AgentHarnessExamplesConfig.from_env()
    slack_setup = await open_slack_project_connection()
    try:
        app_state = AgentHarnessExamplesApp(config, slack_setup.connection if slack_setup else None)
        # ...
    finally:
        if slack_setup is not None:
            await slack_setup.aclose()

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

Agent 把这个 tool source 和它自己的 tool 一起传入：

claude_agent_sdk_order_support.pyPython
    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),
        # ...
    )

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

调用如何运行

如 Slack tool 所述，每次调用 slack_post_thread_reply 都要先经人批准，然后在自己的 CallSingleAttemptTool Step 中只运行一次。各框架这样把调用交给这个 Step：

在 Claude Agent SDK 中，Connector Tool Call 一律不在 CLI 内部运行，读操作也一样。每次调用都会结束当前 segment，并在自己的 tool Step 中运行，因为 Connector 需要这个 Step 来提供 Call ID。之后恢复运行的 CLI 会拿到记录下来的结果。CLI 看到的 tool 名称是 mcp__tools__slack_post_thread_reply；tool policy、审批和记录使用 slack_post_thread_reply。

同一个模型响应中的调用，每次只有一个离开 turn，到它的 tool Step 中运行。适配器会在 turn 内拒绝该响应中后面的写操作，因此它不会运行，也不留下记录；模型读取前一个写操作的结果后，再决定是否重新调用后面的写操作。不要把 Connector tool 放进 allowed_tools（allowedTools）：它们的调用必须离开 CLI，所以这样的条目会让 RunModel 这次尝试失败。

配置 connection

试用时，先设置 DEX_CONNECTOR_CONFIG_FILE 再启动示例。然后让 Agent 在某个 Slack thread 中回复，并给出 channel ID 和 thread timestamp：

Reply in Slack thread C0123456789 1710000000.000100: Order 1002 ships on Thursday.

这次发帖和 cancel_order 一样，会等待审批。

Worker 从哪里找到 connection support-slack、Slack app 需要什么，见 Connections（英文）。

下一步