MCP
July 31, 2025 · View on GitHub
Model Context Protocol (MCP) 是一个开放协议，旨在实现 LLM 应用与外部数据源和工具之间的无缝集成。
无论您是构建 AI 驱动的 IDE、增强聊天界面，还是创建自定义 AI 工作流，MCP 都提供了一种标准化的方式来连接 LLM 与外部世界。
简单来说，MCP 是一种客户端-服务器架构的协议，允许 LLM 应用程序（如 Claude、各种 IDE 等）通过标准化的接口访问外部数据和功能。这解决了 LLM 在实际应用中常见的一些痛点：
- LLM 无法直接访问实时数据（如天气、股票行情等）
- LLM 无法执行外部操作（如发送邮件、控制设备等）
- LLM 无法访问用户的本地文件或其他私有数据
通过 MCP，这些限制得到了优雅的解决，同时保持了安全性和可扩展性。
MCP的核心架构
MCP 采用客户端-服务器架构，主要包含以下几个组件：
- **MCP 主机（Host）：**如 Claude Desktop、IDE 或其他 AI 工具，通过 MCP 访问数据
- **MCP 客户端（Client）：**与服务器保持 1:1 连接的协议客户端
- **MCP 服务器（Server）：**轻量级程序，通过标准化的 MCP 协议公开特定功能
- **本地数据源：**计算机上的文件、数据库和服务，MCP 服务器可以安全访问这些内容
- **远程服务：**通过互联网可用的外部系统（例如通过 API），MCP 服务器可以连接这些服务
主机可以同时连接多个服务器，每个服务器提供不同的功能，形成一个生态系统：
主机（Claude、IDE 等）<--MCP 协议--> 服务器 A <--> 本地数据源 A
<--MCP 协议--> 服务器 B <--> 本地数据源 B
<--MCP 协议--> 服务器 C <--> 远程服务 C
MCP的核心概念
MCP 服务器可以提供三种主要类型的功能：
- **资源（Resources）：**客户端可以读取的文件类数据（如 API 响应或文件内容）
- **工具（Tools）：**LLM 可以调用的函数（需要用户批准）
- **提示（Prompts）：**帮助用户完成特定任务的预写模板
资源（Resources）
资源是可以被客户端读取的文件类数据。它们可以是文本或二进制形式，并有唯一的 URI 标识。
资源可以是：
- 直接资源：固定内容的资源
- 资源模板：可以通过参数动态生成的资源
例如，一个文件系统 MCP 服务器可以将本地文件作为资源提供给 LLM，使其能够读取用户的文件。
工具（Tools）
工具是 MCP 中最强大的原语之一，允许服务器向客户端公开可执行的功能。通过工具，LLM 可以与外部系统交互，执行计算，并在现实世界中采取行动。
每个工具都有明确的定义，包括：
- 名称
- 描述
- 输入参数模式（使用 JSON Schema）
- 输出格式
工具设计为由模型控制，但通常需要人类批准才能执行，这保证了安全性。
提示（Prompts）
提示是预定义的模板，可以帮助用户完成特定任务。它们可以包含动态部分，嵌入资源上下文，并支持多步工作流。
MCP 服务器开发案例：天气服务器
让我们通过一个实际例子来理解 MCP 服务器的开发。我们将构建一个简单的天气服务器，它提供两个工具：获取天气警报和获取天气预报。
步骤 1：设置环境
# 创建项目目录
uv init weather
cd weather
# 创建虚拟环境并激活
uv venv
source .venv/bin/activate # MacOS/Linux
# 或 .venv\Scripts\activate # Windows
# 安装依赖
uv add "mcp[cli]" httpx
# 创建服务器文件
touch weather.py
步骤 2：实现天气服务器
from typing import Any
import httpx
from mcp.server.fastmcp import FastMCP
# 初始化 FastMCP 服务器
mcp = FastMCP("weather")
# 常量
NWS_API_BASE = "https://api.weather.gov"
USER_AGENT = "weather-app/1.0"
async def make_nws_request(url: str) -> dict[str, Any] | None:
"""向 NWS API 发送请求并进行适当的错误处理。"""
headers = {
"User-Agent": USER_AGENT,
"Accept": "application/geo+json"
}
async with httpx.AsyncClient() as client:
try:
response = await client.get(url, headers=headers, timeout=30.0)
response.raise_for_status()
return response.json()
except Exception:
return None
def format_alert(feature: dict) -> str:
"""将警报特征格式化为可读字符串。"""
props = feature["properties"]
return f"""Event: {props.get('event', 'Unknown')}
Area: {props.get('areaDesc', 'Unknown')}
Severity: {props.get('severity', 'Unknown')}
Description: {props.get('description', 'No description available')}
Instructions: {props.get('instruction', 'No specific instructions provided')}"""
@mcp.tool()
async def get_alerts(state: str) -> str:
"""获取美国州的天气警报。
Args:
state: 美国州的两字母代码（例如 CA, NY）
"""
url = f"{NWS_API_BASE}/alerts/active/area/{state}"
data = await make_nws_request(url)
if not data or "features" not in data:
return "无法获取警报或未找到警报。"
if not data["features"]:
return "该州没有活跃警报。"
alerts = [format_alert(feature) for feature in data["features"]]
return "\n---\n".join(alerts)
@mcp.tool()
async def get_forecast(latitude: float, longitude: float) -> str:
"""获取某个位置的天气预报。
Args:
latitude: 位置的纬度
longitude: 位置的经度
"""
# 首先获取预报网格端点
points_url = f"{NWS_API_BASE}/points/{latitude},{longitude}"
points_data = await make_nws_request(points_url)
if not points_data:
return "无法获取该位置的预报数据。"
# 从 points 响应获取预报 URL
forecast_url = points_data["properties"]["forecast"]
forecast_data = await make_nws_request(forecast_url)
if not forecast_data:
return "无法获取详细预报。"
# 将周期格式化为可读的预报
periods = forecast_data["properties"]["periods"]
forecasts = []
for period in periods[:5]: # 仅显示接下来的 5 个周期
forecast = f"""{period['name']}:
Temperature: {period['temperature']}°{period['temperatureUnit']}
Wind: {period['windSpeed']} {period['windDirection']}
Forecast: {period['detailedForecast']}"""
forecasts.append(forecast)
return "\n---\n".join(forecasts)
if __name__ == "__main__":
# 初始化并运行服务器
mcp.run(transport='stdio')
步骤 3：配置 Claude Desktop 连接服务器
要使用 Claude Desktop 连接到我们的服务器，需要编辑配置文件：
{
"mcpServers": {
"weather": {
"command": "uv",
"args": ["--directory", "/ABSOLUTE/PATH/TO/PARENT/FOLDER/weather", "run", "weather.py"]
}
}
}
此配置告诉 Claude Desktop 如何启动我们的天气服务器，并使 Claude 能够使用我们实现的工具。
MCP 客户端开发案例
接下来，让我们看看如何开发一个 MCP 客户端，该客户端可以连接到任何 MCP 服务器并利用其功能。
步骤 1：设置环境
# 创建项目目录
uv init mcp-client
cd mcp-client
# 创建虚拟环境
uv venv
source .venv/bin/activate # MacOS/Linux
# 或 .venv\Scripts\activate # Windows
# 安装依赖
uv add mcp anthropic python-dotenv
# 创建主文件
touch client.py
步骤 2：设置 API 密钥
创建 .env 文件存储 Anthropic API 密钥：
ANTHROPIC_API_KEY=<your key here>
步骤 3：实现 MCP 客户端
import asyncio
from typing import Optional
from contextlib import AsyncExitStack
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from anthropic import Anthropic
from dotenv import load_dotenv
load_dotenv() # 从 .env 加载环境变量
class MCPClient:
def __init__(self):
# 初始化会话和客户端对象
self.session: Optional[ClientSession] = None
self.exit_stack = AsyncExitStack()
self.anthropic = Anthropic()
async def connect_to_server(self, server_script_path: str):
"""连接到 MCP 服务器
Args:
server_script_path: 服务器脚本路径 (.py 或 .js)
"""
is_python = server_script_path.endswith('.py')
is_js = server_script_path.endswith('.js')
if not (is_python or is_js):
raise ValueError("服务器脚本必须是 .py 或 .js 文件")
command = "python" if is_python else "node"
server_params = StdioServerParameters(
command=command,
args=[server_script_path],
env=None
)
stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))
self.stdio, self.write = stdio_transport
self.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))
await self.session.initialize()
# 列出可用工具
response = await self.session.list_tools()
tools = response.tools
print("\n连接到服务器，可用工具:", [tool.name for tool in tools])
async def process_query(self, query: str) -> str:
"""使用 Claude 和可用工具处理查询"""
if not self.session:
return "未连接到任何服务器"
messages = [
{
"role": "user",
"content": query
}
]
response = await self.session.list_tools()
available_tools = [{
"name": tool.name,
"description": tool.description,
"input_schema": tool.inputSchema
} for tool in response.tools]
# 初始 Claude API 调用
claude_response = await self.anthropic.messages.create(
model="claude-3-opus-20240229",
max_tokens=1000,
messages=messages,
tools=available_tools
)
# 处理工具调用
message = claude_response.content[0]
if hasattr(message, 'tool_calls') and message.tool_calls:
print("\nClaude 请求使用工具:", message.tool_calls[0].name)
# 执行工具调用
tool_call = message.tool_calls[0]
tool_name = tool_call.name
tool_params = tool_call.params
print(f"使用参数执行 {tool_name}:", tool_params)
tool_response = await self.session.execute_tool(tool_name, tool_params)
# 将工具结果发送回 Claude
messages.append({
"role": "assistant",
"content": [message]
})
messages.append({
"role": "user",
"content": [
{
"type": "tool_result",
"tool_call_id": tool_call.id,
"content": tool_response
}
]
})
# 获取最终回答
final_response = await self.anthropic.messages.create(
model="claude-3-opus-20240229",
max_tokens=1000,
messages=messages
)
return final_response.content[0].text
else:
# 没有工具调用时直接返回回答
return message.text
async def close(self):
"""关闭连接和资源"""
await self.exit_stack.aclose()
async def main():
# 创建客户端实例
client = MCPClient()
try:
# 连接到天气服务器
print("连接到天气服务器...")
await client.connect_to_server("../weather/weather.py")
# 交互式循环
while True:
query = input("\n输入查询 (输入 'exit' 退出): ")
if query.lower() == 'exit':
break
print("\n处理查询...")
response = await client.process_query(query)
print("\n回答:", response)
finally:
# 关闭连接
await client.close()
if __name__ == "__main__":
asyncio.run(main())
这个客户端可以连接到任何 MCP 服务器，获取其可用工具，然后将用户查询发送到 Claude 进行处理。Claude 可以调用服务器提供的工具，客户端将结果返回给 Claude 以生成最终回答。