一、引言：为什么需要MCP协议？

介绍AI Agent在实际开发中面临的工具集成痛点，引出MCP（Model Context Protocol）协议诞生的背景与价值。

• AI Agent与外部工具交互的常见挑战
• MCP协议的核心定位：标准化工具调用接口
• 本文目标与读者预期收获

二、MCP协议核心概念速览

快速建立对MCP协议的基础认知，为后续实战打下理论基础。

• 协议架构概览：Client、Server、Transport层
• 核心通信模型：请求-响应与通知机制
• 关键数据结构：Tool、Resource、Prompt的定义
• 传输层选择：stdio vs SSE

三、开发环境准备

搭建MCP开发所需的基础环境与工具链。

• 语言与SDK选择：Python（官方SDK）vs TypeScript
• 安装MCP SDK与依赖
• 项目目录结构设计
• 调试工具推荐：MCP Inspector

四、实战一：编写第一个MCP Server

从零实现一个简单的MCP Server，暴露一个计算器工具。

• 初始化Server实例
• 注册Tool：定义名称、描述、输入Schema
• 实现Tool处理函数
• 启动Server并验证连通性

下面是一个完整的Python MCP Server示例，实现了一个加法计算器工具：

import asyncio
from mcp.server import Server, NotificationOptions
from mcp.server.models import InitializationOptions
from mcp.types import Tool, TextContent
from pydantic import BaseModel, Field
定义加法工具的输入参数Schema
class AddInput(BaseModel):
"""加法计算器的输入参数"""
a: float = Field(description="第一个加数")
b: float = Field(description="第二个加数")
初始化MCP Server实例
server = Server("calculator-server")
@server.list_tools()
async def handle_list_tools() -> list[Tool]:
"""注册工具列表：向Client暴露可用的工具"""
return [
Tool(
name="add",
description="计算两个数字的和",
inputSchema=AddInput.model_json_schema(),
)
]
@server.call_tool()
async def handle_call_tool(
name: str, arguments: dict
) -> list[TextContent]:
"""实现工具处理函数：根据工具名称分发到具体逻辑"""
if name != "add":
raise ValueError(f"未知工具: {name}")
# 解析输入参数
a = arguments.get("a", 0)
b = arguments.get("b", 0)
result = a + b
返回计算结果
return [TextContent(type="text", text=str(result))]
async def main():
"""启动MCP Server，使用stdio传输层"""
async with server.run_stdio():
print("MCP Server已启动，等待Client连接...", flush=True)
await asyncio.Future()  # 保持Server运行
if name == "main":
asyncio.run(main())

代码说明：

• 初始化Server：通过Server("calculator-server")创建实例，指定Server名称。
• 注册Tool：使用@server.list_tools()装饰器定义工具列表，每个Tool包含名称、描述和输入Schema。
• 实现处理函数：使用@server.call_tool()装饰器处理工具调用请求，根据工具名称分发逻辑。
• 启动Server：通过server.run_stdio()使用stdio传输层启动，等待Client连接。

五、实战二：构建文件系统工具链

实现一个具备文件读写、搜索能力的MCP Server，作为Agent的本地文件助手。

• 设计工具集：read_file、write_file、search_files
• 处理文件路径安全与权限校验
• 支持大文件分块读取
• 错误处理与异常返回规范

下面是一个完整的Python MCP Server示例，实现了文件系统的读写与搜索功能：

