深入理解 AI Agent：设计原理与工程实践

本文集文档发布于灏天文库

协作工具 MCP 服务器

文档摘要

源文件：chapter4/collaboration-tools/README.md 协作工具 MCP 服务器 一个功能完整的 Model Context Protocol（MCP）服务器，为 AI Agent 提供一系列协作工具，包括浏览器自动化、人工介入（HITL）协助、通知推送以及定时器管理。 功能特性 🌐 浏览器自动化（基于 browser-use） 导航到指定 URL 并管理浏览器标签页 从网页中提取内容 借助 AI Agent 执行高层浏览器任务 截取屏幕快照 完整的虚拟浏览器能力 🤝 子 Agent 管理（子 Agent 管理） 以同步（等待结果）或异步（返回 ）模式生成子 Agent 向子 Agent 发送后续消息，也可取消正在运行的子 Agent

源文件：chapter4/collaboration-tools/README.md

协作工具 MCP 服务器

一个功能完整的 Model Context Protocol（MCP）服务器，为 AI Agent 提供一系列协作工具，包括浏览器自动化、人工介入（HITL）协助、通知推送以及定时器管理。

功能特性

🌐 浏览器自动化（基于 browser-use）

  • 导航到指定 URL 并管理浏览器标签页
  • 从网页中提取内容
  • 借助 AI Agent 执行高层浏览器任务
  • 截取屏幕快照
  • 完整的虚拟浏览器能力

🤝 子 Agent 管理（子 Agent 管理）

  • 以同步（等待结果）或异步（返回 task_id）模式生成子 Agent
  • 向子 Agent 发送后续消息，也可取消正在运行的子 Agent
  • 两种上下文传递策略，并支持可观测（提供上下文文本与 token 数）：
    • minimal —— 只传递任务本身，外加可选的人工挑选片段（成本最低、最隐私，但可能让子 Agent 信息不足）
    • llm_generated —— 额外进行一次 LLM 调用，从父 Agent 的轨迹中合成出一份紧凑、已做隐私过滤的交接上下文
  • 子 Agent 的系统提示使用带标签的上下文来源（[FROM_MAIN_AGENT] / [FROM_USER] / [TOOL_RESULT]），并采用标准化的 JSON 输出

👤 人工介入（HITL）

  • 针对敏感操作请求管理员批准
  • 请求人工管理员输入
  • 管理待处理的批准请求
  • 可配置超时时间与通知渠道

📧 邮件通知

  • 通过 SMTP 或 SendGrid 发送邮件
  • 支持 HTML 邮件
  • 支持抄送收件人与附件
  • 灵活的配置方式

💬 即时通讯

  • Telegram 机器人集成
  • Slack Webhook 支持
  • Discord Webhook 支持
  • 可配置默认频道

⏰ 定时器与调度

  • 设置一次性定时器
  • 创建周期性定时器
  • 取消并管理定时器
  • 定时器持久化存储
  • 定时器到期时的回调通知

安装

  1. 克隆仓库并进入项目目录：
cd projects/week4/collaboration-tools
​
  1. 安装依赖：
pip install -r requirements.txt
​
  1. 复制示例环境文件并完成配置：
cp env.example .env
# Edit .env with your configuration
​
  1. 安装 Playwright 浏览器（用于浏览器自动化）：
playwright install chromium
​

配置

通过在 .env 中设置环境变量来配置服务器：

浏览器设置

BROWSER_HEADLESS=false
BROWSER_USER_DATA_DIR=~/.config/collaboration-tools/browser
​

邮件配置

# SMTP (Gmail example)
SMTP_HOST=smtp.gmail.com
SMTP_PORT=587
SMTP_USERNAME=your-email@gmail.com
SMTP_PASSWORD=your-app-password
SMTP_FROM_EMAIL=your-email@gmail.com

# Or use SendGrid
SENDGRID_API_KEY=your-sendgrid-api-key
​

即时通讯

TELEGRAM_BOT_TOKEN=your-telegram-bot-token
TELEGRAM_DEFAULT_CHAT_ID=your-chat-id
SLACK_WEBHOOK_URL=https://hooks.slack.com/services/YOUR/WEBHOOK
DISCORD_WEBHOOK_URL=https://discord.com/api/webhooks/YOUR/WEBHOOK
​

