Skip to main content
Glama
quarktimes

MCP Agent

by quarktimes
OverviewSchemaRelated ServersScoreDiscussions
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: LW MCP Agents

架构

┌─────────────────────────────────────────────────────────────────┐
│                        用户终端                                    │
│              "北京今天天气怎么样？需要带伞吗？"                      │
└────────────────────────┬────────────────────────────────────────┘
                         │
                         ▼
┌─────────────────────────────────────────────────────────────────┐
│  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 installed

F
license - not found
-
quality - not tested
C
maintenance

How are these scores calculated?

Maintenance

–Maintainers
–Response time
–Release cycle
–Releases (12mo)
Commit activity

Resources

Unclaimed servers have limited discoverability.

Looking for Admin?

If you are the server author, to access and configure the admin panel.

Latest Blog Posts

MCP directory API

We provide all the information about MCP servers via our MCP API.

curl -X GET 'https://glama.ai/api/mcp/v1/servers/quarktimes/mcp-agent'

If you have feedback or need assistance with the MCP directory API, please join our Discord server