import asyncio
import os
import fnmatch
from pathlib import Path
from typing import Optional
from mcp.server import Server, NotificationOptions
from mcp.server.models import InitializationOptions
from mcp.types import Tool, TextContent
from pydantic import BaseModel, Field
定义各工具的输入参数Schema
class ReadFileInput(BaseModel):
"""读取文件内容的输入参数"""
path: str = Field(description="文件路径（绝对路径或相对于工作目录的路径）")
encoding: str = Field(default="utf-8", description="文件编码，默认为utf-8")
max_size: int = Field(default=1048576, description="最大读取字节数，默认1MB")
class WriteFileInput(BaseModel):
"""写入文件内容的输入参数"""
path: str = Field(description="文件路径（绝对路径或相对于工作目录的路径）")
content: str = Field(description="要写入的文件内容")
encoding: str = Field(default="utf-8", description="文件编码，默认为utf-8")
append: bool = Field(default=False, description="是否追加模式，默认为覆盖写入")
class SearchFilesInput(BaseModel):
"""搜索文件的输入参数"""
pattern: str = Field(description="文件匹配模式，支持通配符，如 .py、data_.csv")
directory: str = Field(default=".", description="搜索目录，默认为当前工作目录")
recursive: bool = Field(default=True, description="是否递归搜索子目录")
初始化MCP Server实例
server = Server("filesystem-server")
安全路径校验：防止路径穿越攻击
ALLOWED_BASE_DIRS = [
os.path.abspath("."),  # 当前工作目录
]
def is_path_safe(requested_path: str) -> Optional[str]:
"""校验路径是否安全，返回规范化后的绝对路径，不安全则返回None"""
try:
resolved = os.path.abspath(os.path.normpath(requested_path))
for base in ALLOWED_BASE_DIRS:
if resolved.startswith(base):
return resolved
return None
except Exception:
return None
@server.list_tools()
async def handle_list_tools() -> list[Tool]:
"""注册工具列表：向Client暴露文件系统工具集"""
return [
Tool(
name="read_file",
description="读取指定文件的内容，支持指定编码和大小限制",
inputSchema=ReadFileInput.model_json_schema(),
),
Tool(
name="write_file",
description="向指定文件写入内容，支持覆盖和追加模式",
inputSchema=WriteFileInput.model_json_schema(),
),
Tool(
name="search_files",
description="在指定目录中搜索匹配模式的文件，支持通配符和递归搜索",
inputSchema=SearchFilesInput.model_json_schema(),
),
]
@server.call_tool()
async def handle_call_tool(
name: str, arguments: dict
) -> list[TextContent]:
"""实现工具处理函数：根据工具名称分发到具体逻辑"""
if name == "read_file":
return await handle_read_file(arguments)
elif name == "write_file":
return await handle_write_file(arguments)
elif name == "search_files":
return await handle_search_files(arguments)
else:
raise ValueError(f"未知工具: {name}")
async def handle_read_file(arguments: dict) -> list[TextContent]:
"""处理读取文件请求"""
path = arguments.get("path", "")
encoding = arguments.get("encoding", "utf-8")
max_size = arguments.get("max_size", 1048576)
safe_path = is_path_safe(path)
if not safe_path:
    return [TextContent(type="text", text=f"错误：路径 '{path}' 不在允许的访问范围内")]
if not os.path.isfile(safe_path):
return [TextContent(type="text", text=f"错误：文件 '{path}' 不存在或不是普通文件")]
try:
file_size = os.path.getsize(safe_path)
if file_size > max_size:
return [TextContent(
type="text",
text=f"错误：文件大小 {file_size} 字节超过限制 {max_size} 字节，请增大max_size参数"
)]
with open(safe_path, "r", encoding=encoding) as f:
    content = f.read()
