MCP Server 开发实战 & Skill 生命周期管理

第一部分：MCP Server 从零开发

为什么需要自己写 MCP Server

社区有大量预建的 MCP Server（文件系统、GitHub、Slack、数据库等），但企业场景中的内部 API、专有数据源、定制业务流程仍然需要自建 Server。MCP 的价值恰恰在于让你用统一协议封装任何后端能力。

MCP Server 的最小实现

一个 MCP Server 本质上是一个进程，通过 stdio 或 HTTP SSE 与 Host 通信，暴露 Tools、Resources、Prompts 三种能力：

# Python MCP Server 最小实现
from mcp.server import Server, NotificationOptions
from mcp.server.models import InitializationCapabilities
from mcp.server.stdio import stdio_server
import mcp.types as types

# 1. 创建 Server 实例
server = Server("my-knowledge-server")

# 2. 注册工具（Tool）
@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
    return [
        types.Tool(
            name="search_knowledge",
            description="搜索内部知识库。",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "搜索关键词"
                    },
                    "top_k": {
                        "type": "integer",
                        "default": 5,
                        "minimum": 1,
                        "maximum": 20
                    }
                },
                "required": ["query"]
            }
        )
    ]

@server.call_tool()
async def handle_call_tool(
    name: str, arguments: dict
) -> list[types.TextContent | types.ImageContent]:
    if name == "search_knowledge":
        query = arguments["query"]
        top_k = arguments.get("top_k", 5)
        results = await search_kb(query, top_k)
        return [types.TextContent(
            type="text",
            text=json.dumps(results, ensure_ascii=False)
        )]
    raise ValueError(f"Unknown tool: {name}")

# 3. 注册资源（Resource）
@server.list_resources()
async def handle_list_resources() -> list[types.Resource]:
    return [
        types.Resource(
            uri="knowledge://stats",
            name="知识库统计",
            description="知识库的文档数量、分类分布等统计信息",
            mimeType="application/json"
        )
    ]

@server.read_resource()
async def handle_read_resource(uri: str) -> str:
    if uri == "knowledge://stats":
        stats = await get_kb_stats()
        return json.dumps(stats, ensure_ascii=False)
    raise ValueError(f"Unknown resource: {uri}")

# 4. 启动 Server
async def main():
    async with stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream, write_stream,
            InitializationCapabilities(
                sampling={},
                experimental={},
            )
        )

if __name__ == "__main__":
    asyncio.run(main())

MCP 配置（Host 端）

Server 写好后，需要在 Host（如 Claude Desktop）中配置：

{
  "mcpServers": {
    "my-knowledge-server": {
      "command": "python",
      "args": ["-m", "my_knowledge_server"],
      "env": {
        "API_KEY": "${MY_API_KEY}",
        "DATABASE_URL": "postgresql://..."
      }
    }
  }
}

stdio vs SSE：何时用哪个

MCP Server 的错误处理

@server.call_tool()
async def handle_call_tool(name: str, arguments: dict):
    try:
        if name == "search_knowledge":
            return await handle_search(arguments)
        elif name == "query_database":
            return await handle_query(arguments)
    except ValidationError as e:
        return [types.TextContent(
            type="text",
            text=f"参数错误：{str(e)}\n正确格式请查看工具定义。"
        )]
    except TimeoutError:
        return [types.TextContent(
            type="text",
            text="操作超时，请尝试缩小查询范围或稍后重试。"
        )]
    except Exception as e:
        # 不暴露内部错误细节给 LLM
        logger.error(f"Tool {name} failed: {e}", exc_info=True)
        return [types.TextContent(
            type="text",
            text=f"执行 {name} 时发生内部错误，已记录日志。"
        )]

MCP 的四个高阶模式

1. MCP Server 链（Chaining）

一个 MCP Server 可以作为另一个 Server 的 Client：

# 编排 Server：聚合多个子 Server 的能力
KnowledgeServer → 知识库搜索
  └─ 调用 → GitHubServer → 获取代码仓库信息
  └─ 调用 → SlackServer → 获取团队讨论记录
  └─ 返回 → 综合结果："关于 X 主题，知识库有3篇文档，
              GitHub有2个相关PR，Slack有5条讨论"

2. MCP 动态工具注册

某些场景下工具列表不是静态的（如 SaaS 平台的按租户定制的工具），可以用 list_tools 的动态实现：根据请求上下文（通过 HTTP Header 传递的 API Key）返回不同的工具列表。

3. MCP 资源流式传输

对于大文件或实时数据流，用 resources/subscribe 和 notifications/resources/updated 实现推送更新。

4. MCP Sampling（Server 请求 LLM）

Server 可以主动请求 Host 调用 LLM，用于 Agent 间的交叉验证或自省：

# Server 请求 LLM 验证自己的输出
result = await server.create_message(
    messages=[{
        "role": "user",
        "content": {
            "type": "text",
            "text": f"请验证以下搜索结果是否与查询'{query}'相关：{results}"
        }
    }],
    max_tokens=200
)

第二部分：Skill 系统架构与生命周期

Skill 系统的三层架构

一个成熟的 Agent 平台通常有三层能力组织：

