Skip to main content
Glama
quarktimes

MCP Agent

by quarktimes
OverviewSchemaSourceRelated ServersDiscussionsAdmin
Python
Remote

MCP Agent

从 JSON-RPC 协议层到 LLM 编排层全链路自建的 Agent 项目。

不是 LangChain/CrewAI 的封装层，而是基于 MCP SDK 构建的 Server-Client 双端系统，包含自实现的 ReAct 循环引擎、双 Provider 抽象、以及面向 Agent 场景的权限控制。

亮点

  • MCP 协议全链路掌控 — 基于官方 SDK 构建 Server/Client 双端，理解 tools/list / tools/call 的序列化、路由、生命周期管理，而非在 LangChain 里加 @Tool 注解

  • 自建 ReAct 循环 — 不依赖 Agent 框架，手动实现 Think → Act → Observe 循环，LLM 返回的 tool_use 与 MCP 的 tools/call 之间的映射关系完全透明

  • 双 Provider 架构 — 通过 Provider 模式抽象 Anthropic 和 OpenAI/DeepSeek 两种 Tool Call 消息格式的差异，ReAct 循环不感知 LLM 厂商

  • 权限层融入协议 — 四道关卡（传输层鉴权 → 工具可见性 → 调用权限 → 参数级权限）在 Server 端实现，权限拒绝用自然语言让 Agent 自愈，而非抛 403

  • 真实外部 API 集成 — OpenWeatherMap（含地理编码，支持中文城市名）+ GitHub REST API，真实 API 失败自动降级 Mock，不阻塞 Agent

  • 双传输模式 — Stdio（子进程通信，本地开发）和 SSE（HTTP 服务，远程接入）均已实现

Related MCP server: MCP Server with External Tools

架构

┌─────────────────────────────────────────────────────────────────┐
│                        用户终端                                    │
│              "北京今天天气怎么样？需要带伞吗？"                      │
└────────────────────────┬────────────────────────────────────────┘
                         │
                         ▼
┌─────────────────────────────────────────────────────────────────┐
│  client/main.py  —  交互式终端                                    │
│                                                                   │
│  ┌───────────────────────────────────────────────────────────┐   │
│  │  client/react_loop.py  —  ReAct 循环引擎                   │   │
│  │                                                           │   │
│  │   Think ──→ Act ──→ Observe ──→ Repeat                    │   │
│  │     │          │          │                               │   │
│  │     ▼          ▼          ▼                               │   │
│  │  LLMClient  MCPClient  结果送回 LLM                       │   │
│  └──────┬─────────┬──────────────────────────────────────────┘   │
│         │         │                                               │
└─────────┼─────────┼───────────────────────────────────────────────┘
          │         │
          │         │ Stdio / SSE 传输
          ▼         ▼
┌─────────────────────────────────────────────────────────────────┐
│  server/main.py  —  MCP Server                                  │
│                                                                   │
│  收到的 JSON-RPC 请求：{"method": "tools/list" | "tools/call"}    │
│                                                                   │
│  ┌──────────────────────────────────────────┐                    │
│  │  路由分发                                  │                    │
│  │                                          │                    │
│  │  @app.list_tools() → 读取注册中心 → 权限过滤 → 返回工具列表   │
│  │  @app.call_tool()  → 权限检查 → 执行 handler → 返回结果      │
│  └──────────┬───────────────────────────────┘                    │
│             │                                                    │
│             ▼                                                    │
│  ┌──────────────────────────────────────────┐                    │
│  │  server/registry.py  —  工具注册中心       │                    │
│  │  (name → {tool_def, handler} 查表路由)    │                    │
│  └──────────┬───────────────────────────────┘                    │
│             │                                                    │
│     ┌───────┼───────────────┐                                    │
│     ▼       ▼               ▼                                    │
│  get_weather  get_github_info  save_note                          │
│     │           │               │                                │
│     ▼           ▼               ▼                                │
│  OpenWeather  GitHub REST    本地文件系统                         │
│  (Geo+Weather)  API                                              │
│                                                                   │
│  ┌──────────────────────────────────────────┐                    │
│  │  server/auth/permission.py  —  权限四关   │                    │
│  │  ① 传输层鉴权  ② 工具可见性  ③ 调用权限  ④ 参数级权限     │                    │
│  └──────────────────────────────────────────┘                    │
└─────────────────────────────────────────────────────────────────┘