HITL 设置

HITL_ADMIN_EMAIL=admin@example.com
HITL_TIMEOUT_SECONDS=3600
​

浏览器任务（AI Agent）

OPENAI_API_KEY=your-openai-api-key
OPENAI_MODEL=gpt-5.6-luna
​

通用 OpenRouter 回退：所有 LLM 入口（spawn_subagent、智能工具、
browser-use）都通过 src/llm_fallback.py 解析凭据。当未设置
OPENAI_API_KEY 而设置了 OPENROUTER_API_KEY 时，它们会走 OpenRouter
（base_url=https://openrouter.ai/api/v1，模型 id 映射为
provider/model 形式，例如 gpt-5.6-luna → openai/gpt-5.6-luna）。
两个 key 都未设置时，子 Agent 以确定性的离线模式运行（不会编造输出）。

用法

命令行入口 (main.py)

不启动 MCP 服务器，也可以用统一的命令行入口列出、单独调用协作工具，或运行端到端演示。
帮助信息为中文，-h 可查看任意子命令的参数：

python main.py --help            # 总览
python main.py list              # 列出全部协作工具（子 Agent / HITL / 多渠道通知）
python main.py demo              # 端到端协作演示：客服协调 Agent 处理一笔退款
python main.py subagent -h       # 子 Agent 子命令帮助
python main.py hitl -h           # HITL 子命令帮助
python main.py notify -h         # 通知子命令帮助
​

常用示例：

# 对比两种上下文传递策略（minimal vs llm_generated）
python main.py subagent compare

# 创建子 Agent（同步、最小化上下文）
python main.py subagent spawn --task "查询订单 A12345 状态" --strategy minimal --role 订单查询助手

# 关键决策请求管理员批准；--auto-approve 在后台模拟管理员应答，便于离线演示闭环
python main.py hitl approve --message "删除 1000 条记录？" --timeout 5 --auto-approve

# 多渠道通知
python main.py notify slack --message "部署完成"
​

demo 会串联三类协作工具：① 委派子 Agent 审批退款并对比上下文策略；② 大额操作
触发 HITL 审批（演示"超时前批准"与"超时保守默认"两种路径）；③ 向协作者多渠道通知结果。
其中 HITL 与通知路径完全离线可跑；子 Agent 的真实执行与 llm_generated 策略需要
OPENAI_API_KEY（未配置时会明确提示，命令仍可正常解析运行）。

运行 MCP 服务器

使用 stdio 传输启动服务器：

python src/main.py
​

也可以作为 MCP 服务器接入任意兼容 MCP 的客户端。

快速入门演示

运行快速入门演示，一睹全部功能：

python quickstart.py
​

子 Agent 上下文策略对比（对比效果）

在同一任务下用两种上下文传递策略各派生一个子 Agent，并打印差异
（交接的上下文 token 数、额外的准备成本、是否泄露隐私数据，以及两个子 Agent 各自的结果）。
需要 OPENAI_API_KEY（默认模型 gpt-5.6-luna，可用 OPENAI_MODEL 覆盖）：

export OPENAI_API_KEY=sk-...
python subagent_comparison.py
​

通常 minimal 消耗的 token 远少且从不泄露隐私字段，但子 Agent 可能返回
need_info；llm_generated 则多花一次 LLM 调用，交接更丰富、已做隐私过滤的上下文，
使子 Agent 能完成任务。

配合 Claude Desktop 使用

将其加入 Claude Desktop 的配置（claude_desktop_config.json）：

{
  "mcpServers": {
    "collaboration-tools": {
      "command": "python",
      "args": ["/path/to/collaboration-tools/src/main.py"],
      "env": {
        "OPENAI_API_KEY": "your-key-here"
      }
    }
  }
}
​

可用工具

浏览器工具

  • mcp_browser_navigate —— 导航到指定 URL
  • mcp_browser_get_content —— 获取页面内容
  • mcp_browser_execute_task —— 执行 AI 驱动的浏览器任务
  • mcp_browser_screenshot —— 截取屏幕快照
  • mcp_browser_list_tabs —— 列出所有打开的标签页

