从理论到实践：构建你的第一个MCP Server
MCP（Model Context Protocol）是2025年出现的开放协议，旨在标准化AI Agent与外部工具和资源的连接方式。如果说LLM是Agent的大脑，MCP就是Agent的神经系统。本文将从零开始带你构建一个完整的MCP Server。
MCP协议核心概念
为什么需要MCP
在MCP之前，每个Agent框架都有自己的工具接入方式——LangChain用Tools类、AutoGPT用插件系统、OpenAI用Function Calling。这导致工具开发者需要为每个框架分别适配，碎片化严重。
MCP的价值是提供统一标准：一次开发，任何支持MCP的Agent都能使用。到2026年，MCP已经成为事实标准，主流Agent框架和模型提供商都已支持。
MCP的三种能力
一个MCP Server可以暴露三种能力：
Tools（工具）：可执行的函数，Agent可以调用并获取返回值。如查询数据库、调用API、运行代码。
Resources（资源）：只读的数据源，Agent可以读取作为上下文。如文件内容、数据库表结构、API文档。
Prompts（模板）：预定义的Prompt模板，Agent可以基于模板创建对话。如代码审查模板、文档摘要模板。
构建你的第一个MCP Server
项目目标
我们将构建一个"代码分析助手"MCP Server，提供：
- 一个工具：分析Python文件的代码质量
- 一个资源：提供项目的README内容
- 一个模板：代码审查的Prompt模板
环境准备
# 安装MCP Python SDK
pip install mcp
# 项目结构
mcp-code-analyzer/
├── server.py # MCP Server主程序
├── analyzer.py # 代码分析逻辑
└── pyproject.toml
实现工具：代码质量分析
from mcp.server import Server
from mcp.types import Tool, TextContent
import ast
import os
server = Server("code-analyzer")
@server.list_tools()
async def list_tools():
return [
Tool(
name="analyze_code_quality",
description="分析Python文件的代码质量，返回复杂度、行数、函数数等指标",
inputSchema={
"type": "object",
"properties": {
"file_path": {
"type": "string",
"description": "Python文件的路径"
}
},
"required": ["file_path"]
}
)
]
@server.call_tool()
async def call_tool(name: str, arguments: dict):
if name == "analyze_code_quality":
file_path = arguments["file_path"]
result = analyze_python_file(file_path)
return [TextContent(
type="text",
text=f"代码分析结果：\n"
f"文件：{file_path}\n"
f"总行数：{result['total_lines']}\n"
f"代码行数：{result['code_lines']}\n"
f"函数数量：{result['functions']}\n"
f"类数量：{result['classes']}\n"
f"最大函数复杂度：{result['max_complexity']}\n"
f"建议：{'关注高复杂度函数' if result['max_complexity'] > 10 else '复杂度合理'}"
)]
def analyze_python_file(file_path: str) -> dict:
"""分析Python文件的代码质量"""
with open(file_path, 'r') as f:
source = f.read()
tree = ast.parse(source)
# 统计基础指标
lines = source.splitlines()
code_lines = [l for l in lines if l.strip() and not l.strip().startswith('#')]
# 统计函数和类
functions = [n for n in ast.walk(tree) if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef))]
classes = [n for n in ast.walk(tree) if isinstance(n, ast.ClassDef)]
# 计算圈复杂度
max_complexity = 0
for func in functions:
complexity = sum(1 for n in ast.walk(func)
if isinstance(n, (ast.If, ast.While, ast.For, ast.And, ast.Or))) + 1
max_complexity = max(max_complexity, complexity)
return {
"total_lines": len(lines),
"code_lines": len(code_lines),
"functions": len(functions),
"classes": len(classes),
"max_complexity": max_complexity
}
实现资源：README访问
from mcp.types import Resource
@server.list_resources()
async def list_resources():
resources = []
# 暴露项目根目录下的README
readme_path = os.path.join(os.getcwd(), "README.md")
if os.path.exists(readme_path):
resources.append(Resource(
uri="file:///README.md",
name="项目README",
description="项目的README文档",
mimeType="text/markdown"
))
return resources
@server.read_resource()
async def read_resource(uri: str):
if uri == "file:///README.md":
readme_path = os.path.join(os.getcwd(), "README.md")
with open(readme_path, 'r') as f:
content = f.read()
return content
raise ValueError(f"Unknown resource: {uri}")
实现Prompt模板：代码审查
from mcp.types import Prompt, PromptMessage, PromptArgument
@server.list_prompts()
async def list_prompts():
return [
Prompt(
name="code_review",
description="对指定代码进行深度审查",
arguments=[
PromptArgument(
name="code",
description="要审查的代码",
required=True
),
PromptArgument(
name="focus",
description="审查重点（security/performance/style/all）",
default="all"
)
]
)
]
@server.get_prompt()
async def get_prompt(name: str, arguments: dict):
if name == "code_review":
code = arguments.get("code", "")
focus = arguments.get("focus", "all")
focus_instructions = {
"security": "重点关注安全漏洞、输入验证、权限控制",
"performance": "重点关注性能瓶颈、内存使用、算法效率",
"style": "重点关注代码风格、命名规范、注释完整性",
"all": "全面审查：安全性、性能、可读性、最佳实践"
}
prompt_text = f"""你是一位资深代码审查专家。请审查以下代码：
```python
{code}
审查重点：{focus_instructions.get(focus, focus_instructions[‘all’])}
请按以下格式输出审查结果：
问题列表（按严重程度排序：Critical/Warning/Suggestion）
修复建议
总体评价（1-5分）"""
return PromptMessage( role="user", content=TextContent(type="text", text=prompt_text) )
### 启动Server
```python
async def main():
from mcp.server.stdio import stdio_server
async with stdio_server() as (read_stream, write_stream):
await server.run(read_stream, write_stream)
if __name__ == "__main__":
import asyncio
asyncio.run(main())
连接Agent使用
在Claude Desktop中配置
在Claude Desktop的配置文件中添加：
{
"mcpServers": {
"code-analyzer": {
"command": "python",
"args": ["/path/to/server.py"]
}
}
}
配置后重启Claude Desktop，你就可以让Claude分析代码文件了：“帮我分析一下main.py的代码质量”——Claude会自动调用你的MCP工具。
在自定义Agent中使用
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
async def use_mcp_server():
server_params = StdioServerParameters(
command="python",
args=["/path/to/server.py"]
)
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
# 初始化连接
await session.initialize()
# 列出可用工具
tools = await session.list_tools()
print(f"可用工具: {[t.name for t in tools]}")
# 调用工具
result = await session.call_tool(
"analyze_code_quality",
{"file_path": "main.py"}
)
print(result.content[0].text)
进阶实践
错误处理
MCP工具执行可能失败，需要优雅处理：
@server.call_tool()
async def call_tool(name: str, arguments: dict):
try:
if name == "analyze_code_quality":
file_path = arguments["file_path"]
if not os.path.exists(file_path):
return [TextContent(type="text", text=f"错误：文件不存在 - {file_path}")]
# ... 正常处理
except Exception as e:
return [TextContent(type="text", text=f"工具执行错误：{str(e)}")]
性能优化
对于耗时操作，使用进度通知：
@server.call_tool()
async def call_tool(name: str, arguments: dict):
if name == "analyze_large_codebase":
# 发送进度通知
await server.send_notification("progress", {"current": 30, "total": 100, "message": "正在扫描文件..."})
# ... 处理
await server.send_notification("progress", {"current": 60, "total": 100, "message": "正在分析..."})
# ... 处理
安全考虑
- 路径验证：防止路径遍历攻击，限制可访问的文件路径范围
- 资源限制：限制单个工具调用的最大执行时间和内存
- 权限控制：敏感操作需要用户确认
测试你的MCP Server
import pytest
from mcp import ClientSession
from mcp.client.stdio import stdio_client
@pytest.mark.asyncio
async def test_analyze_code_quality():
server_params = StdioServerParameters(
command="python",
args=["server.py"]
)
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
result = await session.call_tool(
"analyze_code_quality",
{"file_path": "test_sample.py"}
)
assert "总行数" in result.content[0].text
assert "函数数量" in result.content[0].text
结语
MCP的出现标志着AI Agent工具生态从碎片化走向标准化。构建MCP Server不难，难的是设计好工具的边界——什么应该暴露给Agent，什么不应该；什么作为工具（可执行），什么作为资源（只读）。好的MCP Server设计能让Agent的能力自然延伸，而非生硬拼接。在2026年，MCP生态已经有了数千个开源Server，覆盖从数据库到代码仓库到云服务的各种场景。掌握MCP开发，就是掌握了Agent时代的"API设计"技能。
本文同步发布于 硅基AGI论坛