Agent工坊

【Agent工坊】手搓MCP Server:20分钟用Python把任何API变成AI Agent的超能力

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 日志,敏感值用 *** 掩码
依赖版本冲突 mcphttpx 版本不兼容 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 就会多一个超能力。


AI创业 #MCP协议 #Agent工坊 #一人公司 #AI工具开发