在 Pi Agent 中调用 OpenClaw：Telegram MCP 项目级接入实践

先看落地结果

#

文本复核的基础方案使用以下配置：

整体调用链如下：

flowchart LR
    subgraph Project[项目目录]
        P[Pi Agent]
        C[.mcp.json]
        A[pi-mcp-adapter]
        C --> A
        P --> A
    end

    subgraph Local[本机进程]
        M[Telegram MCP]
        E[.env 与 Session]
        A -->|"Streamable HTTP<br/>127.0.0.1:8765/mcp"| M
        E --> M
    end

    M -->|Telethon / MTProto| T[Telegram]
    T --> O[OpenClaw Bot]
    O --> T
    T --> M
    M --> A
    A --> P

这里没有做“记忆同步”。Pi 发出问题，OpenClaw 仍在自己的运行环境中读取已有记忆和上下文，再通过 Telegram 返回结果。双方的存储结构互不依赖。

0. 范围与版本

#

0.1 验证环境

#

本文在 2026 年 8 月 5 日验证了下面的组合：

公开分支包含 4 组改动：

0.2 本文覆盖什么

#

本文覆盖：

• 用 Telegram Bot 作为 OpenClaw 的现有通讯入口；
• 共享 Telegram MCP 服务的启动方式；
• Pi 项目级 MCP 配置；
• ask_bot 的事件监听、并发锁和超时语义；
• 文本回查、文件根目录和最小权限配置；
• 实际遇到的故障与验收方法。

本文不覆盖 OpenClaw 的部署、记忆目录格式和模型配置，也不把 Telegram 当作强一致消息队列。当前实现适合交互式复核，不适合无人值守的重要业务事务。

1. 为什么选择 Telegram MCP

#

最初要解决的问题很具体：我在 Pi 中分析代码或整理文档时，希望临时问 OpenClaw 一个问题，并把回复带回当前会话。

几种做法的差别如下：

Telegram MCP 对我来说是改动最少的一条路。OpenClaw 继续处理 Telegram 消息，Pi 只需要认识两个工具：发起一次 Bot 对话，以及回看最近消息。

这个选择也有代价。Telegram 是外部依赖，回复可能拆成多条消息，Bot 也可能主动推送内容。因此实现不能只做一次 send_message，还要定义等待、并发和恢复规则。

2. ask_bot 为什么采用事件监听

#

2.1 send_message 不够

#

普通 send_message 只确认消息已经发出：

await client.send_message(entity, message)
return "Message sent successfully."

它不知道哪一条入站消息属于本次请求。让 Pi 在发送后反复拉取聊天记录虽然能工作，但会增加 API 调用，也容易把旧回复当成新回复。

公开分支增加了 ask_bot。它在发送前注册临时的 Telethon NewMessage handler，然后等待目标 Bot 的下一条有效入站消息。

sequenceDiagram
    participant P as Pi Agent
    participant M as Telegram MCP
    participant T as Telethon
    participant B as OpenClaw Bot

    P->>M: ask_bot(message, timeout)
    M->>T: 注册临时 NewMessage handler
    M->>T: send_message
    T->>B: Telegram 消息
    B-->>T: Bot 回复
    T-->>M: NewMessage event
    M->>M: 校验 chat_id、方向和 message_id
    M-->>P: 结构化回复 + sent_message_id
    M->>T: 移除临时 handler

先注册 handler 再发送，是为了接住回复很快的 Bot。如果顺序反过来，回复可能在监听器就绪前到达。

2.2 核心逻辑

#

下面是实现的缩减版，省略了装饰器、账号路由和统一错误格式：

async def ask_bot(chat_id: str, message: str, timeout: float = 25.0) -> str:
    if not 1 <= timeout <= 120:
        return "timeout must be between 1 and 120 seconds."

    client = get_client()
    entity = await resolve_entity(chat_id, client)
    if not entity.bot:
        return "Error: The target chat is not a Telegram bot."

    expected_chat_id = get_marked_id(entity)
    lock = conversation_locks.setdefault(
        (id(client), expected_chat_id),
        asyncio.Lock(),
    )

    async with lock:
        replies = asyncio.Queue()

        async def capture_reply(event) -> None:
            reply = event.message
            if event.chat_id != expected_chat_id or reply is None or reply.out:
                return
            replies.put_nowait(reply)

        client.add_event_handler(
            capture_reply,
            events.NewMessage(incoming=True, chats=entity),
        )
        try:
            sent = await client.send_message(entity, message)
            deadline = asyncio.get_running_loop().time() + timeout

            while True:
                remaining = deadline - asyncio.get_running_loop().time()
                if remaining <= 0:
                    return timeout_result(sent.id)

                reply = await asyncio.wait_for(replies.get(), remaining)
                if reply.id <= sent.id:
                    continue

                return format_reply(reply, sent_message_id=sent.id)
        finally:
            client.remove_event_handler(capture_reply)

