MCP 服务开发实战：从协议到实现的完整指南

前言：MCP 是什么？

如果你已经读过《OpenClaw Agent 实战》和《OpenClaw Skills 入门》，你会知道 Agent 通过 Function Calling 调用工具、通过 Skills 获得领域知识。

但有一个问题：如果要让 Agent 访问外部服务（比如 Linear、GitHub、数据库），每次都需要单独编写 Skills，不仅重复劳动，还难以复用。

MCP (Model Context Protocol) 就是为了解决这个问题而生的。

MCP 的核心理念

USB-C 的比喻：MCP 就像 USB-C 接口——一个标准化的连接方式。

USB-C 之前：每种设备需要专用接口（鼠标、键盘、显示器、硬盘…）
USB-C 之后：所有设备用同一个接口，即插即用
MCP 之前：每个 AI 应用需要为每个外部服务单独开发集成
MCP 之后：AI 应用用标准协议连接所有外部服务，写一次到处运行

MCP 的价值

对于开发者：

✅ 一次开发，到处集成（Claude、ChatGPT、Cursor、VS Code…）
✅ 标准化的工具接口，减少学习成本
✅ 可复用的服务器生态

对于 AI 应用：

✅ 访问丰富的数据源和工具
✅ 降低集成复杂度
✅ 提升用户体验

对于用户：

✅ 更强大的 AI 助手
✅ 数据和工具的无缝连接
✅ 统一的使用体验

MCP 协议规范

核心概念

MCP 遵循 客户端-服务器架构：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
┌─────────────────────┐
│   MCP Host          │  AI 应用（Claude Desktop、VS Code、OpenClaw）
│  (AI Application)   │
├─────────────────────┤
│  MCP Client 1       │──────┐
│  MCP Client 2       │      │
│  MCP Client 3       │      │
└─────────────────────┘      │
                              │
         ┌────────────────────┴───────────────────┐
         │                                         │
    ┌────▼─────┐                            ┌─────▼────┐
    │MCP Server│                            │MCP Server│
    │   (Local)│                            │ (Remote) │
    │  STDIO   │                            │   HTTP   │
    └──────────┘                            └──────────┘
    文件系统、数据库                          Linear、GitHub

参与者：

MCP Host：AI 应用（如 Claude Desktop、VS Code、OpenClaw）
MCP Client：Host 中的连接器，维护与服务器的连接
MCP Server：提供上下文和能力的程序

两种服务器类型：

本地服务器：使用 STDIO 传输，运行在同一台机器
远程服务器：使用 HTTP 传输，运行在远程服务器

协议层次

MCP 包含两层：

1. 传输层（Transport Layer）

负责通信机制和认证：

STDIO 传输：

使用标准输入/输出流通信
仅限本地进程，无网络开销
性能最优，适合本地工具

Streamable HTTP 传输：

HTTP POST 发送客户端消息
Server-Sent Events (SSE) 流式响应
支持远程服务器
支持 OAuth 认证

2. 数据层（Data Layer）

基于 JSON-RPC 2.0 的协议：

核心组件：

生命周期管理：连接初始化、能力协商、会话控制
服务器特性：工具（Tools）、资源（Resources）、提示（Prompts）
客户端特性：采样（Sampling）、根目录（Roots）、征询（Elicitation）
实用工具：日志、进度跟踪、通知

消息格式

所有 MCP 消息遵循 JSON-RPC 2.0 规范。

请求（Request）

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
{
  "jsonrpc": "2.0",
  "id": "request-123",
  "method": "tools/call",
  "params": {
    "name": "get_weather",
    "arguments": {
      "city": "Beijing"
    }
  }
}

字段说明：

jsonrpc: 必须为 “2.0”
id: 字符串或整数，唯一标识符
method: 方法名称（如 tools/call、resources/list）
params: 参数对象

响应（Response）

