MCP 协议与工具调用 · 运维手册
适用场景：开发 MCP Server 为 AI 应用提供工具能力，或集成 MCP Client 调用外部工具服务。
1. MCP 协议概述
1.1 什么是 MCP
MCP（Model Context Protocol）是 Anthropic 提出的开放协议，让 AI 模型能够安全、标准化地调用外部工具。
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ AI Host │ │ MCP Client │ │ MCP Server │
│ (Claude等) │ ◄──►│ (集成层) │ ◄──►│ (工具服务) │
└─────────────┘ └─────────────┘ └─────────────┘
1.2 核心概念
| 概念 | 作用 | 示例 |
|---|---|---|
| Tool | 可调用的函数 | check_service, query_metrics |
| Resource | 可读取的数据源 | 文件、数据库、API 响应 |
| Prompt | 可复用的提示词模板 | 故障诊断流程、代码审查模板 |
| Server | 工具提供者 | 自定义 MCP Server |
| Client | 工具消费者 | Claude Desktop、Cursor |
2. 协议架构
2.1 通信机制
传输方式： - stdio：标准输入输出（本地进程） - HTTP：RESTful API（远程服务） - SSE：Server-Sent Events（流式响应）
2.2 消息类型
// 初始化
{"jsonrpc": "2.0", "id": 1, "method": "initialize",
"params": {"protocolVersion": "2024-11-05", "capabilities": {}, "clientInfo": {"name": "test", "version": "1.0"}}}
// 工具列表
{"jsonrpc": "2.0", "method": "tools/list"}
// 工具调用
{"jsonrpc": "2.0", "method": "tools/call",
"params": {"name": "check_service", "arguments": {"service": "nginx"}}}
// 工具返回
{"jsonrpc": "2.0", "result": {"content": [{"type": "text", "text": "nginx is running"}]}}
3. Python MCP Server 开发
3.1 环境搭建
3.2 基础 Server
# server.py
from mcp.server import Server
from mcp.types import Tool, TextContent
app = Server("ops-agent")
@app.tool()
async def check_service(service_name: str) -> list[TextContent]:
"""检查服务状态"""
import subprocess
result = subprocess.run(
["systemctl", "status", service_name],
capture_output=True, text=True
)
return [TextContent(type="text", text=result.stdout[:2000])]
@app.tool()
async def query_metrics(metric: str) -> list[TextContent]:
"""查询 Prometheus 指标"""
import requests
resp = requests.get(
f"http://prometheus:9090/api/v1/query?query={metric}"
)
return [TextContent(type="text", text=resp.text[:2000])]
if __name__ == "__main__":
app.run()
3.3 启动配置
// claude_desktop_config.json
{
"mcpServers": {
"ops-agent": {
"command": "python",
"args": ["server.py"],
"env": {
"PROMETHEUS_URL": "http://prometheus:9090"
}
}
}
}
4. 高级功能
4.1 资源订阅
@app.resource()
async def get_system_info() -> TextContent:
"""系统信息"""
import psutil
info = {
"cpu": psutil.cpu_percent(),
"memory": psutil.virtual_memory().percent,
"disk": psutil.disk_usage('/').percent
}
return TextContent(type="text", text=json.dumps(info))
4.2 流式响应
@app.tool()
async def streaming_query(query: str) -> AsyncGenerator[TextContent, None]:
"""流式查询"""
async for chunk in llm.stream(query):
yield TextContent(type="text", text=chunk)
4.3 错误处理
@app.tool()
async def safe_tool_call(name: str, args: dict) -> list[TextContent]:
"""安全的工具调用"""
try:
result = await execute_tool(name, args)
return [TextContent(type="text", text=result)]
except Exception as e:
return [TextContent(
type="text",
text=f"Error: {str(e)}",
isError=True
)]
5. 安全最佳实践
5.1 工具白名单
ALLOWED_TOOLS = {
"check_service",
"query_metrics",
"read_log",
"disk_usage"
}
@app.tool()
async def dispatch_tool(name: str, args: dict):
if name not in ALLOWED_TOOLS:
raise ValueError(f"Tool {name} not allowed")
# ... 执行逻辑
5.2 输入验证
import re
SAFE_PATTERN = re.compile(r'^[a-zA-Z0-9_\-\.]+$')
def validate_service_name(name: str) -> bool:
return bool(SAFE_PATTERN.match(name))
5.3 超时控制
import asyncio
@app.tool()
async def timed_tool(name: str, args: dict):
try:
result = await asyncio.wait_for(
execute_tool(name, args),
timeout=30.0
)
return result
except asyncio.TimeoutError:
return [TextContent(type="text", text="Request timed out", isError=True)]
6. 调试与测试
6.1 启用调试日志
6.2 本地测试
6.3 集成测试
import pytest
from mcp.client.stdio import stdio_client
@pytest.mark.asyncio
async def test_check_service():
async with stdio_client(["python", "server.py"]) as (read, write):
# 发送工具列表请求
await write('{"jsonrpc":"2.0","id":1,"method":"tools/list"}')
# 读取响应
response = await read()
assert "check_service" in response
7. 常见问题排查
问题 1：工具调用超时
排查： - 检查服务端日志 - 验证网络连通性 - 调整超时参数
问题 2：JSON-RPC 格式错误
排查： - 检查消息格式是否符合 2.0 规范 - 验证 id 字段唯一性 - 检查 content type 头
问题 3：工具返回空结果
排查： - 检查工具参数验证 - 验证数据源可达性 - 查看错误日志