Model Context Protocol（MCP）是一种开放标准，旨在标准化大型语言模型（LLMs）与外部上下文和工具的交互方式。在本文中，我们将深入探讨如何使用Python构建MCP服务器和客户端，并比较官方的python-sdk与fastmcp这两个项目的区别与联系。
MCP的核心概念
在开始实际编码之前，让我们先了解MCP的三个核心概念：
- 工具（Tools）：允许LLM执行操作，例如计算、API调用或数据处理。
- 资源（Resources）：为LLM提供只读数据，如文件内容、数据库记录等。
- 提示（Prompts）：帮助LLM更有效地与服务器交互的模板。
这三个概念共同构成了MCP的基础功能，使AI模型能够安全地访问外部工具和数据。
python-sdk vs fastmcp
在实现MCP服务器之前，我们需要了解两个主要的Python实现：
python-sdk
python-sdk是Model Context Protocol的官方Python实现。作为官方实现，它提供了完整而灵活的API，支持MCP协议的所有功能。
特点：
- 完整实现MCP规范
- 提供低级和高级API
- 支持多种传输方式（SSE和STDIO）
- 灵活性高，适合需要精细控制的场景
- 持续维护和更新
fastmcp
fastmcp是由Jerad Lowin创建的更高级别的封装库，目标是提供更Pythonic、更简洁的MCP服务器开发体验。
特点：
- 简洁的装饰器语法
- 自动类型处理
- 集成的CLI工具
- 快速开发体验
值得注意的是，fastmcp已经被整合到官方python-sdk中，原仓库不再单独维护。所以现在推荐直接使用官方python-sdk。
代码对比
让我们通过代码示例来比较这两种方式：
使用python-sdk低级API：
from mcp.server.lowlevel import Server
from mcp.types import Tool, TextContent
# 创建MCP服务器
mcp_server = Server("MCPDemoServer")
# 定义工具
ADD_TOOL = Tool(
name="add",
description="Add two numbers",
inputSchema={
"type": "object",
"properties": {
"a": {"type": "integer", "description": "First number"},
"b": {"type": "integer", "description": "Second number"}
},
"required": ["a", "b"]
}
)
@mcp_server.call_tool()
async def call_tool(name: str, arguments: dict):
if name == "add":
a = arguments.get("a", 0)
b = arguments.get("b", 0)
result = a + b
return [TextContent(type="text", text=str(result))]
else:
raise ValueError(f"Unknown tool: {name}")
使用fastmcp高级API：
from fastmcp import FastMCP
mcp = FastMCP("Demo 🚀")
@mcp.tool()
def add(a: int, b: int) -> int:
"""Add two numbers"""
return a + b
可以看到，fastmcp大大简化了代码，使开发者能够专注于业务逻辑而非协议细节。
实现MCP服务器
现在，让我们看看如何实现一个完整的MCP服务器，支持SSE和STDIO两种传输方式。
服务器核心组件
我们的服务器需要以下组件：
- 工具定义和处理
- 资源定义和处理
- SSE传输支持
- STDIO传输支持
1. 工具定义和处理
# 定义工具
ADD_TOOL = Tool(
name="add",
description="Add two numbers",
inputSchema={
"type": "object",
"properties": {
"a": {"type": "integer", "description": "First number"},
"b": {"type": "integer", "description": "Second number"}
},
"required": ["a", "b"]
}
)
AVAILABLE_TOOLS = [ADD_TOOL]
@mcp_server.call_tool()
async def call_tool(name: str, arguments: dict):
logger.info(f"[mcp_server.call_tool] name={name}, arguments={arguments}")
if name == "add":
a = arguments.get("a", 0)
b = arguments.get("b", 0)
result = a + b
return [TextContent(type="text", text=str(result))]
else:
raise ValueError(f"Unknown tool: {name}")
@mcp_server.list_tools()
async def list_tools():
logger.info("[mcp_server.list_tools] called")
return AVAILABLE_TOOLS
2. 资源定义和处理
# 定义资源
GREETING_RESOURCE = {
"uri_pattern": "greeting://{name}",
"name": "greeting",
"description": "Get a personalized greeting"
}
AVAILABLE_RESOURCES = [GREETING_RESOURCE]
@mcp_server.list_resources()
async def list_resources():
logger.info("[mcp_server.list_resources] called")
return AVAILABLE_RESOURCES
@mcp_server.read_resource()
async def read_resource(uri: str):
logger.info(f"[mcp_server.read_resource] uri={uri}")
if uri.startswith("greeting://"):
name = uri.replace("greeting://", "")
return f"Hello, {name}!"
else:
raise ValueError(f"Unknown resource URI: {uri}")
3. SSE传输支持
SSE（Server-Sent Events）允许服务器向客户端推送事件，适合Web应用的实时通信。
# SSE端点
async def sse_endpoint(request: Request):
logger.info("[SSE] => sse_endpoint called")
# 创建SSE传输
sse_transport = SseServerTransport("/sse/messages")
scope = request.scope
receive = request.receive
send = request._send
async with sse_transport.connect_sse(scope, receive, send) as (read_st, write_st):
logger.info("[SSE] run mcp_server with SSE transport => calling 'mcp_server.run'")
init_opts = mcp_server.create_initialization_options()
# 运行服务器
await mcp_server.run(read_st, write_st, init_opts)
logger.info("[SSE] => SSE session ended")
return
我们使用Starlette框架提供HTTP端点，并使用 SseServerTransport处理SSE连接。
4. STDIO传输支持
STDIO传输使用标准输入/输出流进行通信，适合命令行应用和本地集成。
# 运行Stdio服务器
def run_stdio_server():
"""运行标准输入输出服务器"""
import sys
import asyncio
logger.info("Starting MCP server with stdio transport")
# 创建初始化选项
init_opts = mcp_server.create_initialization_options()
# 使用低级API运行服务器
async def run():
try:
# 创建传输
from mcp.server.stdio import StdioServerTransport
transport = StdioServerTransport(sys.stdin.buffer, sys.stdout.buffer)
# 使用连接上下文管理器
async with transport.connect() as (read_stream, write_stream):
await mcp_server.run(read_stream, write_stream, init_opts)
except Exception as e:
logger.error(f"STDIO server error: {e}")
import traceback
logger.error(traceback.format_exc())
sys.exit(1)
# 运行异步函数
asyncio.run(run())
完整服务器
将这些组件组合起来，我们得到一个完整的MCP服务器，可以根据命令行参数选择使用SSE或STDIO传输：
def main():
"""主函数，根据参数运行不同类型的服务器"""
import argparse
parser = argparse.ArgumentParser(description="MCP Demo Server")
parser.add_argument("--transport", choices=["sse", "stdio"], default="sse",
help="Transport type (sse or stdio)")
args = parser.parse_args()
if args.transport == "stdio":
run_stdio_server()
else: # 默认使用SSE
uvicorn.run(app, host="0.0.0.0", port=8000)
if __name__ == "__main__":
main()
实现MCP客户端
在服务器实现之后，我们需要一个客户端来与服务器交互。客户端也需要支持SSE和STDIO两种传输方式。
客户端核心组件
- 传输方式选择
- 客户端会话管理
- 工具调用和资源访问
1. 传输方式选择
async def run_client(transport_type: str, server_command: Optional[str] = None, server_url: Optional[str] = None):
"""运行MCP客户端"""
logger.info(f"Starting MCP client with {transport_type} transport")
if transport_type == "stdio":
if not server_command:
server_path = os.path.join(SCRIPT_DIR, "server.py")
server_command = f"{sys.executable} {server_path} --transport stdio"
# 创建服务器参数
cmd_parts = server_command.split()
server_params = StdioServerParameters(
command=cmd_parts[0],
args=cmd_parts[1:] if len(cmd_parts) > 1 else []
)
# 连接到服务器
async with stdio_client(server_params) as (read, write):
# 创建客户端会话
async with ClientSession(read, write) as session:
# 初始化连接
await session.initialize()
# 调用工具和读取资源
await demo_client_operations(session)
elif transport_type == "sse":
if not server_url:
server_url = "http://localhost:8000/sse"
# 连接到服务器
async with sse_client(server_url) as (read, write):
# 创建客户端会话
async with ClientSession(read, write) as session:
# 初始化连接
await session.initialize()
# 调用工具和读取资源
await demo_client_operations(session)
2. 工具调用和资源访问
async def demo_client_operations(session: ClientSession):
"""执行演示操作"""
try:
# 列出可用工具
logger.info("Listing available tools...")
tools = await session.list_tools()
logger.info(f"Available tools: {tools}")
# 调用add工具
logger.info("Calling 'add' tool...")
result = await session.call_tool("add", arguments={"a": 5, "b": 3})
logger.info(f"Result of add(5, 3): {result}")
# 列出可用资源
logger.info("Listing available resources...")
resources = await session.list_resources()
logger.info(f"Available resources: {resources}")
# 读取greeting资源
logger.info("Reading 'greeting' resource...")
try:
uri = "greeting://World"
content, mime_type = await session.read_resource(uri)
logger.info(f"Greeting content: {content}, mime type: {mime_type}")
except Exception as e:
logger.error(f"Error reading resource: {e}")
except Exception as e:
logger.error(f"Error during client operations: {e}")
测试MCP实现
为了确保我们的实现正常工作，我们需要编写测试代码来验证服务器和客户端的功能。
def test_stdio():
"""测试标准输入/输出连接"""
logger.info("=== 测试 STDIO 连接 ===")
# 运行客户端（客户端会自动启动服务器）
return_code = run_client(transport="stdio")
if return_code == 0:
logger.info("STDIO 测试成功!")
else:
logger.error(f"STDIO 测试失败，返回码: {return_code}")
return return_code
def test_sse():
"""测试 SSE 连接"""
logger.info("=== 测试 SSE 连接 ===")
# 启动服务器
server_proc = run_server(transport="sse")
try:
# 等待服务器启动
logger.info("等待服务器启动...")
time.sleep(2)
# 运行客户端
return_code = run_client(transport="sse", server_url="http://localhost:8000/sse")
if return_code == 0:
logger.info("SSE 测试成功!")
else:
logger.error(f"SSE 测试失败，返回码: {return_code}")
return return_code
finally:
# 停止服务器
logger.info("停止服务器...")
server_proc.terminate()
server_proc.wait()
实际应用与挑战
在实现MCP服务器和客户端的过程中，我们遇到了一些挑战：
- 类型兼容性问题：MCP API需要特定的类型（如AnyUrl），但在不同版本中可能有差异。
- 错误处理：需要处理各种异常情况，如连接失败、协议错误等。
- 调试复杂性：由于涉及到网络通信和异步编程，调试可能变得复杂。
选择哪个框架？
- 如果你是MCP新手：建议使用官方python-sdk中的高级API（原fastmcp）。
- 如果你需要最大灵活性：使用python-sdk的低级API。
- 如果你正在构建生产系统：使用官方python-sdk，因为它有持续的维护和更新。
总结
Model Context Protocol为大型语言模型提供了一种标准化的方式来访问外部工具和数据。通过官方python-sdk或其高级封装，我们可以轻松构建MCP服务器和客户端。
fastmcp作为一个高层封装，极大地简化了MCP服务器的开发，但现在已经被整合到官方python-sdk中，所以推荐直接使用官方SDK。
无论选择哪种方式，MCP都为AI应用开发提供了强大的工具，使LLM能够安全、标准化地与外部世界交互。
评论区