成功响应：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
{
  "jsonrpc": "2.0",
  "id": "request-123",
  "result": {
    "content": [
      {
        "type": "text",
        "text": "Beijing: Sunny, 25°C"
      }
    ]
  }
}

错误响应：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
{
  "jsonrpc": "2.0",
  "id": "request-123",
  "error": {
    "code": -32600,
    "message": "Invalid Request",
    "data": {
      "details": "Missing required parameter: city"
    }
  }
}

通知（Notification）

单向消息，无需响应：

1
2
3
4
5
6
7
8
{
  "jsonrpc": "2.0",
  "method": "notifications/message",
  "params": {
    "level": "info",
    "message": "Server started successfully"
  }
}

服务器特性

MCP 服务器可以提供三种核心能力：

1. Tools（工具）

让 AI 模型执行操作：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
{
  "name": "get_weather",
  "description": "Get current weather for a city",
  "inputSchema": {
    "type": "object",
    "properties": {
      "city": {
        "type": "string",
        "description": "City name"
      }
    },
    "required": ["city"]
  }
}

2. Resources（资源）

提供上下文数据：

1
2
3
4
5
{
  "uri": "file:///path/to/document.md",
  "name": "Project Documentation",
  "mimeType": "text/markdown"
}

3. Prompts（提示）

预定义的交互模板：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
{
  "name": "analyze_code",
  "description": "Analyze code quality",
  "arguments": [
    {
      "name": "language",
      "description": "Programming language",
      "required": true
    }
  ]
}

编写一个 MCP 服务

现在让我们用 Python 编写一个完整的 MCP 服务器。我们将实现一个天气查询服务，支持：

✅ HTTP 传输（远程访问）
✅ OAuth 认证
✅ 工具（Tools）
✅ 资源（Resources）

项目结构

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
weather-mcp-server/
├── pyproject.toml          # 项目配置
├── README.md               # 说明文档
├── weather_server/
│   ├── __init__.py
│   ├── server.py           # MCP 服务器主程序
│   ├── tools.py            # 工具实现
│   ├── resources.py        # 资源实现
│   └── auth.py             # OAuth 认证
└── config/
    └── mcp.json            # 服务器配置

1. 安装依赖

1
2
3
4
5
6
7
# 使用 uv（推荐）
uv init weather-mcp-server
cd weather-mcp-server
uv add "mcp[cli]"

# 或使用 pip
pip install "mcp[cli]"

2. 服务器主程序

weather_server/server.py：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
"""
Weather MCP Server - 天气查询 MCP 服务
支持 HTTP 传输和 OAuth 认证
"""
from mcp.server import Server
from mcp.server.sse import SseServerTransport
from mcp.server.stdio import StdioServerTransport
from starlette.applications import Starlette
from starlette.routing import Route
import logging

from .tools import register_tools
from .resources import register_resources
from .auth import OAuthProvider

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# 创建 MCP 服务器实例
app = Server("weather-mcp-server")

# 注册工具和资源
register_tools(app)
register_resources(app)

# OAuth 提供者
oauth_provider = OAuthProvider()

async def handle_sse(request):
    """处理 SSE 连接"""
    async with SseServerTransport("/messages") as (read_stream, write_stream):
        await app.run(
            read_stream,
            write_stream,
            app.create_initialization_options()
        )

async def handle_messages(request):
    """处理客户端消息"""
    async with SseServerTransport("/messages") as (read_stream, write_stream):
        await app.run(
            read_stream,
            write_stream,
            app.create_initialization_options()
        )

# 创建 Starlette 应用
starlette_app = Starlette(
    debug=True,
    routes=[
        Route("/sse", endpoint=handle_sse),
        Route("/messages", endpoint=handle_messages, methods=["POST"]),
    ],
)

def run_http():
    """运行 HTTP 服务器"""
    import uvicorn
    uvicorn.run(starlette_app, host="0.0.0.0", port=8000)

def run_stdio():
    """运行 STDIO 服务器"""
    import asyncio
    asyncio.run(
        app.run(
            StdioServerTransport(),
            app.create_initialization_options()
        )
    )