return [TextContent(type="text", text=content)]
except UnicodeDecodeError:
return [TextContent(type="text", text=f"错误：无法使用编码 '{encoding}' 解码文件，请尝试其他编码")]
except Exception as e:
return [TextContent(type="text", text=f"读取文件失败: {str(e)}")]
async def handle_write_file(arguments: dict) -> list[TextContent]:
"""处理写入文件请求"""
path = arguments.get("path", "")
content = arguments.get("content", "")
encoding = arguments.get("encoding", "utf-8")
append = arguments.get("append", False)
safe_path = is_path_safe(path)
if not safe_path:
return [TextContent(type="text", text=f"错误：路径 '{path}' 不在允许的访问范围内")]
try:
确保父目录存在
os.makedirs(os.path.dirname(safe_path), exist_ok=True)
mode = "a" if append else "w"
with open(safe_path, mode, encoding=encoding) as f:
f.write(content)
action = "追加" if append else "写入"
return [TextContent(type="text", text=f"成功{action}文件: {path}")]
except Exception as e:
return [TextContent(type="text", text=f"写入文件失败: {str(e)}")]
async def handle_search_files(arguments: dict) -> list[TextContent]:
"""处理搜索文件请求"""
pattern = arguments.get("pattern", "")
directory = arguments.get("directory", ".")
recursive = arguments.get("recursive", True)
safe_dir = is_path_safe(directory)
if not safe_dir:
return [TextContent(type="text", text=f"错误：目录 '{directory}' 不在允许的访问范围内")]
if not os.path.isdir(safe_dir):
return [TextContent(type="text", text=f"错误：'{directory}' 不是有效的目录")]
try:
matched_files = []
if recursive:
for root, dirs, files in os.walk(safe_dir):
for filename in files:
if fnmatch.fnmatch(filename, pattern):
full_path = os.path.join(root, filename)
matched_files.append(full_path)
else:
for entry in os.listdir(safe_dir):
full_path = os.path.join(safe_dir, entry)
if os.path.isfile(full_path) and fnmatch.fnmatch(entry, pattern):
matched_files.append(full_path)
if not matched_files:
return [TextContent(type="text", text=f"未找到匹配 '{pattern}' 的文件")]
result = f"找到 {len(matched_files)} 个匹配文件:\n" + "\n".join(matched_files)
return [TextContent(type="text", text=result)]
except Exception as e:
return [TextContent(type="text", text=f"搜索文件失败: {str(e)}")]
async def main():
"""启动MCP Server，使用stdio传输层"""
async with server.run_stdio():
print("文件系统MCP Server已启动，等待Client连接...", flush=True)
await asyncio.Future()  # 保持Server运行
if name == "main":
asyncio.run(main())

代码说明：

• 输入Schema定义：使用Pydantic的BaseModel为每个工具定义输入参数，包含类型注解和描述信息，自动生成JSON Schema。
• 路径安全校验：is_path_safe()函数通过规范化路径并检查是否在允许的基目录范围内，防止路径穿越攻击。
• read_file工具：支持指定编码和最大读取大小，对大文件进行保护，处理编码错误等异常情况。
• write_file工具：支持覆盖写入和追加模式，自动创建不存在的父目录。
• search_files工具：使用fnmatch模块支持通配符模式匹配，可选择是否递归搜索子目录。
• 错误处理：每个工具都包含完善的异常捕获，返回友好的错误提示信息。

六、实战三：集成外部API——天气查询Agent

让MCP Server调用第三方REST API，将外部数据能力注入Agent。

• 在Tool中发起HTTP请求
• 解析API响应并格式化返回
• 处理API限流与超时
• 配置环境变量管理API Key

下面是一个完整的Python MCP Server示例，实现了调用OpenWeatherMap API的天气查询工具：