┌─────────────────────────────────────────────┐
│  Layer 3: Skill（技能）                       │
│  完整的业务能力单元                            │
│  例：旅行规划、文档生成、数据分析               │
│  特点：多步骤、有状态、有降级策略               │
├─────────────────────────────────────────────┤
│  Layer 2: Tool（工具）                        │
│  可被 Agent 直接调用的单一功能                  │
│  例：搜索文档、发送邮件、查询数据库              │
│  特点：无状态、原子性、独立执行                  │
├─────────────────────────────────────────────┤
│  Layer 1: MCP Server / API（基础能力）         │
│  底层数据源和服务接口                           │
│  例：文件系统、数据库、第三方API                │
│  特点：与 Agent 解耦、标准化协议                 │
└─────────────────────────────────────────────┘

Skill 的生命周期管理

Skill 生命周期：
  注册 → 发现 → 激活 → 执行 → 结果回传 → (更新/卸载)
   │      │      │      │        │
   │      │      │      │        └─ 输出结果 + 状态持久化
   │      │      │      └─ Agent 驱动：ReAct 循环中调用工具
   │      │      └─ 上下文注入：SKILL.md 内容进入 System Prompt
   │      └─ 触发匹配：用户输入命中触发条件
   └─ 静态注册：文件系统中的 SKILL.md 被扫描加载

  热更新流程：
  1. 文件系统监控 SKILL.md 变化
  2. 变更的 Skill 重新解析 → 更新内存中的 Skill 注册表
  3. 新会话使用新版本，已有会话保持旧版本（避免状态不一致）

SKILL.md 触发词设计的工程实践

SKILL.md 的触发条件设计直接影响 Skill 的召回率：

• 关键词触发：覆盖常见表达。例：旅行 Skill → "旅行、旅游、出行、攻略、去哪玩、订机票、订酒店"
• 语义触发：用 Embedding 做意图匹配。"帮我规划一个三天两夜的行程"→ 语义相似度匹配旅行 Skill
• 排除规则：明确不触发的场景。"单纯的天气查询不要触发旅行 Skill"
• 优先级：多个 Skill 同时被触发时的仲裁规则。按 specificity（特异度）排序，更具体的优先

Skill 之间的协作模式

Skill 的版本管理与灰度发布

# Skill 版本策略
skill_registry = {
    "travel-planner": {
        "versions": {
            "v1.0.0": {"path": "skills/travel/v1", "status": "deprecated"},
            "v1.1.0": {"path": "skills/travel/v1.1", "status": "stable"},
            "v2.0.0-beta": {"path": "skills/travel/v2", "status": "beta"},
        },
        "routing": {
            "strategy": "weighted",  # 加权路由
            "weights": {"v1.1.0": 0.9, "v2.0.0-beta": 0.1}  # 10% 流量到 beta
        }
    }
}

第三部分：Agent 后端架构模式精华

后端 Agent 的五个横切关注点

无论 Agent 的业务逻辑是什么，这五个横切关注点必须被正确设计：

1. 会话管理

# 会话状态机
[IDLE] → 用户发消息 → [RUNNING] → 工具调用 → [WAITING_TOOL]
                                            ↓
                                    工具返回 → [RUNNING] → 完成 → [IDLE]

# 关键设计决策
- 会话超时：空闲 30 分钟自动清理
- 并发言论：同一会话的多个请求必须串行化
- 中断恢复：用户中断后如何优雅停止工具执行
- 会话迁移：WebSocket 断线重连后恢复状态

2. 流式响应的背压处理

SSE 流式输出时，如果前端消费慢于后端生产（移动网络、性能差的设备），需要背压机制：

• 后端用有界队列缓冲事件（如最多 100 个待发送事件）
• 队列满时丢弃旧的中间状态事件（thinking、tool_start），但保留 token 和最终结果
• 或使用 TCP backpressure 原生机制（asyncio 的 write() 返回时会等待）

3. 多模型路由

class ModelRouter:
    """根据任务类型路由到最优模型"""

    ROUTING_RULES = {
        "classification": "gpt-4o-mini",      # 便宜够用
        "summarization": "claude-haiku",       # 快速总结
        "complex_reasoning": "claude-sonnet",  # 强推理
        "code_generation": "deepseek-coder",   # 代码专长
        "creative_writing": "gpt-4o",          # 创意写作
        "tool_selection": "gpt-4o-mini",       # 工具选择不需要最强模型
        "safety_check": "gpt-4o",             # 安全审查需要强模型
    }

    async def route(self, task_type: str, budget: float):
        primary = self.ROUTING_RULES.get(task_type, "gpt-4o-mini")
        # 根据预算和可用性做 fallback
        return self.select_with_fallback(primary, budget)

4. 多租户隔离

5. 成本控制

• 按会话预算：每个会话最多消耗 $X，超出后降级到便宜模型
• 按用户配额：免费用户每天 50 次调用，付费用户 500 次
• 智能缓存：相同/相似问题直接返回缓存答案
• 模型降级链：Sonnet → Haiku → gpt-4o-mini → 本地小模型
• Token 预算：每次 Agent 推理限制总 Token 消耗（含工具调用往返）

Agent 后端的三个反模式

Agent 开发的问题排查速查表

← 返回首页