if __name__ == "__main__":
    import sys
    if len(sys.argv) > 1 and sys.argv[1] == "--stdio":
        run_stdio()
    else:
        run_http()

3. 工具实现

weather_server/tools.py：

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
"""
工具实现：天气查询
"""
from mcp.types import Tool, TextContent
import requests
from typing import Any

async def register_tools(app):
    """注册工具到服务器"""

    @app.list_tools()
    async def list_tools():
        """返回可用工具列表"""
        return [
            Tool(
                name="get_weather",
                description="Get current weather information for a city",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "city": {
                            "type": "string",
                            "description": "City name (e.g., 'Beijing', 'Shanghai')"
                        }
                    },
                    "required": ["city"]
                }
            ),
            Tool(
                name="get_forecast",
                description="Get 5-day weather forecast for a city",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "city": {
                            "type": "string",
                            "description": "City name"
                        },
                        "days": {
                            "type": "integer",
                            "description": "Number of days (1-5)",
                            "default": 5
                        }
                    },
                    "required": ["city"]
                }
            )
        ]

    @app.call_tool()
    async def call_tool(name: str, arguments: Any):
        """执行工具调用"""
        if name == "get_weather":
            city = arguments.get("city")
            weather_data = await get_weather_data(city)

            return [
                TextContent(
                    type="text",
                    text=f"Weather in {city}:\n"
                         f"Temperature: {weather_data['temp']}°C\n"
                         f"Condition: {weather_data['condition']}\n"
                         f"Humidity: {weather_data['humidity']}%\n"
                         f"Wind: {weather_data['wind']} km/h"
                )
            ]

        elif name == "get_forecast":
            city = arguments.get("city")
            days = arguments.get("days", 5)
            forecast_data = await get_forecast_data(city, days)

            forecast_text = f"{days}-day forecast for {city}:\n"
            for day in forecast_data:
                forecast_text += f"\n{day['date']}: {day['condition']}, {day['temp_min']}-{day['temp_max']}°C"

            return [
                TextContent(
                    type="text",
                    text=forecast_text
                )
            ]

        else:
            raise ValueError(f"Unknown tool: {name}")

async def get_weather_data(city: str) -> dict:
    """
    获取天气数据
    实际项目中应该调用真实的天气 API
    这里使用模拟数据
    """
    # 模拟 API 调用
    # 实际使用时替换为 OpenWeatherMap、WeatherAPI 等服务
    weather_database = {
        "beijing": {
            "temp": 22,
            "condition": "Sunny",
            "humidity": 45,
            "wind": 12
        },
        "shanghai": {
            "temp": 25,
            "condition": "Partly Cloudy",
            "humidity": 65,
            "wind": 15
        },
        "shenzhen": {
            "temp": 28,
            "condition": "Rainy",
            "humidity": 80,
            "wind": 8
        }
    }

    return weather_database.get(city.lower(), {
        "temp": 20,
        "condition": "Unknown",
        "humidity": 50,
        "wind": 10
    })

async def get_forecast_data(city: str, days: int) -> list:
    """
    获取天气预报数据
    模拟实现
    """
    forecast = []
    for i in range(days):
        forecast.append({
            "date": f"2026-03-{8+i:02d}",
            "condition": ["Sunny", "Cloudy", "Rainy", "Partly Cloudy"][i % 4],
            "temp_min": 18 + i,
            "temp_max": 25 + i
        })

    return forecast

4. 资源实现

weather_server/resources.py：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
"""
资源实现：天气报告
"""
from mcp.types import Resource, TextResourceContents
from datetime import datetime