import asyncio
import os
import json
import time
from typing import Optional
from urllib.parse import urlencode
import aiohttp
from mcp.server import Server, NotificationOptions
from mcp.server.models import InitializationOptions
from mcp.types import Tool, TextContent
from pydantic import BaseModel, Field
---------------------------------------------------------------------------
环境变量管理：API Key 必须通过环境变量注入，禁止硬编码
---------------------------------------------------------------------------
API_KEY_ENV = "OPENWEATHERMAP_API_KEY"
BASE_URL = "https://api.openweathermap.org/data/2.5/weather"
限流配置：同一IP/Key 每分钟最多 60 次（免费套餐限制）
RATE_LIMIT_WINDOW = 60       # 秒
RATE_LIMIT_MAX_CALLS = 60    # 最大调用次数
_request_timestamps: list[float] = []
def _check_rate_limit() -> Optional[str]:
"""检查是否触发限流，若超限则返回错误提示，否则返回 None"""
now = time.time()
# 清理窗口外的旧时间戳
global _request_timestamps
_request_timestamps = [t for t in _request_timestamps if now - t < RATE_LIMIT_WINDOW]
if len(_request_timestamps) >= RATE_LIMIT_MAX_CALLS:
return f"错误：已达到限流上限（{RATE_LIMIT_MAX_CALLS}次/{RATE_LIMIT_WINDOW}秒），请稍后再试"
_request_timestamps.append(now)
return None
---------------------------------------------------------------------------
输入 Schema 定义
---------------------------------------------------------------------------
class GetWeatherInput(BaseModel):
"""天气查询工具的输入参数"""
city: str = Field(description="城市名称，支持中文，如 '北京'、'London'")
country_code: Optional[str] = Field(
default=None,
description="国家代码（ISO 3166-1 alpha-2），如 'CN'、'GB'，可选"
)
units: str = Field(
default="metric",
description="温度单位：metric（摄氏度）、imperial（华氏度）、standard（开尔文）"
)
lang: str = Field(
default="zh_cn",
description="返回信息的语言，如 'zh_cn'、'en'、'ja'"
)
---------------------------------------------------------------------------
初始化 MCP Server
---------------------------------------------------------------------------
server = Server("weather-server")
@server.list_tools()
async def handle_list_tools() -> list[Tool]:
"""注册工具列表"""
return [
Tool(
name="get_weather",
description="查询指定城市的实时天气信息，包括温度、湿度、风速、天气描述等",
inputSchema=GetWeatherInput.model_json_schema(),
)
]
@server.call_tool()
async def handle_call_tool(name: str, arguments: dict) -> list[TextContent]:
"""工具调用分发"""
if name != "get_weather":
raise ValueError(f"未知工具: {name}")
return await handle_get_weather(arguments)
async def handle_get_weather(arguments: dict) -> list[TextContent]:
"""处理天气查询请求"""
# 1. 检查 API Key 是否已配置
api_key = os.environ.get(API_KEY_ENV)
if not api_key:
return [TextContent(
type="text",
text=f"错误：未设置环境变量 {API_KEY_ENV}，请通过 export {API_KEY_ENV}=your_key 配置"
)]
# 2. 限流检查
rate_limit_error = _check_rate_limit()
if rate_limit_error:
    return [TextContent(type="text", text=rate_limit_error)]
3. 解析参数
city = arguments.get("city", "")
country_code = arguments.get("country_code")
units = arguments.get("units", "metric")
lang = arguments.get("lang", "zh_cn")
if not city.strip():
return [TextContent(type="text", text="错误：城市名称不能为空")]
4. 构建查询参数
params = {
"q": f"{city},{country_code}" if country_code else city,
"appid": api_key,
"units": units,
"lang": lang,
}
url = f"{BASE_URL}?{urlencode(params)}"
5. 发起 HTTP 请求（带超时）
timeout = aiohttp.ClientTimeout(total=10)  # 10 秒超时
try:
async with aiohttp.ClientSession(timeout=timeout) as session:
async with session.get(url) as response:
# 5a. 处理 HTTP 错误状态码
if response.status == 401:
return [TextContent(
type="text",
text="错误：API Key 无效，请检查 OPENWEATHERMAP_API_KEY 是否正确"
)]
elif response.status == 404:
return [TextContent(
type="text",
text=f"错误：未找到城市 '{city}'，请检查城市名称或国家代码是否正确"
)]
elif response.status == 429:
return [TextContent(
type="text",
text="错误：请求过于频繁，已达到 API 限流上限，请稍后再试"
)]
elif response.status != 200:
return [TextContent(
type="text",
text=f"错误：API 返回异常状态码 {response.status}，请稍后重试"
)]
        # 5b. 解析 JSON 响应
        data = await response.json()