目录

交互流程

一次完整的 ReAct 循环

你: 北京今天天气怎么样？需要带伞吗？

              ┌──────────────────────────────────────┐
Step 1        │  LLM (Claude/DeepSeek)               │
              │  根据 tools/list 返回的 Tool 定义，     │
              │  选择 get_weather(city="北京") 并返回   │
              │  tool_use content block               │
              └──────────┬───────────────────────────┘
                         │
                         ▼
              ┌──────────────────────────────────────┐
              │  MCP Client                          │
              │  提取 tool_use → 组装 JSON-RPC       │
              │  → tools/call → MCP Server           │
              │  ① Geo API: 北京 → (39.9, 116.4)    │
              │  ② Weather API: 26°C, 多云转晴       │
              │  → 返回结果字符串                     │
              └──────────┬───────────────────────────┘
                         │
                         ▼
              ┌──────────────────────────────────────┐
Step 2        │  LLM                                 │
              │  拿到 tool_result → 继续推理           │
              │  → 返回纯文本回复                      │
              └──────────┬───────────────────────────┘
                         │
                         ▼
              用户看到: ☀️ 北京市 当前天气 26°C

多步工具链

你: 帮我查一下北京的天气，然后记一条笔记

Step 1: LLM 返回 tool_use → get_weather(city="北京")
Step 2: 结果送回 LLM → 继续推理 → 返回 tool_use → save_note(content="...", title="北京天气")
Step 3: 结果送回 LLM → 返回纯文本 → "已查好天气并保存为笔记"

模块详解

Server 层

server/main.py — MCP Server 核心入口。注册两个端点路由：

app = Server("mcp-agent")

@app.list_tools()
async def list_tools():
    """Agent 通过 tools/list 发现工具——经过权限过滤。"""
    all_tools = get_all_tools()
    visible = perm.filter_tools_by_role(role, all_tools)
    return [t["tool_def"] for t in visible]

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    """执行工具调用——先过权限，再执行 handler。"""
    if not perm.check_call_permission(user, name):
        return permission_error("你无权调用这个工具")

    tool = get_tool(name)
    result = await tool["handler"](**arguments)
    return result

关键设计：

  • 两个 @app 装饰器对应 MCP 协议的两个端点，内部注册到 self.request_handlers[RequestType] 字典

  • 权限检查在 handler 执行之前，不侵入工具逻辑

  • tool["handler"](**arguments) 用字典展开实现动态调用，Server 不需要 if-else

server/registry.py — 工具注册中心。独立为模块的原因见架构决策。

_tool_registry = {}  # {name: {tool_def, handler}}

def register_tool(name, description, input_schema):
    """装饰器：把函数注册为 MCP Tool。"""
    def decorator(func):
        _tool_registry[name] = {
            "tool_def": Tool(name=name, description=description, inputSchema=input_schema),
            "handler": func,
        }
        return func
    return decorator

Tools 层

server/tools/weather.py — 天气查询。OpenWeatherMap 两步走策略：

get_weather(city="广州")
  → Geo API: /geo/1.0/direct?q=广州 → (23.13, 113.26)
  → Weather API: /data/2.5/weather?lat=23.13&lon=113.26
  → 格式化输出：🌤️ 广州市 当前天气 32°C

  • 两步策略（参考 embabel-weather）：Geo 编码支持中文城市名，然后坐标查天气

  • 自动降级：真实 API 失败回退 Mock 数据

  • 生活建议：根据温湿度自动生成"不需要带伞""注意防晒"等提示

server/tools/github.py — GitHub 仓库/用户查询。调用 GitHub REST API，处理 404/403/超时。

server/tools/note.py — 本地笔记读写。演示 Tool 操作文件系统的场景：save_note(content, title, filename) + list_notes()。