async def register_resources(app):
    """注册资源到服务器"""

    @app.list_resources()
    async def list_resources():
        """返回可用资源列表"""
        return [
            Resource(
                uri="weather://report/daily",
                name="Daily Weather Report",
                description="Today's weather summary for major cities",
                mimeType="text/markdown"
            ),
            Resource(
                uri="weather://alerts/active",
                name="Active Weather Alerts",
                description="Current weather warnings and advisories",
                mimeType="application/json"
            )
        ]

    @app.read_resource()
    async def read_resource(uri: str):
        """读取资源内容"""
        if uri == "weather://report/daily":
            report = generate_daily_report()
            return TextResourceContents(
                uri=uri,
                mimeType="text/markdown",
                text=report
            )

        elif uri == "weather://alerts/active":
            alerts = get_active_alerts()
            return TextResourceContents(
                uri=uri,
                mimeType="application/json",
                text=str(alerts)  # 实际应返回 JSON 字符串
            )

        else:
            raise ValueError(f"Unknown resource: {uri}")

def generate_daily_report() -> str:
    """
    生成每日天气报告
    """
    return f"""# Daily Weather Report - {datetime.now().strftime('%Y-%m-%d')}

## Major Cities Summary

### Beijing
- **Temperature**: 22°C
- **Condition**: Sunny
- **Air Quality**: Good

### Shanghai
- **Temperature**: 25°C
- **Condition**: Partly Cloudy
- **Air Quality**: Moderate

### Shenzhen
- **Temperature**: 28°C
- **Condition**: Rainy
- **Air Quality**: Good

## Travel Recommendations

- **Beijing**: Perfect for outdoor activities
- **Shanghai**: Carry an umbrella just in case
- **Shenzhen**: Indoor activities recommended
"""

def get_active_alerts() -> dict:
    """
    获取活动警报
    """
    return {
        "alerts": [
            {
                "city": "Shenzhen",
                "type": "Heavy Rain Warning",
                "severity": "moderate",
                "expires": "2026-03-08T18:00:00Z"
            }
        ],
        "lastUpdated": datetime.now().isoformat()
    }

5. OAuth 认证实现

weather_server/auth.py：

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
"""
OAuth 2.0 认证实现
"""
from mcp.server.auth import OAuthProvider
from starlette.responses import JSONResponse
from typing import Optional
import secrets
import time

class SimpleOAuthProvider(OAuthProvider):
    """
    简单的 OAuth 提供者实现
    实际项目中应该使用成熟的 OAuth 库（如 authlib）
    """

    def __init__(self):
        # 存储授权码和访问令牌
        # 实际项目中应使用数据库
        self.authorization_codes = {}
        self.access_tokens = {}
        self.clients = {
            "mcp-client": {
                "client_secret": "secret-key-change-in-production",
                "redirect_uris": ["http://localhost:3000/callback"]
            }
        }

    async def get_client(self, client_id: str) -> Optional[dict]:
        """获取客户端信息"""
        return self.clients.get(client_id)

    async def authorize(
        self,
        client_id: str,
        redirect_uri: str,
        scope: str,
        state: str
    ) -> str:
        """
        生成授权码
        实际项目中应该：
        1. 验证用户身份（登录）
        2. 让用户授权
        3. 生成安全的授权码
        """
        # 验证 redirect_uri
        client = await self.get_client(client_id)
        if not client or redirect_uri not in client["redirect_uris"]:
            raise ValueError("Invalid client or redirect URI")

        # 生成授权码
        code = secrets.token_urlsafe(32)
        self.authorization_codes[code] = {
            "client_id": client_id,
            "redirect_uri": redirect_uri,
            "scope": scope,
            "expires_at": time.time() + 600  # 10 分钟有效期
        }

        return code

    async def exchange_code(
        self,
        code: str,
        client_id: str,
        client_secret: str,
        redirect_uri: str
    ) -> dict:
        """
        用授权码交换访问令牌
        """
        # 验证授权码
        code_data = self.authorization_codes.get(code)
        if not code_data:
            raise ValueError("Invalid authorization code")

        # 检查是否过期
        if time.time() > code_data["expires_at"]:
            del self.authorization_codes[code]
            raise ValueError("Authorization code expired")

        # 验证客户端
        if (code_data["client_id"] != client_id or
            code_data["redirect_uri"] != redirect_uri):
            raise ValueError("Invalid client credentials")

        # 验证客户端密钥
        client = await self.get_client(client_id)
        if client["client_secret"] != client_secret:
            raise ValueError("Invalid client secret")

        # 删除已使用的授权码
        del self.authorization_codes[code]

        # 生成访问令牌
        access_token = secrets.token_urlsafe(32)
        self.access_tokens[access_token] = {
            "client_id": client_id,
            "scope": code_data["scope"],
            "expires_at": time.time() + 3600  # 1 小时有效期
        }

        return {
            "access_token": access_token,
            "token_type": "Bearer",
            "expires_in": 3600,
            "scope": code_data["scope"]
        }

    async def validate_token(self, token: str) -> Optional[dict]:
        """
        验证访问令牌
        """
        token_data = self.access_tokens.get(token)
        if not token_data:
            return None

        # 检查是否过期
        if time.time() > token_data["expires_at"]:
            del self.access_tokens[token]
            return None

        return token_data

    async def revoke_token(self, token: str):
        """撤销访问令牌"""
        if token in self.access_tokens:
            del self.access_tokens[token]