except asyncio.TimeoutError:
return [TextContent(
type="text",
text="错误：请求超时（10秒），请检查网络连接或稍后重试"
)]
except aiohttp.ClientError as e:
return [TextContent(
type="text",
text=f"错误：网络请求失败 - {str(e)}"
)]
except json.JSONDecodeError:
return [TextContent(
type="text",
text="错误：API 返回数据格式异常，无法解析"
)]
6. 提取并格式化返回数据
try:
weather_desc = data["weather"][0]["description"]
temp = data["main"]["temp"]
feels_like = data["main"]["feels_like"]
humidity = data["main"]["humidity"]
pressure = data["main"]["pressure"]
wind_speed = data["wind"]["speed"]
wind_deg = data.get("wind", {}).get("deg", 0)
visibility = data.get("visibility", 0)
clouds = data["clouds"]["all"]
city_name = data["name"]
country = data["sys"]["country"]
# 温度单位符号
unit_symbol = "°C" if units == "metric" else "°F" if units == "imperial" else "K"
# 风向文字描述
wind_directions = ["北", "东北", "东", "东南", "南", "西南", "西", "西北"]
wind_dir = wind_directions[round(wind_deg / 45) % 8]
formatted = (
f"🌍 {city_name}, {country}\n"
f"━━━━━━━━━━━━━━━━━━━━\n"
f"🌡️ 温度：{temp}{unit_symbol}（体感 {feels_like}{unit_symbol}）\n"
f"☁️ 天气：{weather_desc}\n"
f"💧 湿度：{humidity}%\n"
f"📊 气压：{pressure} hPa\n"
f"🌬️ 风速：{wind_speed} m/s（{wind_dir}风）\n"
f"👁️ 能见度：{visibility} 米\n"
f"☁️ 云量：{clouds}%\n"
f"━━━━━━━━━━━━━━━━━━━━\n"
f"数据来源：OpenWeatherMap"
)
return [TextContent(type="text", text=formatted)]
except (KeyError, IndexError, TypeError) as e:
return [TextContent(
type="text",
text=f"错误：解析天气数据时出错 - {str(e)}"
)]
async def main():
"""启动 MCP Server"""
async with server.run_stdio():
print("天气查询 MCP Server 已启动，等待 Client 连接...", flush=True)
await asyncio.Future()
if name == "main":
asyncio.run(main())

代码说明：

• 环境变量管理：API Key 通过 OPENWEATHERMAP_API_KEY 环境变量注入，启动前需执行 export OPENWEATHERMAP_API_KEY=your_key，代码中通过 os.environ.get() 读取并给出明确的未配置提示。
• HTTP 请求：使用 aiohttp 异步库发起 GET 请求，通过 urlencode 构建查询参数，支持城市名称、国家代码、温度单位和语言设置。
• 响应解析与格式化：从 JSON 响应中提取温度、体感温度、湿度、气压、风速、风向、能见度、云量等信息，使用 Emoji 和分隔线进行美观格式化输出。
• 错误处理：覆盖多种异常场景——API Key 无效（401）、城市不存在（404）、限流（429）、网络超时（asyncio.TimeoutError，10秒超时）、网络异常（aiohttp.ClientError）、JSON 解析异常（json.JSONDecodeError）、数据字段缺失（KeyError）等，每种错误都返回友好的中文提示。
• 限流保护：内置基于时间窗口的限流器 _check_rate_limit()，记录最近 60 秒内的请求时间戳，超过 60 次则返回限流提示，避免触发 OpenWeatherMap 免费套餐的 API 限制。
• 输入 Schema：使用 Pydantic 的 BaseModel 定义 GetWeatherInput，包含城市名称、可选国家代码、温度单位和语言参数，自动生成 JSON Schema 供 Client 校验。

七、Client端集成：让AI模型调用你的工具

编写MCP Client，连接Server并让LLM自动调用已注册工具。

• 初始化Client并连接Server
• 获取可用工具列表
• 构造工具调用请求并处理响应
• 与主流LLM（如Claude、GPT）的集成示例

下面是一个完整的Python MCP Client示例，连接前面创建的calculator-server，获取工具列表并调用add工具进行计算：