Client 层

client/mcp_client.py — MCP 传输层封装。屏蔽 Stdio 和 SSE 的协议差异：

class MCPClient:
    async def connect(self):
        # Stdio: 作为子进程启动 python -m server.main
        #   → 获取 (read_stream, write_stream)
        #   → 创建 ClientSession → 完成 initialize 握手
        # SSE: 连接 HTTP 服务端，同上

    async def list_tools(self) -> list[dict]:
        result = await self._inner.list_tools()
        # _inner.list_tools() 内部:
        #   ① 创建 ListToolsRequest（method="tools/list" 硬编码在 Pydantic 模型中）
        #   ② 序列化为 JSON-RPC → 写入 stdin
        #   ③ 等待 stdout 响应 → 反序列化为 Python 对象
        return [{"name": t.name, "description": t.description, "inputSchema": t.inputSchema}
                for t in result.tools]

    async def call_tool(self, name, arguments) -> str:
        result = await self._inner.call_tool(name, arguments)
        return "".join(c.text for c in result.content)

client/llm_client.py — LLM Provider 系统。支持 Anthropic 和 OpenAI 兼容两种 API：

# 核心：BaseProvider → AnthropicProvider / OpenAIProvider
class AnthropicProvider(BaseProvider):
    # 调用 anthropic.messages.create()
    # 解析 response.content 中的 tool_use block

class OpenAIProvider(BaseProvider):
    # 调用 openai.chat.completions.create()
    # 解析 response.choices[0].message.tool_calls

# 自动选择：
#   LLM_PROVIDER=openai → OpenAI
#   ANTHROPIC_API_KEY=sk-ant-... → Anthropic
#   ANTHROPIC_API_KEY=sk-...（非 sk-ant-） → OpenAI（兼容 Key 混放的情况）

两种消息格式差异由 Provider 内部消化，ReAct 循环不感知：

# ReAct 循环中，通过 llm.format_tool_call() 和 format_tool_result() 抽象
messages.append(self.llm.format_tool_call(tool_call))     # Anthropic/OpenAI 各有一套
messages.append(self.llm.format_tool_result(tc_id, result))

client/react_loop.py — ReAct 循环引擎：

class ReActLoop:
    async def run(self, user_message: str) -> str:
        messages = [{"role": "user", "content": user_message}]
        tools = await self.mcp.list_tools()

        for step in range(max_steps):
            reply_text, tool_uses = await self.llm.chat(messages, tools)

            if reply_text and not tool_uses:  # 纯文本 → 终结
                return reply_text

            for tc in tool_uses:
                messages.append(self.llm.format_tool_call(tc))
                result = await self.mcp.call_tool(tc["name"], tc["input"])
                messages.append(self.llm.format_tool_result(tc["id"], result))

权限层

server/auth/permission.py — 面向 Agent 场景的四道关卡（设计思路见同名文章）：

关卡

拦截点

实现

① 传输层鉴权

连接建立时

verify_auth_token(token) → UserContext

② 工具可见性

tools/list

filter_tools_by_role() — admin 看到 5 个，readonly 看到 2 个

③ 调用权限

tools/call

check_call_permission() — 功能权限 + 频率限制(30次/分钟) + 时段限制(写工具 8:00-22:00)

④ 参数级权限

tools/call

check_resource_permission() — 参数中的资源 ID 是否在白名单内

关键设计：

  • 权限拒绝不抛异常 — 返回自然语言提示让 Agent 重试（呼应"404 和 403 不适合 Agent 场景"）

  • isError=False — 即使权限被拒也不设为 true，Agent 可以继续思考

通信协议详解

tools/list 的 JSON-RPC 链路