通知工具

  • mcp_send_email —— 发送邮件通知
  • mcp_send_telegram_message —— 发送 Telegram 消息
  • mcp_send_slack_message —— 发送 Slack 消息
  • mcp_send_discord_message —— 发送 Discord 消息

子 Agent 工具

  • mcp_spawn_subagent —— 生成子 Agent（同步/异步，minimal/llm_generated 上下文）
  • mcp_send_message_to_subagent —— 向子 Agent 发送后续消息
  • mcp_cancel_subagent —— 取消子 Agent
  • mcp_get_subagent_status —— 获取子 Agent 的状态/结果（用于异步）

人工介入工具

  • mcp_request_admin_approval —— 请求管理员批准
  • mcp_request_admin_input —— 请求管理员输入
  • mcp_respond_to_request —— 回应批准请求（管理员侧）
  • mcp_list_pending_requests —— 列出待处理请求

定时器工具

  • mcp_set_timer —— 设置一次性定时器
  • mcp_set_recurring_timer —— 设置周期性定时器
  • mcp_cancel_timer —— 取消定时器
  • mcp_list_timers —— 列出全部定时器
  • mcp_get_timer_status —— 获取定时器状态

用法示例

浏览器自动化

# Navigate to a website
await mcp_browser_navigate(url="https://example.com")

# Execute a complex task
await mcp_browser_execute_task(
    task="Search for AI agent tutorials on Google and extract the top 5 results"
)

# Take a screenshot
await mcp_browser_screenshot(full_page=True)
​

通知

# Send email
await mcp_send_email(
    to_email="user@example.com",
    subject="Task Completed",
    body="Your task has finished successfully!"
)

# Send Slack message
await mcp_send_slack_message(
    message="🎉 Deployment successful!"
)
​

人工介入

# Request approval for sensitive action
result = await mcp_request_admin_approval(
    request_message="Delete 1000 records from database?",
    urgent=True,
    timeout_seconds=300
)

if result["approved"]:
    # Proceed with action
    pass
​

定时器

# Set a timer
await mcp_set_timer(
    duration_seconds=300,
    timer_name="Check website",
    callback_message="Time to check the website status"
)

# Set recurring timer
await mcp_set_recurring_timer(
    interval_seconds=3600,
    max_occurrences=24,
    timer_name="Hourly health check"
)
​

架构

服务器由若干模块化组件组成：

collaboration-tools/
├── src/
│   ├── main.py              # MCP server entry point
│   ├── config.py            # Configuration management
│   ├── browser_tools.py     # Browser automation
│   ├── notification_tools.py # Email & IM notifications
│   ├── hitl_tools.py        # Human-in-the-loop
│   └── timer_tools.py       # Timer management
├── requirements.txt         # Python dependencies
├── env.example             # Example configuration
└── README.md               # This file
​

环境要求

  • Python 3.11+
  • OpenAI API key（用于浏览器 AI Agent 任务）
  • 可选：邮件/即时通讯服务的凭据
  • 浏览器自动化所需的 Playwright 浏览器

故障排查

浏览器问题

若浏览器自动化失败：

# Reinstall Playwright browsers
playwright install chromium --force
​

邮件问题

  • 对于 Gmail，请使用 App Password
  • 切勿开启"不太安全的应用访问"（改用应用专用密码）

Telegram 问题

LangChain/Pydantic 问题

若出现 "ChatOpenAI is not fully defined" 之类的报错或 Pydantic 校验错误：

  • 这是 LangChain 与 Pydantic v2 之间已知的兼容性问题
  • 解决办法：ChatOpenAI 现在改为按需初始化，只在用到时（browser_execute_task）才创建
  • 简单的浏览器导航并不需要 OpenAI API key
  • 只有自主浏览器任务（browser_execute_task）才需要 OPENAI_API_KEY

许可证

MIT License

参与贡献

欢迎贡献！请随时提交 issue 或 pull request。

支持

如有疑问或问题，请在仓库中提 issue。

作者与出处
原作者： bojieli
来源：bojieli
许可证：Apache-2.0
整理： 灏天文库整理
由灏天文库结构化整理，提供目录导航、全文检索与在线阅读，便于系统化学习
发布者: 作者: bojieli 转发
评论区 (0)
U
0/500