import asyncio
import json
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
---------------------------------------------------------------------------
1. 定义Server连接参数（使用stdio传输层）
---------------------------------------------------------------------------
指定要启动的MCP Server命令和参数
server_params = StdioServerParameters(
command="python",
args=["calculator_server.py"],  # 假设calculator-server代码保存在此文件中
)
async def main():
"""完整的MCP Client示例：连接Server、发现工具、调用工具"""
# -----------------------------------------------------------------------
# 2. 建立连接：启动Server子进程并创建Client会话
# -----------------------------------------------------------------------
print("正在连接MCP Server...", flush=True)
async with stdio_client(server_params) as (read, write):
    async with ClientSession(read, write) as session:
        # 初始化会话（必须调用，完成协议握手）
        await session.initialize()
        print("MCP Client已连接，协议版本:", session.server_initialization_result.protocol_version, flush=True)
    # -------------------------------------------------------------------
    # 3. 工具发现：获取Server注册的所有工具列表
    # -------------------------------------------------------------------
    print("\n--- 获取工具列表 ---", flush=True)
    tools_result = await session.list_tools()
    tools = tools_result.tools

    print(f"发现 {len(tools)} 个工具:", flush=True)
    for tool in tools:
        print(f"  - {tool.name}: {tool.description}", flush=True)
        print(f"    输入Schema: {json.dumps(tool.inputSchema, indent=4, ensure_ascii=False)}", flush=True)

    # -------------------------------------------------------------------
    # 4. 构造参数并调用 add 工具
    # -------------------------------------------------------------------
    print("\n--- 调用 add 工具 ---", flush=True)

    # 准备调用参数：计算 3.14 + 2.86
    tool_name = "add"
    arguments = {
        "a": 3.14,
        "b": 2.86,
    }

    print(f"调用工具: {tool_name}", flush=True)
    print(f"参数: a={arguments['a']}, b={arguments['b']}", flush=True)

    # 发起工具调用
    result = await session.call_tool(tool_name, arguments)

    # -------------------------------------------------------------------
    # 5. 结果解析：处理返回的 TextContent
    # -------------------------------------------------------------------
    print(f"\n调用结果:", flush=True)
    for content in result.content:
        if content.type == "text":
            print(f"  {content.text}", flush=True)

    # 将结果转换为数值并验证
    try:
        result_value = float(result.content[0].text)
        expected = arguments["a"] + arguments["b"]
        print(f"\n验证: {arguments['a']} + {arguments['b']} = {result_value}", flush=True)
        print(f"预期结果: {expected}", flush=True)
        print(f"匹配: {'✅ 正确' if abs(result_value - expected) < 1e-9 else '❌ 错误'}", flush=True)
    except (ValueError, IndexError) as e:
        print(f"结果解析失败: {e}", flush=True)

    # -------------------------------------------------------------------
    # 6. 错误处理示例：调用不存在的工具
    # -------------------------------------------------------------------
    print("\n--- 错误处理示例：调用不存在的工具 ---", flush=True)
    try:
        await session.call_tool("nonexistent_tool", {})
    except Exception as e:
        print(f"预期错误: {e}", flush=True)
print("\nMCP Client已断开连接。", flush=True)
if name == "main":
asyncio.run(main())

代码说明：

• 连接建立：使用StdioServerParameters指定启动Server的命令和参数，通过stdio_client()上下文管理器启动Server子进程并建立双向通信管道，再通过ClientSession完成协议握手。
• 工具发现：调用session.list_tools()获取Server注册的所有工具列表，每个工具包含name、description和inputSchema（JSON Schema格式），Client可据此动态生成调用界面或LLM工具描述。
• 参数构造与调用：根据add工具的输入Schema构造arguments字典（包含a和b两个浮点数），通过session.call_tool()发起调用，返回CallToolResult对象。
• 结果解析：从result.content中提取TextContent类型的响应文本，将其转换为数值并与预期结果进行比对验证。
• 错误处理：演示了调用不存在的工具时Server返回的异常，Client应捕获Exception并给出友好提示。
• 资源管理：使用async with上下文管理器确保Server子进程和会话在退出时自动清理，避免资源泄漏。

八、进阶：多Server编排与工具路由

当Agent需要同时使用多个MCP Server时，如何统一管理与路由。

• 多Server连接与生命周期管理
• 工具名称冲突的解决策略
• 基于语义的工具路由设计思路
• 示例：文件系统Server + 数据库Server + 搜索Server

九、生产化部署与最佳实践

将MCP工具链从开发环境推向生产环境的关键考量。

• Server的进程管理与守护
• 日志与监控
• 安全加固：输入校验、权限控制、沙箱隔离
• 性能优化：连接复用与缓存策略