Claude Code、Hermes Agent、Cursor 都支持 MCP 协议——但官方的 MCP Server 只有几十个。你的内部 API、数据库、SaaS 工具呢?本文教你从零写一个 MCP Server,用 67 行 Python 把飞书审批流接给 AI Agent,从此告别"这个工具 Agent 不支持"的烦恼。
为什么你必须学会写 MCP Server
2026年5月,Anthropic 的 Claude Code v2.1.158 把 Auto Mode 扩展到了 Bedrock/Vertex。OpenAI 的 GPT-5.5 通过 MCP 接了 GitHub/Linear/Slack。Hermes Agent 有了自己的 MCP Catalog(一键安装)。MCP 协议已经是 AI Agent 世界的 HTTP——它是 Agent 调用外部工具的通用语言。
但问题是:官方的 MCP Server 就那么几十个。你公司的飞书审批、Jira 自定义字段、内部 CRM 系统——不可能指望别人帮你写好。学会手搓 MCP Server,就是把你的整个业务系统变成 AI Agent 的外挂。
这篇文章给你一个「MCP Server 脚手架模板」,改 3 个参数就能把你的 REST API 变成 Agent 的工具。
MCP Server 的本质(30秒理解)
MCP 的全称是 Model Context Protocol。它定义了三样东西:
┌──────────────────────────────────────┐
│ AI Agent(客户端) │
│ Claude Code / Hermes Agent / Cursor │
└──────────┬───────────────────────────┘
│ MCP 协议(JSON-RPC over stdio)
▼
┌──────────────────────────────────────┐
│ MCP Server(你写的) │
│ 暴露 Tools / Resources / Prompts │
└──────────┬───────────────────────────┘
│ HTTP/gRPC
▼
┌──────────────────────────────────────┐
│ 你的业务系统(API/DB/飞书…) │
└──────────────────────────────────────┘
一个 MCP Server 就是一个常驻进程,通过标准输入输出(stdio)和 Agent 通信。启动时它告诉 Agent "我有哪些工具",Agent 随时调用这些工具,Server 执行并返回结果。
你只需要做三件事:
1. 定义你的工具有哪些(名称、参数、描述)
2. 实现每个工具的实际逻辑(调 API / 查数据库 / 发消息)
3. 用 stdio 和 Agent 通信(MCP SDK 帮你做了)
实战:写一个飞书审批 MCP Server
场景设定
你是 AI 创业者,团队用飞书管理审批流(请假、报销、采购)。你希望 Claude Code 或 Hermes Agent 能直接:
- 查询「待我审批」的列表 → list_pending_approvals
- 审批通过一条申请 → approve_request
- 查询某条审批的详情 → get_approval_detail
第一步:安装 MCP Python SDK
pip install mcp
MCP SDK 是 Anthropic 官方维护的 Python 库,封装了 stdio 通信、工具注册、类型定义等底层细节。
第二步:脚手架代码(67行)
#!/usr/bin/env python3
"""
飞书审批 MCP Server
暴露 3 个工具:list_pending_approvals / approve_request / get_approval_detail
"""
import json, os, httpx
from mcp.server import Server
from mcp.types import Tool, TextContent
# ========== 配置(改成你自己的)==========
FEISHU_APP_ID = os.environ.get("FEISHU_APP_ID", "")
FEISHU_APP_SECRET = os.environ.get("FEISHU_APP_SECRET", "")
FEISHU_BASE_URL = "https://open.feishu.cn/open-apis"
# ========== MCP Server 初始化 ==========
server = Server("feishu-approval")
# ========== 工具函数:获取飞书 tenant_access_token ==========
async def get_feishu_token():
"""获取飞书 tenant_access_token(有效期2小时,生产环境应加缓存)"""
async with httpx.AsyncClient() as client:
resp = await client.post(
f"{FEISHU_BASE_URL}/auth/v3/tenant_access_token/internal",
json={"app_id": FEISHU_APP_ID, "app_secret": FEISHU_APP_SECRET}
)
return resp.json()["tenant_access_token"]
# ========== 工具 1:查询待审批列表 ==========
@server.list_tools()
async def list_tools() -> list[Tool]:
return [
Tool(
name="list_pending_approvals",
description="查询当前用户待审批的飞书审批单列表,返回审批单ID、申请人、标题、提交时间",
inputSchema={
"type": "object",
"properties": {
"page_size": {
"type": "integer",
"description": "每页返回数量,默认10,最大50",
"default": 10
}
}
}
),
Tool(
name="approve_request",
description="审批通过一条飞书审批单。需要提供审批单的唯一标识 instance_code",
inputSchema={
"type": "object",
"properties": {
"instance_code": {
"type": "string",
"description": "审批单的唯一标识,可从 list_pending_approvals 的返回结果中获取"
},
"comment": {
"type": "string",
"description": "审批意见,可选",
"default": "同意"
}
},
"required": ["instance_code"]
}
),
Tool(
name="get_approval_detail",
description="查询一条飞书审批单的详细信息,包括审批历史、表单内容、当前状态",
inputSchema={
"type": "object",
"properties": {
"instance_code": {
"type": "string",
"description": "审批单的唯一标识"
}
},
"required": ["instance_code"]
}
)
]
# ========== 工具调用处理 ==========
@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
token = await get_feishu_token()
headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
async with httpx.AsyncClient(base_url=FEISHU_BASE_URL, headers=headers) as client:
if name == "list_pending_approvals":
resp = await client.get(
"/approval/v4/instances",
params={"page_size": arguments.get("page_size", 10), "user_id_type": "user_id"}
)
data = resp.json()
# 简化:只返回关键字段
items = data.get("data", {}).get("instance_list", [])
result = [
{"instance_code": i["instance_code"], "title": i["title"],
"applicant": i.get("applicant_name", "未知"), "start_time": i["start_time"]}
for i in items
]
return [TextContent(type="text", text=json.dumps(result, ensure_ascii=False, indent=2))]
elif name == "approve_request":
instance_code = arguments["instance_code"]
comment = arguments.get("comment", "同意")
resp = await client.post(
f"/approval/v4/instances/{instance_code}/approve",
json={"comment": comment, "user_id_type": "user_id"}
)
return [TextContent(type="text", text=f"✅ 审批通过: {instance_code}\n意见: {comment}")]
elif name == "get_approval_detail":
instance_code = arguments["instance_code"]
resp = await client.get(f"/approval/v4/instances/{instance_code}")
return [TextContent(type="text", text=json.dumps(resp.json(), ensure_ascii=False, indent=2))]
return [TextContent(type="text", text=f"未知工具: {name}")]
# ========== 启动 Server ==========
if __name__ == "__main__":
import asyncio
asyncio.run(server.run())
第三步:配置 Claude Code 使用你的 MCP Server
在项目根目录的 .claude/mcp.json 中添加:
{
"mcpServers": {
"feishu-approval": {
"command": "python3",
"args": ["/path/to/feishu_approval_server.py"],
"env": {
"FEISHU_APP_ID": "cli_xxx你的AppID",
"FEISHU_APP_SECRET": "你的AppSecret"
}
}
}
}
配置 Hermes Agent(config.yaml):
mcp_servers:
feishu-approval:
command: python3
args:
- /path/to/feishu_approval_server.py
env:
FEISHU_APP_ID: cli_xxx你的AppID
FEISHU_APP_SECRET: 你的AppSecret
重启 Agent 后,直接对话:
你: 查一下我有哪些待审批的飞书申请
Agent: [调用 list_pending_approvals 工具]
找到 3 条待审批:
1. 张三的报销申请(2小时前提交)
2. 李四的请假申请(5小时前提交)
3. 王五的采购申请(昨天提交)
你: 把张三的报销通过了吧
Agent: [调用 approve_request 工具]
✅ 已通过张三的报销申请
通用模板:把任意 REST API 包装成 MCP Server
你的业务系统大概率也是 REST API。把这个模板改 3 处就能用:
#!/usr/bin/env python3
"""
通用 REST API → MCP Server 模板
使用时改 3 处:TOOLS 定义、API_BASE、call_tool 里的请求逻辑
"""
from mcp.server import Server
from mcp.types import Tool, TextContent
import httpx, json, os
API_BASE = os.environ.get("MY_API_BASE_URL", "https://your-api.example.com")
API_KEY = os.environ.get("MY_API_KEY", "")
server = Server("my-api-mcp")
# ===== 改这里 1:定义你的工具 =====
TOOLS = [
Tool(
name="search_items", # 改成你的工具名
description="根据关键词搜索资源列表", # 详细描述,Agent 会根据这个决定何时调用
inputSchema={
"type": "object",
"properties": {
"keyword": {"type": "string", "description": "搜索关键词"},
"limit": {"type": "integer", "description": "返回数量上限", "default": 10}
},
"required": ["keyword"]
}
),
# 添加更多工具...
]
@server.list_tools()
async def list_tools():
return TOOLS
# ===== 改这里 2:API 认证逻辑 =====
async def get_auth_headers():
"""根据你的 API 认证方式修改"""
# 方式A: Bearer Token
return {"Authorization": f"Bearer {API_KEY}"}
# 方式B: API Key in header
# return {"X-API-Key": API_KEY}
# ===== 改这里 3:实现每个工具的 API 调用 =====
@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
headers = await get_auth_headers()
async with httpx.AsyncClient(base_url=API_BASE, headers=headers) as client:
if name == "search_items":
resp = await client.get(
"/api/v1/items", # 改成你的 API 路径
params={"q": arguments["keyword"], "limit": arguments.get("limit", 10)}
)
return [TextContent(type="text", text=json.dumps(resp.json(), ensure_ascii=False, indent=2))]
# 添加更多工具的处理逻辑...
return [TextContent(type="text", text=f"Unknown tool: {name}")]
if __name__ == "__main__":
import asyncio
asyncio.run(server.run())
生产环境必须注意的 5 个坑
经过实战(我们团队接了飞书、Notion、内部 CRM 共 4 个自定义 MCP Server),这些坑每个都踩过:
| 坑 | 症状 | 解决方案 |
|---|---|---|
| Token 不缓存 | 每次工具调用都重新获取 token,飞书 QPS 限制触发后所有调用失败 | 加内存缓存(functools.lru_cache),设置 TTL 为 token 有效期的 80% |
| 工具描述太简略 | Agent 不知道该什么时候调用,每次都问"要不要我帮你查一下?" | description 写清楚:做什么、何时用、返回什么。至少 20 个词 |
| 错误不返回有意义信息 | API 返回 500,MCP Server 直接抛异常,Agent 看到的是 "Internal error" | 用 try/except 包裹每个工具调用,返回人类可读的错误信息 |
| 环境变量泄露到日志 | MCP Server 把 API Key 打印到了 stderr | 生产环境关闭 debug 日志,敏感值用 *** 掩码 |
| 依赖版本冲突 | mcp 和 httpx 版本不兼容 |
用 pip freeze > requirements.txt 锁定版本,部署时用 pip install -r requirements.txt |
Token 缓存实现(直接复制)
from functools import lru_cache
import time
@lru_cache(maxsize=1)
def _cached_token(ttl_hash):
"""TTL hash trick: lru_cache 根据参数变化决定是否重新执行"""
return get_feishu_token_sync()
def get_token_with_cache():
# ttl_hash 每 5400 秒(90 分钟)变化一次,token 有效期 2 小时
ttl_hash = int(time.time() / 5400)
return _cached_token(ttl_hash)
你的第一个 MCP Server 路线图
| 阶段 | 内容 | 时间 |
|---|---|---|
| Level 0 | 复制上面的通用模板,改 API_BASE 和工具名,跑通 python3 server.py |
5 分钟 |
| Level 1 | 接入 Claude Code 或 Hermes Agent,对话验证 Agent 能正确调用你的工具 | 10 分钟 |
| Level 2 | 加 Token 缓存、错误处理、日志 | 30 分钟 |
| Level 3 | 打包成 pip 包或 Docker 镜像,发布到团队内部 Registry | 1 小时 |
| Level 4 | 提交到 Hermes MCP Catalog / Claude Code 社区,供其他创业者使用 | 半天 |
总结
MCP Server 不是高级魔法——它就是一个常驻 Python 进程,定义了 Agent 能调用的函数。 学会写 MCP Server,等于给了你的 AI Agent 一把万能钥匙:它能打开你公司任何一个业务系统的大门。
今天就开始行动:
1. 从飞书/Jira/Notion/你的内部 CRM 中挑一个最常用的
2. 复制上面的通用模板
3. 改 3 个地方:API 地址、工具定义、请求逻辑
4. 接到 Claude Code 试试"帮我查一下今天的待审批"
20 分钟后,你的 Agent 就会多一个超能力。
