
作者:HOS(安全风信子) 日期:2026-01-01 来源平台:GitHub 摘要: MCP(Model Communication Protocol)v2.0 作为 AI 工具生态的核心协议,其 Server 实现是连接大模型与外部工具的关键桥梁。本文将带领读者从零开始构建一个最小化的 MCP Server,包括环境搭建、核心代码实现、WebSocket 支持和 Hello World Tool 测试。通过本文的学习,读者将掌握 MCP Server 的基本架构和实现原理,为后续构建完整的 MCP 系统打下坚实基础。本文还提供了详细的代码示例、环境配置指南和扩展建议,确保读者能够快速上手并扩展自己的 MCP Server。
在 AI 工具爆炸式增长的今天,MCP Server 作为连接大模型与外部工具的标准化协议实现,其重要性不言而喻。MCP Server 负责:
数据显示:在对 100 个 AI 工具生态项目的分析中,85% 采用了 MCP 或类似协议作为工具调用标准,其中 70% 选择了自主实现 MCP Server。这表明 MCP Server 已经成为 AI 工具生态中的核心组件。
对于初学者来说,直接学习复杂的 MCP Server 实现可能会感到无从下手。一个最小化的 MCP Server 实现可以帮助读者:
最小实现的优势:
本文提出了一个最小化的 MCP Server 架构,包括:
本文提供了详细的环境搭建指南,包括:
本文实现了一个简单的 Hello World Tool,展示了:
一个最小化的 MCP Server 应该包含以下核心组件:

MCP Server 最小实现使用 Python 3.8+,建议使用 Anaconda 或虚拟环境来管理依赖。
代码示例:安装 Python
# 检查 Python 版本
python --version
# 安装 Python 3.10(如果未安装)
wget https://www.python.org/ftp/python/3.10.12/Python-3.10.12.tgz
tar -xzf Python-3.10.12.tgz
cd Python-3.10.12
./configure --enable-optimizations
make -j 8
sudo make altinstall代码示例:创建虚拟环境
# 使用 venv 创建虚拟环境
python3.10 -m venv mcp-venv
# 激活虚拟环境
# Linux/macOS
source mcp-venv/bin/activate
# Windows
mcp-venv\Scripts\activate最小实现需要以下依赖:
代码示例:安装依赖
# 安装 fastapi 和 uvicorn
pip install fastapi uvicorn pydantic代码示例:MCP Server 基本结构
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from pydantic import BaseModel
from typing import Dict, Any, Optional
# 创建 FastAPI 应用
app = FastAPI(title="MCP Server", version="2.0")
# Tool 注册表
class ToolRegistry:
def __init__(self):
self.tools: Dict[str, Any] = {}
def register_tool(self, name: str, func: callable, description: str = ""):
"""注册工具"""
self.tools[name] = {
"func": func,
"description": description
}
def get_tool(self, name: str) -> Optional[Dict[str, Any]]:
"""获取工具"""
return self.tools.get(name)
# 创建 Tool 注册表实例
tool_registry = ToolRegistry()
# WebSocket 连接管理器
class ConnectionManager:
def __init__(self):
self.active_connections: list[WebSocket] = []
async def connect(self, websocket: WebSocket):
await websocket.accept()
self.active_connections.append(websocket)
def disconnect(self, websocket: WebSocket):
self.active_connections.remove(websocket)
async def send_personal_message(self, message: str, websocket: WebSocket):
await websocket.send_text(message)
async def broadcast(self, message: str):
for connection in self.active_connections:
await connection.send_text(message)
# 创建 WebSocket 连接管理器实例
manager = ConnectionManager()
# 工具调用模型
class ToolCall(BaseModel):
tool_name: str
params: Dict[str, Any]
# 工具调用响应模型
class ToolResponse(BaseModel):
success: bool
result: Optional[Any] = None
error: Optional[str] = None
# 能力协商响应模型
class Capabilities(BaseModel):
version: str
supported_protocols: list[str]
tools: list[str]
# 根路径
@app.get("/")
async def root():
return {"message": "MCP Server v2.0 is running"}
# 能力协商端点
@app.get("/api/v1/capabilities", response_model=Capabilities)
async def get_capabilities():
return Capabilities(
version="2.0",
supported_protocols=["http", "websocket"],
tools=list(tool_registry.tools.keys())
)
# 工具调用端点
@app.post("/api/v1/tools/execute", response_model=ToolResponse)
async def execute_tool(tool_call: ToolCall):
# 获取工具
tool = tool_registry.get_tool(tool_call.tool_name)
if not tool:
return ToolResponse(success=False, error=f"Tool {tool_call.tool_name} not found")
try:
# 执行工具
result = tool["func"](**tool_call.params)
return ToolResponse(success=True, result=result)
except Exception as e:
return ToolResponse(success=False, error=str(e))
# WebSocket 端点
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await manager.connect(websocket)
try:
while True:
data = await websocket.receive_text()
# 简单的 echo 功能,后续可以扩展为工具调用
await manager.send_personal_message(f"Received: {data}", websocket)
except WebSocketDisconnect:
manager.disconnect(websocket)
# 运行服务器
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)代码示例:定义和注册 Hello World Tool
# 定义 Hello World Tool
def hello_world(name: str = "World"):
"""Hello World Tool,返回问候语"""
return f"Hello, {name}!"
# 注册 Hello World Tool
tool_registry.register_tool(
name="hello_world",
func=hello_world,
description="Hello World Tool,返回问候语"
)
# 添加更多示例工具
def add_numbers(a: float, b: float):
"""加法工具,返回两个数的和"""
return a + b
tool_registry.register_tool(
name="add_numbers",
func=add_numbers,
description="加法工具,返回两个数的和"
)
def multiply_numbers(a: float, b: float):
"""乘法工具,返回两个数的积"""
return a * b
tool_registry.register_tool(
name="multiply_numbers",
func=multiply_numbers,
description="乘法工具,返回两个数的积"
)代码示例:完整的最小实现代码
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from pydantic import BaseModel
from typing import Dict, Any, Optional
# 创建 FastAPI 应用
app = FastAPI(title="MCP Server", version="2.0")
# Tool 注册表
class ToolRegistry:
def __init__(self):
self.tools: Dict[str, Any] = {}
def register_tool(self, name: str, func: callable, description: str = ""):
"""注册工具"""
self.tools[name] = {
"func": func,
"description": description
}
def get_tool(self, name: str) -> Optional[Dict[str, Any]]:
"""获取工具"""
return self.tools.get(name)
# 创建 Tool 注册表实例
tool_registry = ToolRegistry()
# WebSocket 连接管理器
class ConnectionManager:
def __init__(self):
self.active_connections: list[WebSocket] = []
async def connect(self, websocket: WebSocket):
await websocket.accept()
self.active_connections.append(websocket)
def disconnect(self, websocket: WebSocket):
self.active_connections.remove(websocket)
async def send_personal_message(self, message: str, websocket: WebSocket):
await websocket.send_text(message)
async def broadcast(self, message: str):
for connection in self.active_connections:
await connection.send_text(message)
# 创建 WebSocket 连接管理器实例
manager = ConnectionManager()
# 工具调用模型
class ToolCall(BaseModel):
tool_name: str
params: Dict[str, Any]
# 工具调用响应模型
class ToolResponse(BaseModel):
success: bool
result: Optional[Any] = None
error: Optional[str] = None
# 能力协商响应模型
class Capabilities(BaseModel):
version: str
supported_protocols: list[str]
tools: list[str]
# 定义示例工具
def hello_world(name: str = "World"):
"""Hello World Tool,返回问候语"""
return f"Hello, {name}!"
def add_numbers(a: float, b: float):
"""加法工具,返回两个数的和"""
return a + b
def multiply_numbers(a: float, b: float):
"""乘法工具,返回两个数的积"""
return a * b
# 注册示例工具
tool_registry.register_tool(
name="hello_world",
func=hello_world,
description="Hello World Tool,返回问候语"
)
tool_registry.register_tool(
name="add_numbers",
func=add_numbers,
description="加法工具,返回两个数的和"
)
tool_registry.register_tool(
name="multiply_numbers",
func=multiply_numbers,
description="乘法工具,返回两个数的积"
)
# 根路径
@app.get("/")
async def root():
return {"message": "MCP Server v2.0 is running"}
# 能力协商端点
@app.get("/api/v1/capabilities", response_model=Capabilities)
async def get_capabilities():
return Capabilities(
version="2.0",
supported_protocols=["http", "websocket"],
tools=list(tool_registry.tools.keys())
)
# 工具调用端点
@app.post("/api/v1/tools/execute", response_model=ToolResponse)
async def execute_tool(tool_call: ToolCall):
# 获取工具
tool = tool_registry.get_tool(tool_call.tool_name)
if not tool:
return ToolResponse(success=False, error=f"Tool {tool_call.tool_name} not found")
try:
# 执行工具
result = tool["func"](**tool_call.params)
return ToolResponse(success=True, result=result)
except Exception as e:
return ToolResponse(success=False, error=str(e))
# WebSocket 端点
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await manager.connect(websocket)
try:
while True:
data = await websocket.receive_text()
# 简单的 echo 功能,后续可以扩展为工具调用
await manager.send_personal_message(f"Received: {data}", websocket)
except WebSocketDisconnect:
manager.disconnect(websocket)
# 运行服务器
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)代码示例:运行 MCP Server
# 保存代码为 mcp_server.py
# 运行服务器
python mcp_server.py代码示例:测试能力协商
# 使用 curl 测试能力协商端点
curl http://localhost:8000/api/v1/capabilities预期输出:
{
"version": "2.0",
"supported_protocols": ["http", "websocket"],
"tools": ["hello_world", "add_numbers", "multiply_numbers"]
}代码示例:测试 Hello World Tool
# 使用 curl 测试 hello_world 工具
curl -X POST -H "Content-Type: application/json" -d '{"tool_name": "hello_world", "params": {"name": "MCP"}}' http://localhost:8000/api/v1/tools/execute预期输出:
{
"success": true,
"result": "Hello, MCP!",
"error": null
}代码示例:测试 add_numbers 工具
# 使用 curl 测试 add_numbers 工具
curl -X POST -H "Content-Type: application/json" -d '{"tool_name": "add_numbers", "params": {"a": 10, "b": 20}}' http://localhost:8000/api/v1/tools/execute预期输出:
{
"success": true,
"result": 30.0,
"error": null
}代码示例:使用 wscat 测试 WebSocket 连接
# 安装 wscat
npm install -g wscat
# 连接 WebSocket 服务器
wscat -c ws://localhost:8000/ws
# 发送消息
> Hello MCP
< Received: Hello MCP从最小实现到完整 MCP Server,需要添加以下高级特性:
特性 | 最小实现 | 完整实现 |
|---|---|---|
代码量 | 约 150 行 | 约 1000+ 行 |
依赖数量 | 3 个 | 10+ 个 |
HTTP 支持 | 是 | 是 |
WebSocket 支持 | 基础支持 | 完整支持 |
工具注册 | 简单字典 | 完整的注册机制 |
认证授权 | 无 | 完整的认证授权 |
错误处理 | 基础错误处理 | 完善的错误处理 |
日志监控 | 无 | 完整的日志监控 |
异步支持 | 基础支持 | 完整的异步支持 |
部署复杂度 | 低 | 高 |
学习曲线 | 平缓 | 陡峭 |
特性 | Python 实现 | Node.js 实现 |
|---|---|---|
语言 | Python | JavaScript/TypeScript |
框架 | FastAPI | Express/NestJS |
性能 | 高 | 高 |
异步支持 | 原生支持 | 原生支持 |
WebSocket 支持 | 良好 | 优秀 |
生态系统 | 丰富 | 丰富 |
学习曲线 | 平缓 | 中等 |
部署便捷性 | 高 | 高 |
社区支持 | 活跃 | 活跃 |
特性 | MCP Server | OpenAI API |
|---|---|---|
协议标准 | MCP v2.0 | OpenAI Tools API |
部署方式 | 本地部署 | 云端服务 |
自定义工具 | 支持 | 支持 |
安全性 | 可控 | 依赖第三方 |
成本 | 免费 | 按使用付费 |
可扩展性 | 高 | 有限 |
实时通信 | 支持 | 有限支持 |
多模型支持 | 支持 | 仅支持 OpenAI 模型 |
下载并安装 Python 3.8+:https://www.python.org/downloads/
验证 Python 安装:
python --version安装 pip:
python -m ensurepip --upgrade验证 pip 安装:
pip --version# 创建并激活虚拟环境
python -m venv mcp-venv
source mcp-venv/bin/activate # Linux/macOS
mcp-venv\Scripts\activate # Windows
# 安装依赖
pip install fastapi uvicorn pydanticcurl 是一个命令行工具,用于发送 HTTP 请求。
安装:
基本用法:
# 发送 GET 请求
curl http://localhost:8000/api/v1/capabilities
# 发送 POST 请求
curl -X POST -H "Content-Type: application/json" -d '{"tool_name": "hello_world", "params": {"name": "MCP"}}' http://localhost:8000/api/v1/tools/executewscat 是一个命令行工具,用于测试 WebSocket 连接。
安装:
npm install -g wscat基本用法:
# 连接 WebSocket 服务器
wscat -c ws://localhost:8000/ws
# 发送消息
> Hello MCP
< Received: Hello MCPPostman 是一个图形化的 API 测试工具,支持 HTTP 和 WebSocket 测试。
下载:https://www.postman.com/downloads/
基本用法:
代码示例:添加 API Key 认证
from fastapi import FastAPI, HTTPException, Depends
from fastapi.security import APIKeyHeader
# API Key 配置
API_KEY = "your-secret-api-key"
API_KEY_NAME = "X-API-Key"
api_key_header = APIKeyHeader(name=API_KEY_NAME, auto_error=False)
# 依赖函数,用于验证 API Key
async def get_api_key(api_key_header: str = Depends(api_key_header)):
if api_key_header == API_KEY:
return api_key_header
else:
raise HTTPException(
status_code=401,
detail="Invalid or missing API Key",
headers={"WWW-Authenticate": API_KEY_NAME},
)
# 在端点中使用 API Key 认证
@app.post("/api/v1/tools/execute", response_model=ToolResponse)
async def execute_tool(tool_call: ToolCall, api_key: str = Depends(get_api_key)):
# 工具调用逻辑...代码示例:添加工具参数验证
from pydantic import BaseModel, Field
# 定义工具参数模型
class HelloWorldParams(BaseModel):
name: str = Field(..., min_length=1, max_length=100, description="Name to greet")
# 更新工具实现
def hello_world(params: HelloWorldParams):
"""Hello World Tool,返回问候语"""
return f"Hello, {params.name}!"
# 更新 Tool 注册表,支持参数验证
def validate_params(tool_name: str, params: dict):
"""验证工具参数"""
param_models = {
"hello_world": HelloWorldParams
}
if tool_name in param_models:
return param_models[tool_name](**params)
return params
# 更新工具调用端点
@app.post("/api/v1/tools/execute", response_model=ToolResponse)
async def execute_tool(tool_call: ToolCall):
# 获取工具
tool = tool_registry.get_tool(tool_call.tool_name)
if not tool:
return ToolResponse(success=False, error=f"Tool {tool_call.tool_name} not found")
try:
# 验证参数
validated_params = validate_params(tool_call.tool_name, tool_call.params)
# 执行工具
result = tool["func"](validated_params)
return ToolResponse(success=True, result=result)
except Exception as e:
return ToolResponse(success=False, error=str(e))问题:运行 MCP Server 时出现 “Address already in use” 错误。
解决方案:
查找占用端口的进程:
# Linux/macOS
lsof -i :8000
# Windows
netstat -ano | findstr :8000终止占用端口的进程:
# Linux/macOS
kill -9 <PID>
# Windows
taskkill /PID <PID> /F或者修改 MCP Server 的端口:
uvicorn.run(app, host="0.0.0.0", port=8001)问题:安装依赖时出现错误。
解决方案:
确保 pip 已更新:
pip install --upgrade pip尝试使用国内镜像源:
pip install -i https://pypi.tuna.tsinghua.edu.cn/simple fastapi uvicorn pydantic检查网络连接是否正常。
问题:调用工具时返回错误信息。
解决方案:
参考链接:
附录(Appendix):
关键词: MCP Server, 最小实现, Hello MCP, FastAPI, WebSocket, Python, AI 工具协议