6. 运行服务器

HTTP 模式（远程访问）：

1
2
3
4
python -m weather_server.server
# 服务器运行在 http://localhost:8000
# SSE 端点: http://localhost:8000/sse
# 消息端点: http://localhost:8000/messages

STDIO 模式（本地访问）：

1
2
python -m weather_server.server --stdio
# 通过标准输入/输出通信

配置与使用

使用 mcporter 配置

HTTP 服务器配置

1
2
3
4
5
6
# 添加远程 MCP 服务器
mcporter config add weather-server http://localhost:8000/mcp

# 如果需要认证，添加授权头
mcporter config add weather-server http://localhost:8000/mcp \
  --header "Authorization: Bearer YOUR_TOKEN"

STDIO 服务器配置

1
2
# 添加本地 MCP 服务器
mcporter config add weather-local --command "python -m weather_server.server --stdio"

配置文件示例

config/mcporter.json：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
{
  "mcpServers": {
    "weather-server": {
      "url": "http://localhost:8000/mcp",
      "headers": {
        "Authorization": "Bearer YOUR_ACCESS_TOKEN"
      }
    },
    "weather-local": {
      "command": "python",
      "args": ["-m", "weather_server.server", "--stdio"],
      "env": {
        "WEATHER_API_KEY": "your-api-key"
      }
    }
  }
}

在 OpenClaw 中使用

1. 查看可用工具

1
mcporter list weather-server --schema

输出：

1
2
3
4
5
6
7
weather-server
  Tools:
    - get_weather: Get current weather information for a city
    - get_forecast: Get 5-day weather forecast for a city
  Resources:
    - weather://report/daily: Daily Weather Report
    - weather://alerts/active: Active Weather Alerts

2. 调用工具

1
2
3
4
5
# 查询北京天气
mcporter call weather-server.get_weather city="Beijing"

# 获取 3 天预报
mcporter call weather-server.get_forecast city="Shanghai" days=3

3. 读取资源

1
2
# 读取每日报告
mcporter call weather-server.resources/read uri="weather://report/daily"

OAuth 认证流程

如果你的 MCP 服务器需要认证：

1
2
3
4
5
6
7
8
9
# 1. 启动 OAuth 流程
mcporter auth weather-server

# 2. 浏览器会自动打开授权页面
# 3. 用户登录并授权
# 4. 自动获取访问令牌并保存

# 重置认证
mcporter auth weather-server --reset

OAuth 流程详解：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
1. Client → Authorization Server
   GET /authorize?
     client_id=mcp-client&
     redirect_uri=http://localhost:3000/callback&
     scope=read:weather&
     state=random_state

2. User Authorizes
   用户登录并同意授权

