MCP Agent
by quarktimes
README.md
# 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 服务，远程接入）均已实现
## 架构
```
┌─────────────────────────────────────────────────────────────────┐
│ 用户终端 │
│ "北京今天天气怎么样？需要带伞吗？" │
└────────────────────────┬────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ 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 — 权限四关 │ │
│ │ ① 传输层鉴权 ② 工具可见性 ③ 调用权限 ④ 参数级权限 │ │
│ └──────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
```
---
## 目录
- [交互流程](#交互流程)
- [模块详解](#模块详解)
- [Server 层](#server-层)
- [Tools 层](#tools-层)
- [Client 层](#client-层)
- [权限层](#权限层)
- [通信协议详解](#通信协议详解)
- [Provider 系统](#provider-系统)
- [架构决策](#架构决策)
- [快速开始](#快速开始)
- [文件结构](#文件结构)
- [技术要点](#技术要点)
---
## 交互流程
### 一次完整的 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 核心入口。注册两个端点路由：
```python
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`** — 工具注册中心。独立为模块的原因见[架构决策](#架构决策)。
```python
_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](https://github.com/quarktimes/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 的协议差异：
```python
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：
```python
# 核心：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 循环不感知：
```python
# 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 循环引擎：
```python
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 场景的四道关卡（设计思路见[同名文章](面试题-MCP权限设计.md)）：
| 关卡 | 拦截点 | 实现 |
|------|--------|------|
| ① 传输层鉴权 | 连接建立时 | `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 决定的：
```python
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 循环本身不应该绑定厂商消息格式。
---
## 快速开始
### 安装
```bash
pip install -e .
cp .env.example .env
```
### 配置（二选一）
```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
```
### 运行
```bash
# 交互式 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
Maintenance
ActivityStale
ResponsivenessNo issues