Client                                          Server
  │                                                │
  │  self._inner.list_tools()                      │
  │    → ListToolsRequest(method="tools/list")     │
  │    → 序列化:                                   │
  │      {"jsonrpc": "2.0", "id": 1,               │
  │       "method": "tools/list"}                  │
  │    → 写入 stdin                              │
  │                                                │
  │                             收到 stdin         │
  │                             反序列化           │
  │                             查找 handler:      │
  │    request_handlers[                            │
  │      ListToolsRequest  ← key 是 Python 类型    │
  │    ] = @app.list_tools()                       │
  │                             执行 list_tools()  │
  │                                                │
  │  ◄── 写入 stdout ── 返回工具列表 JSON-RPC      │

tools/call 的 JSON-RPC 链路

Client                                          Server
  │                                                │
  │  self._inner.call_tool("get_weather",          │
  │    {"city": "北京"})                          │
  │    → CallToolRequest(method="tools/call")      │
  │    → {"method": "tools/call",                  │
  │       "params": {"name": "get_weather",        │
  │                  "arguments": {"city": "北京"}} │
  │    → 写入 stdin                              │
  │                                                │
  │                             → 权限关卡 ③④     │
  │                             → get_tool(name)   │
  │                             → await handler(   │
  │                                 **arguments)   │
  │                             → 调 OpenWeather   │
  │                             → 返回结果         │

函数名与 method 的映射关系

ClientSession.list_tools() 的 list_tools 这个函数名不是路由的 key。路由是由 Pydantic 模型的 method literal 决定的：

class ListToolsRequest(PaginatedRequest[Literal["tools/list"]]):
    method: Literal["tools/list"] = "tools/list"    # ← 硬编码

# Server 收到 JSON-RPC 后:
# 1. 根据 "tools/list" → 反序列化为 ListToolsRequest 实例
# 2. type(instance) == ListToolsRequest → 从 request_handlers 查
# 3. 找到 @app.list_tools() 注册的 handler

Provider 系统

两种格式差异由 Provider 封装，ReAct 循环不感知：

Anthropic (Claude)

OpenAI 兼容 (DeepSeek)

Assistant 消息

content: [{type: "tool_use", id, name, input}]

tool_calls: [{id, function: {name, arguments}}]

Tool 结果

role: "user", content: [{type: "tool_result", tool_use_id}]

role: "tool", tool_call_id

SDK

anthropic

openai

自动检测逻辑（按优先级）：

1. LLM_PROVIDER=openai / =anthropic     → 直接确定
2. ANTHROPIC_API_KEY=sk-ant-...         → Anthropic
3. ANTHROPIC_API_KEY=sk-...（非ant）    → OpenAI（兼容 Key 混放）
4. OPENAI_API_KEY=sk-...                → OpenAI
5. 默认                                 → Anthropic

架构决策

1. 为什么自建 ReAct 循环，而非 LangChain/CrewAI

如果要快速投产，LangGraph + langchain-mcp-adapters 是合理选择。但本项目的核心目的不是生产一个 Agent 框架——市面上已经有足够多了——而是展示从协议层到编排层的全链路掌控能力。

关键差异在于：LangChain 把 tools/list 和 tools/call 封装在 AgentExecutor 内部，开发者看不到 JSON-RPC 的流动。本项目让每一步都透明。

2. 为什么 registry.py 要独立

python -m server.main 会把 server/main.py 加载为 __main__。当 tools/weather.py 做 from server.main import register_tool 时，Python 又加载一次 server/main.py 作为 server.server。两次加载产生两个 _tool_registry 实例——装饰器写入一个，Server 启动读取另一个。

独立为模块后，不管 from 路径如何，始终是同一个模块实例。

3. 为什么权限在 Server 端而非 Client 端

权限是资源级的——同一个 query_inventory 工具，不同用户的 warehouse_id 范围不同。Client 端无法预知参数里的权限边界。Server 端在 tools/call 时拿到具体参数值后才能检查。

Agent 场景下 403 没有意义。用自然语言告诉 Agent "为什么不行"，它才能调整参数重试。

4. 为什么工具注册用装饰器

装饰器紧邻函数定义，工具名、参数 Schema 和执行函数写在一起。不会出现"改了 inputSchema 忘了改函数签名"的问题。等价于在 dict 里注册 {name → {tool_def, handler}}，只是语法更紧凑。

5. 为什么支持双 Provider

真实场景下可能需要混用不同 LLM（成本、性能、可用性）。ReAct 循环本身不应该绑定厂商消息格式。

快速开始

安装

pip install -e .
cp .env.example .env

配置（二选一）

# 方案 A: Claude
ANTHROPIC_API_KEY=sk-ant-...

# 方案 B: DeepSeek
LLM_PROVIDER=openai
OPENAI_API_KEY=sk-...
OPENAI_BASE_URL=https://api.deepseek.com
OPENAI_MODEL=deepseek-chat

运行

# 交互式 Agent（默认 Stdio 模式——Client 自动启动 Server 子进程）
python -m client.main

# Server SSE 模式（HTTP 服务，供远程连接——仅限本地开发）
# 生产环境使用 SSE 请加反向代理（Nginx/Caddy）提供 TLS 和访问控制
python -m server.main --transport sse

示例

> 北京今天天气怎么样？需要带伞吗？
🔧 [1/10] get_weather(city='北京')
☀️ 北京市 当前天气
  - 温度：26°C（体感 25°C）
  - 天气：多云转晴
☀️ 不需要带伞。

> 查一下 quarktimes/tiny-claw-cli
🔧 [1/10] get_github_info(repo='quarktimes/tiny-claw-cli')
📦 quarktimes/tiny-claw-cli — Java Agent Harness CLI
  - ⭐ Star: 1

> 帮我查北京天气，然后记一条笔记
🔧 [1/10] get_weather(city='北京')
🔧 [2/10] save_note(content='...', title='北京天气 2026-07-11')
✅ 已查好天气并保存为笔记

文件结构

mcp-agent/
├── pyproject.toml              # 项目元数据 + 依赖
├── .env.example                # 环境变量模板
├── README.md
│
├── server/                     # MCP Server 端
│   ├── main.py                 # Server 入口 + tools/list + tools/call 路由
│   ├── registry.py             # 工具注册中心（规避 __main__ 双重导入）
│   ├── tools/
│   │   ├── weather.py          # get_weather — OpenWeatherMap (Geo+Weather)
│   │   ├── github.py           # get_github_info — GitHub REST API
│   │   └── note.py             # save_note / list_notes — 本地文件系统
│   └── auth/
│       └── permission.py       # 四道关卡权限服务
│
├── client/                     # MCP Client + Agent 端
│   ├── main.py                 # 交互式终端入口
│   ├── mcp_client.py           # MCP 传输层（Stdio / SSE）
│   ├── llm_client.py           # LLM Provider 系统（Anthropic + OpenAI）
│   └── react_loop.py           # ReAct 循环引擎
│
├── tests/
│   └── test_integration.py     # 5 项集成测试
│
└── notes/                      # save_note 输出目录

技术要点

文件

涉及的技术点

server/main.py

MCP Server 搭建、request_handlers 类型路由、命令分发模式

server/registry.py

@register_tool 装饰器、_tool_registry dict 查表、**arguments 动态调用、Python 模块导入陷阱

server/tools/weather.py

Tool 定义规范（name + description + inputSchema）、两步式 API 调用（Geo → Weather）、异常降级

server/tools/github.py

Tool 调用外部 REST API、404/403/超时处理

server/tools/note.py

Tool 操作本地文件系统、输入安全校验

server/auth/permission.py

Agent 场景权限模型、频率限制（滑动窗口）、时段限制、资源级白名单

client/mcp_client.py

AsyncExitStack 异步生命周期管理、Stdio/SSE 传输、ClientSession 握手初始化

client/llm_client.py

Provider 模式、MCP Tool → LLM Tool 格式转换、Anthropic/OpenAI 消息格式差异

client/react_loop.py

ReAct 循环流程、Think→Act→Observe 三步、Provider 无关的消息格式化

This server cannot be deployed

RayrunBuild, deploy, and operate MCP servers without becoming an infrastructure team.ray.runAd

Maintenance

ActivityStale
ResponsivenessNo issues

Resources

Related MCP Connectors

Related MCP Servers