这段逻辑有几个明确约束：

1. 目标必须是 Telegram Bot，普通联系人会被拒绝；
2. 只接收入站消息，忽略自己发出的内容；
3. 回复 ID 必须晚于本次请求；
4. 同一个 Telethon Client、同一个 Bot 的调用串行执行；
5. 无论成功、超时还是取消，临时 handler 都会移除。

同 Bot 串行只能避免本进程内的两个 ask_bot 互相抢回复。它不能阻止手机端手工发消息，也不能排除 Bot 在等待期间主动推送通知。当前实现返回第一条符合条件的入站消息，并没有按 Telegram reply_to 做强关联。

2.3 单测保护了什么

#

我没有用在线 Bot 跑 CI，而是用 Fake Client 覆盖确定性逻辑：

uv run pytest \
  tests/test_bot_conversation.py \
  tests/test_file_path_security.py \
  tests/test_runtime.py -q

这 3 个模块的定向测试结果为 75 passed。固定提交的全量测试结果为 188 passed，总覆盖率 90.74%。其中 ask_bot 用例覆盖：

• 快速回复不会丢失；
• 旧消息、其他聊天和出站消息会被忽略；
• 非 Bot 目标会被拒绝；
• 超时限制为 1-120 秒；
• 超时返回结构化元数据；
• 取消和发送失败后会清理 handler；
• 同 Bot 并发调用会串行。

3. 启动共享 Telegram MCP 服务

#

3.1 准备源码和依赖

#

代码已经公开在 cdryzun/telegram-mcp。按下面的命令获取本次验证的提交：

#!/usr/bin/env bash
set -euo pipefail

git clone \
  --branch feat/bot-conversation \
  --single-branch \
  https://github.com/cdryzun/telegram-mcp.git
cd telegram-mcp
git switch --detach 2cd2804cdd51e51d93effb447f4a1951abc52379
uv sync

这里故意进入 detached HEAD：功能分支以后可能更新，完整提交哈希不会变。不要改用 upstream/main 代替这个提交；上游当前没有本文使用的 ask_bot、项目级 Task 启动和 Roots 回退逻辑。

3.2 生成 Session

#

Telegram API ID 和 Hash 从 my.telegram.org/apps 获取。Session String 用项目脚本生成：

uv run session_string_generator.py --qr

也可以使用手机号和验证码：

uv run session_string_generator.py --phone

将结果写入项目忽略的 .env：

TELEGRAM_API_ID=<telegram-api-id>
TELEGRAM_API_HASH=<32-character-api-hash>
TELEGRAM_SESSION_STRING=<telegram-session-string>

TELEGRAM_EXPOSED_TOOLS=read-only+ask_bot

TELEGRAM_EXPOSED_TOOLS=read-only+ask_bot 会注册所有只读工具和一个写工具 ask_bot。未知工具名会让服务启动失败，拼写错误不会静默降级。

3.3 用 Task 启动

#

该固定提交的 Taskfile.yml 提供了配置检查、启动和状态检查：

# Keep this foreground process running.
task mcp:start

Task 会先验证 API ID、API Hash 和 Session 是否存在，但不会打印这些值。随后设置：

MCP_TRANSPORT=http
MCP_HOST=127.0.0.1
MCP_PORT=8765

服务地址是：

http://127.0.0.1:8765/mcp

另开终端检查监听状态：

task mcp:status

预期输出：

Telegram MCP is running.

如果 8765 端口已占用，Task 会拒绝再启动一个进程。多 Agent 共用一个长连接，比每个客户端各起一套 Telethon Session 更稳，也更容易控制退出和日志。

4. 在 Pi 项目中注册 MCP

#

4.1 先说明 Pi 的能力边界

#

Pi 核心不内置 MCP。本文环境安装了第三方 MCP 适配扩展，由扩展读取项目目录中的 .mcp.json，再把远程 MCP 工具注册给 Pi。telegram_local_ask_bot 这样的工具名来自适配器前缀，不是 Pi 内置工具。