3. Authorization Server → Client
   HTTP/1.1 302 Found
   Location: http://localhost:3000/callback?
     code=AUTHORIZATION_CODE&
     state=random_state

4. Client → Authorization Server
   POST /token
   Content-Type: application/x-www-form-urlencoded

   grant_type=authorization_code&
   code=AUTHORIZATION_CODE&
   client_id=mcp-client&
   client_secret=secret-key&
   redirect_uri=http://localhost:3000/callback

5. Authorization Server → Client
   {
     "access_token": "ACCESS_TOKEN",
     "token_type": "Bearer",
     "expires_in": 3600,
     "scope": "read:weather"
   }

6. Client → MCP Server (with token)
   POST /messages
   Authorization: Bearer ACCESS_TOKEN
   {
     "jsonrpc": "2.0",
     "id": "request-123",
     "method": "tools/call",
     "params": { ... }
   }

最佳实践

安全性考虑

1. 认证与授权

HTTP 服务器：

✅ 使用 OAuth 2.0 进行用户认证
✅ 使用 HTTPS 加密通信
✅ 验证 redirect_uri 防止开放重定向
✅ 使用短期访问令牌（1 小时）
✅ 实现令牌撤销机制

STDIO 服务器：

✅ 从环境变量读取凭据
✅ 不在命令行参数中传递敏感信息
✅ 验证调用者权限

2. 输入验证

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
@app.call_tool()
async def call_tool(name: str, arguments: Any):
    # 验证工具名称
    if name not in ["get_weather", "get_forecast"]:
        raise ValueError(f"Unknown tool: {name}")

    # 验证参数类型
    if not isinstance(arguments, dict):
        raise ValueError("Arguments must be a dictionary")

    # 验证必需参数
    if "city" not in arguments:
        raise ValueError("Missing required parameter: city")

    # 清理输入（防止注入攻击）
    city = arguments["city"].strip()
    if not city.isalnum():
        raise ValueError("Invalid city name")

3. 错误处理

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
@app.call_tool()
async def call_tool(name: str, arguments: Any):
    try:
        # 执行工具逻辑
        result = await execute_tool(name, arguments)
        return [TextContent(type="text", text=result)]

    except ValueError as e:
        # 参数错误
        raise McpError(
            code=-32602,  # Invalid params
            message=str(e)
        )

    except ExternalAPIError as e:
        # 外部 API 错误
        logger.error(f"External API error: {e}")
        raise McpError(
            code=-32001,  # Application error
            message="Weather service temporarily unavailable"
        )

    except Exception as e:
        # 未知错误
        logger.exception("Unexpected error")
        raise McpError(
            code=-32603,  # Internal error
            message="Internal server error"
        )

性能优化

1. 缓存

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
from functools import lru_cache
import time

@lru_cache(maxsize=100)
def get_cached_weather(city: str, cache_key: int):
    """
    缓存天气数据
    cache_key 基于时间，实现 10 分钟缓存
    """
    return fetch_weather_from_api(city)

async def get_weather_data(city: str):
    # 10 分钟缓存
    cache_key = int(time.time() / 600)
    return get_cached_weather(city, cache_key)

2. 异步处理

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
import asyncio

async def get_weather_batch(cities: list[str]) -> list[dict]:
    """并发查询多个城市天气"""
    tasks = [get_weather_data(city) for city in cities]
    results = await asyncio.gather(*tasks, return_exceptions=True)

    # 处理异常
    return [
        {"city": city, "data": result}
        if not isinstance(result, Exception)
        else {"city": city, "error": str(result)}
        for city, result in zip(cities, results)
    ]

3. 连接复用

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
import aiohttp

# 全局会话（复用连接）
session = None

async def get_session():
    global session
    if session is None:
        session = aiohttp.ClientSession()
    return session

async def fetch_weather_from_api(city: str):
    session = await get_session()
    async with session.get(f"https://api.weather.com/{city}") as resp:
        return await resp.json()

日志与监控

1. 结构化日志

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
import logging
import json
from datetime import datetime

