---
title: "把 Connector 用作 tool"
description: "Agent 可以把官方 Connector 的 operation 当作 tool 调用。每次调用都在自己的 tool Step 中运行，使用 Connector 的 credential 和幂等键。"
canonical_url: "https://docs.superdurable.io/zh-Hans/agent-frameworks/connectors-as-tools/"
source: "docs/i18n/zh-Hans/docusaurus-plugin-content-docs/current/agent-frameworks/connectors-as-tools.mdx"
locale: "zh-Hans"
framework: "claude-agent-sdk"
language: "python"
---

> 供 AI Agent 使用的全部页面索引：https://docs.superdurable.io/zh-Hans/llms.txt

# 把 Connector 用作 tool

本文件为 Claude Agent SDK 版本。其他框架：[Google ADK](https://docs.superdurable.io/zh-Hans/agent-frameworks/connectors-as-tools.google-adk.md) · [LangChain and LangGraph](https://docs.superdurable.io/zh-Hans/agent-frameworks/connectors-as-tools.langchain-langgraph.md) · [Mastra](https://docs.superdurable.io/zh-Hans/agent-frameworks/connectors-as-tools.mastra.md) · [OpenAI Agents SDK](https://docs.superdurable.io/zh-Hans/agent-frameworks/connectors-as-tools.openai-agents-sdk.md) · [Pydantic AI](https://docs.superdurable.io/zh-Hans/agent-frameworks/connectors-as-tools.pydantic-ai.md) · [Vercel AI SDK](https://docs.superdurable.io/zh-Hans/agent-frameworks/connectors-as-tools.vercel-ai-sdk.md) · [Built-in agent loop](https://docs.superdurable.io/zh-Hans/agent-frameworks/connectors-as-tools.built-in-agent-loop.md)

本页代码为 Python。其他语言：[TypeScript](https://docs.superdurable.io/zh-Hans/agent-frameworks/connectors-as-tools.typescript.md)

Agent 可以把官方 [Connector](https://docs.superdurable.io/zh-Hans/concepts/connector.md) 的 operation 当作 tool 调用。每次调用都在自己的 tool Step 中运行，使用 Connector 的 credential 和幂等键。

<a id="connectors-as-tools"></a>

## 工作方式

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

“连接 Connector”一节展示了每个框架的接法；[内置 Agent Loop](https://docs.superdurable.io/zh-Hans/agent-frameworks/built-in-agent-loop.md#connect-connectors) 展示了不用框架的接法。[Connector tools reference](https://docs.superdurable.io/zh-Hans/reference/agents/connector-tools.md)（英文）介绍 package、tool 名称、结果大小和 connection。

<a id="connector-tools-offered"></a>

### 提供哪些 operation

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

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

<a id="connector-tools-effects"></a>

### Effect class

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

| Operation                                          | Effect class           | 运行于                           |
| -------------------------------------------------- | ---------------------- | ----------------------------- |
| Query                                              | **read**               | `CallTool`，会重试                |
| Manifest 标记为可安全重试（**agent.retry: safe**）的 Mutation | `idempotent_write`     | `CallTool`，会重试                |
| 其他 Mutation                                        | `non_idempotent_write` | `CallSingleAttemptTool`，只运行一次 |

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

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

<a id="connector-tools-approval"></a>

### 审批

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

<a id="connector-tools-outcomes"></a>

### 结果

Operation 以它的某个分支结束，由分支决定 Tool Call 的记录：**succeeded**、**failed**，或者在写操作可能已经生效时记录为 `outcome_unknown`。[Outcomes](https://docs.superdurable.io/zh-Hans/reference/agents/connector-tools.md#outcomes)（英文）列出了每个分支及其记录。

`outcome_unknown` 的核对方式与其他 tool 相同；参见 [核对 outcome_unknown](https://docs.superdurable.io/zh-Hans/agent-frameworks/tool-calls.md#reconcile-outcome-unknown)。

<a id="connector-tools-write-order"></a>

### 写操作顺序

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

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

<a id="connector-tools-crashes"></a>

### 崩溃

Connector 写操作期间 Worker 崩溃，这个写操作最多只发送一次。可安全重试的写操作会用同一个幂等键重试，或者先在 provider 那里查找之前的写入；单次尝试写操作记录为 `outcome_unknown`。[Crashes](https://docs.superdurable.io/zh-Hans/reference/agents/connector-tools.md#crashes)（英文）逐一列出了每种情况。

<a id="connect-connectors"></a>

## 连接 Connector

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

<a id="slack-connector-tool"></a>

### 示例中的 Slack tool

配置好 Slack connection **support-slack** 后，每个示例 Agent 都会得到 `slack_post_thread_reply`，也就是 Slack operation `postThreadReply`。Worker 在本地从 `DEX_CONNECTOR_CONFIG_FILE` 打开这个 connection，部署后则从 **DEX_PROJECT\_\*** 设置中打开；[Connections](https://docs.superdurable.io/zh-Hans/reference/agents/connector-tools.md#connector-tools-connection)（英文）介绍这两种方式，[the Slack tool in the examples](https://docs.superdurable.io/zh-Hans/reference/agents/connector-tools.md#slack-example)（英文）介绍 Slack app。

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

<a id="offer-connector-tools"></a>

### 提供 tool

**Python**

安装 **dex-agents** 的 **connectors** extra，以及 Connector 包，例如 **dex-connector-slack**。安装命令见 [Connector tools](https://docs.superdurable.io/zh-Hans/reference/agents/connector-tools.md#install)（英文）。

Worker 从项目配置中打开 Slack connection **support-slack**。`connector_tool` 把 operation 变成 tool，由 `ConnectorToolSource` 持有。没有这个 connection 时，Agent 不会得到 Slack tool。

```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/agent-harnesses/agent_harness_examples/slack_connector_tools.py](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses/agent_harness_examples/slack_connector_tools.py)

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

**Python**

```python
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](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses/agent_harness_examples/__main__.py)

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

**Python**

```python
    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](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses/agent_harness_examples/claude_agent_sdk_order_support.py)

<a id="connector-tool-calls"></a>

### 调用如何运行

如 [Slack tool](https://docs.superdurable.io/zh-Hans/agent-frameworks/connectors-as-tools.md#slack-connector-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` 这次尝试失败。

<a id="configure-connector-connection"></a>

### 配置 connection

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

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

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

Worker 从哪里找到 connection **support-slack**、Slack app 需要什么，见 [Connections](https://docs.superdurable.io/zh-Hans/reference/agents/connector-tools.md#connector-tools-connection)（英文）。

<a id="next"></a>

## 下一步

- [Connector tools reference](https://docs.superdurable.io/zh-Hans/reference/agents/connector-tools.md)（英文）