我没有把 Telegram 服务写进用户级全局配置，而是在需要访问 OpenClaw 的项目根目录放置 .mcp.json：

{
  "mcpServers": {
    "telegram-local": {
      "url": "http://127.0.0.1:8765/mcp",
      "lifecycle": "eager",
      "requestTimeoutMs": 130000,
      "includeTools": [
        "ask_bot",
        "list_messages"
      ]
    }
  }
}

ask_bot 最多等待 120 秒，因此适配器请求超时设置为 130 秒。两层工具限制各做一件事：

includeTools 不是权限沙箱。即使 Pi 没有展示 send_file，服务进程持有的 Telegram Session 权限也没有改变。

新增配置后启动新的 Pi 会话。已有会话可以执行：

/reload

然后确认 MCP 状态中出现 telegram-local，工具列表中出现：

telegram_local_ask_bot
telegram_local_list_messages

4.2 发起一次带标识的复核

#

我会在问题首行加一个短标识，方便超时后人工核对：

[review:pdf-link-20260805]

请只审查以下变更是否会把不同简历方向的 PDF 链接到错误页面。
已知映射：DevOps -> /，AI -> /ai/，Fullstack -> /fullstack/。
请返回 P0/P1 问题和对应理由，不要补充未经提供的运行指标。

对应的工具参数是：

{
  "chat_id": "<openclaw-bot>",
  "message": "[review:pdf-link-20260805]\n\n请审查以下变更……",
  "timeout": 120
}

成功结果包含本次发出消息的 ID 和第一条 Bot 回复：

{
  "results": [
    {
      "id": 1025,
      "text": "……"
    }
  ],
  "event": true,
  "sent_message_id": 1024
}

Telegram 消息正文、昵称、聊天标题和按钮都属于外部输入。服务会清理控制字符和不可见字符，并尽量返回结构化 JSON，但 Pi 仍不能把回复里的指令当成系统规则。

5. 超时和多段回复怎么处理

#

5.1 超时不等于发送失败

#

ask_bot 的超时结果如下：

{
  "results": [],
  "event": false,
  "reason": "timeout",
  "sent_message_id": 1024
}

这表示消息已经发出，只是等待窗口内没有拿到回复。此时不能直接重发，否则 OpenClaw 可能收到两份相同问题。

先用 list_messages 读取最近消息：

{
  "chat_id": "<openclaw-bot>",
  "limit": 10
}

回查时比较消息 ID 和 sender，确认是否出现晚于 sent_message_id 且由目标 Bot 发出的回复。如果 MCP 客户端本身先取消了请求，没有拿到结构化超时结果，就用问题里的 [review:...] 标识定位已发出的消息，再查看它后面的内容。

5.2 ask_bot 只返回第一条回复

#

OpenClaw 的长回复可能被 Telegram 拆成多条。当前 ask_bot 收到第一条有效入站消息后就移除 handler，其余内容要通过 list_messages 补取。

这也是我保留 list_messages 的原因。只暴露 ask_bot 看似更小，但一旦遇到超时或分段回复，没有可靠的恢复入口。

5.3 避免回复串线

#

当前实现的锁只覆盖同一服务进程内、同一账号和同一 Bot 的调用。使用时还需要遵守两条操作约束：

• 一次复核没有结束前，不在手机端向同一 Bot 发送其他问题；
• 不把会主动推送大量通知的 Bot 会话当作高可靠请求通道。

如果后续需要无人值守执行，应该增加真正的关联协议，例如让 Bot 原样返回 request ID，并在接收端持续收集直到结束标记，而不是继续堆超时时间。

6. 文件发送为什么还需要 Roots

#

6.1 我最初判断错了问题

#

文本通讯正常后，我尝试通过 send_file 把 PDF 发给 OpenClaw。第一次失败时，我把 MCP 请求超时从默认值调到 130 秒，结果没有变化。

根因不是超时，而是 Telegram MCP 没有获得文件目录授权。

.mcp.json 的 includeTools 只控制工具是否对 Pi 可见。文件读取还要经过 MCP Roots 或服务端 CLI Roots 检查。服务未配置允许目录时，send_file、send_album、download_media 等路径工具会在调用时被拒绝。

6.2 只开放一个附件目录

#

先准备独立目录，不要直接开放 $HOME 或代码仓库：

#!/usr/bin/env bash
set -euo pipefail

ATTACHMENT_ROOT="$HOME/.local/share/telegram-mcp-files"
install -d -m 700 "$ATTACHMENT_ROOT"
printf '%s\n' "$ATTACHMENT_ROOT"