class StructuredLogger:
    def __init__(self, name):
        self.logger = logging.getLogger(name)

    def log(self, level: str, message: str, **kwargs):
        log_data = {
            "timestamp": datetime.now().isoformat(),
            "level": level,
            "message": message,
            **kwargs
        }
        self.logger.log(
            getattr(logging, level.upper()),
            json.dumps(log_data)
        )

# 使用
logger = StructuredLogger("weather-mcp")
logger.log(
    "info",
    "Tool called",
    tool_name="get_weather",
    city="Beijing",
    duration_ms=150
)

2. 性能监控

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
import time
from functools import wraps

def monitor_performance(func):
    @wraps(func)
    async def wrapper(*args, **kwargs):
        start = time.time()
        try:
            result = await func(*args, **kwargs)
            duration = (time.time() - start) * 1000
            logger.log(
                "info",
                "Tool execution completed",
                tool_name=func.__name__,
                duration_ms=duration,
                status="success"
            )
            return result
        except Exception as e:
            duration = (time.time() - start) * 1000
            logger.log(
                "error",
                "Tool execution failed",
                tool_name=func.__name__,
                duration_ms=duration,
                status="error",
                error=str(e)
            )
            raise
    return wrapper

@app.call_tool()
@monitor_performance
async def call_tool(name: str, arguments: Any):
    # ...

总结

核心要点

MCP 是标准化的连接协议
- 一次开发，到处集成
- 降低 AI 应用的集成成本
- 构建可复用的服务器生态
协议层次清晰
- 传输层：STDIO（本地）或 HTTP（远程）
- 数据层：JSON-RPC 2.0 消息格式
- 安全层：OAuth 2.0 认证
Python SDK 简化开发
- 装饰器注册工具和资源
- 自动处理协议细节
- 支持多种传输方式
安全性和性能至关重要
- 严格的输入验证
- 完善的错误处理
- 合理的缓存策略
- 详细的日志监控

下一步

📚 探索 MCP 官方文档
🔧 查看 Python SDK 示例
🌐 浏览 MCP 服务器列表
💡 在你的项目中集成 MCP

资源链接

官方文档：https://modelcontextprotocol.io
协议规范：https://modelcontextprotocol.io/specification/latest
Python SDK：https://github.com/modelcontextprotocol/python-sdk
TypeScript SDK：https://github.com/modelcontextprotocol/typescript-sdk
服务器示例：https://github.com/modelcontextprotocol/servers
OpenClaw 文档：https://docs.openclaw.ai

MCP 正在成为 AI 应用与外部世界连接的标准协议。掌握 MCP 开发，意味着你编写的工具可以被 Claude、ChatGPT、Cursor、VS Code 等所有主流 AI 应用使用。

开始你的 MCP 之旅吧！🚀

前言：MCP 是什么？#

MCP 的核心理念#

MCP 的价值#

MCP 协议规范#

核心概念#

协议层次#

1. 传输层（Transport Layer）#

2. 数据层（Data Layer）#

消息格式#

请求（Request）#

响应（Response）#

通知（Notification）#

服务器特性#

1. Tools（工具）#

2. Resources（资源）#

3. Prompts（提示）#

编写一个 MCP 服务#

项目结构#

1. 安装依赖#

2. 服务器主程序#

3. 工具实现#

4. 资源实现#

5. OAuth 认证实现#

6. 运行服务器#

配置与使用#

使用 mcporter 配置#

HTTP 服务器配置#

STDIO 服务器配置#

配置文件示例#

在 OpenClaw 中使用#

1. 查看可用工具#

2. 调用工具#

3. 读取资源#

OAuth 认证流程#

最佳实践#

安全性考虑#

1. 认证与授权#

2. 输入验证#

3. 错误处理#

性能优化#

1. 缓存#

2. 异步处理#

3. 连接复用#

日志与监控#

1. 结构化日志#

2. 性能监控#

总结#

核心要点#

下一步#

资源链接#