从理论到实践:构建你的第一个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’])}
...