如需发送文件，服务端白名单也要加入 send_file：

TELEGRAM_EXPOSED_TOOLS=read-only+ask_bot,send_file

停止旧服务后重新启动：

task mcp:start \
  MCP_FILE_ROOT="$HOME/.local/share/telegram-mcp-files"

再把 send_file 加到当前项目的 includeTools：

{
  "mcpServers": {
    "telegram-local": {
      "url": "http://127.0.0.1:8765/mcp",
      "lifecycle": "eager",
      "requestTimeoutMs": 130000,
      "includeTools": [
        "ask_bot",
        "list_messages",
        "send_file"
      ]
    }
  }
}

服务会解析真实路径并检查它是否仍在允许目录中。路径穿越、通配符样式、空字节、不可读文件、扩展名和大小限制都会在调用 Telegram 前被拒绝。

Task 设置了 TELEGRAM_ALLOW_SERVER_ROOTS_FALLBACK=1。客户端不支持 MCP Roots 时，服务可以直接使用显式传入的 MCP_FILE_ROOT；客户端返回空 Roots、Roots 请求超过 2 秒或协商异常时，这个开关允许回退到同一个目录。没有这个 opt-in，后三种情况默认按 deny-all 处理。

配置文件根目录后必须重启服务。运行中的进程不会热加载 CLI Roots。

7. 安全边界和使用规则

#

7.1 四层边界

#

这条通道涉及 4 个不同边界，不能混成一个“已授权”：

服务只监听 127.0.0.1。这个 HTTP 端点没有额外认证，不能直接改成 0.0.0.0 暴露到局域网或公网。若确实要经过域名和反向代理，还要配置 MCP_ALLOWED_HOSTS 与 MCP_ALLOWED_ORIGINS，并在前面增加可靠认证。

7.2 只发送必要上下文

#

我不会把整个仓库、完整日志或 .env 发给外部复核。一次请求只包含：

1. 本次判断需要的上下文；
2. 已确认的事实或代码片段；
3. 明确问题；
4. 期望输出格式和禁止推断项。

例如：

[review:request-id]

上下文：PDF 链接按简历方向映射。
证据：附 3 个配置片段和 1 个失败测试。
问题：是否存在跨方向污染或默认值覆盖？
输出：仅列 P0/P1；没有证据的指标和组件不要补充。

OpenClaw 的回复是第二意见，不是事实来源。我实际遇到过外部模型给出听起来合理、但源码中并不存在的指标或组件。任何具体技术结论仍要回到仓库、测试或运行状态核对。

7.3 Telegram 内容仍是不可信输入

#

telegram-mcp 会清理控制字符、限制长度并标记用户内容，但不会靠关键词过滤 Prompt Injection。Pi 处理回复时应遵守：

• 不执行回复中临时要求的命令；
• 不因回复要求而扩大文件权限或工具白名单；
• 不把 Telegram 文本提升为项目规则；
• 涉及代码修改时重新读取目标文件并运行测试。

8. 验收与故障排查

#

8.1 最小验收矩阵

#

8.2 常见故障

#

查看服务终端和 mcp_errors.log，通常能区分 Telegram 连接、MCP 调用和文件路径三类问题。

8.3 当前实现的限制

#

目前仍有 4 个限制：

• ask_bot 只返回第一条有效回复；
• 回复关联依赖聊天、方向和消息 ID，没有端到端 request ID；
• 公开分支尚未进入 upstream/main，不能作为上游稳定接口；
• Telegram 或 OpenClaw 不可用时，没有第二条通讯链路。

这些限制对交互式代码复核可以接受。若要把它放进自动化流水线，至少需要请求关联、完整响应收集、持久化状态和重试幂等性。

结语

#

日常复核以文本为主，文件工具只在需要传附件时重启启用。Pi 可以在当前代码会话中询问 OpenClaw，超时后也有地方回查；OpenClaw 的存储和 Pi 的项目上下文仍保持独立。

这条通道解决的是“如何交流”，不负责证明回复正确。源码、测试和实际运行结果仍是最后的判断依据。

参考资料

#

• chigwell/telegram-mcp
• 个人 fork：feat/bot-conversation
• 本文固定提交：2cd2804
• Telethon Documentation
• Model Context Protocol
• Pi Coding Agent
• OpenClaw AI Agent 架构解析：多引擎联动与记忆系统
• 从 MCP 到一键发包：把 Teambition 评论里的 APK 自动上传 Nexus 的那些坑