# 基于MCP协议搭建JSON-RPC 2.0通信的本地MCP服务器完整指南 ## 1. MCP协议与JSON-RPC 2.0基础 ### 1.1 MCP协议架构概述 Model Context Protocol（MCP）是由Anthropic公司推出的开放协议，旨在为AI模型与外部世界的互动提供统一的接口[ref_1]。MCP采用客户端-服务器架构，主要包含以下核心组件： | 组件类型 | 角色说明 | 功能描述 | |---------|---------|---------| | MCP Host（主机） | 如Claude Desktop、IDE、AI应用等 | 负责发起连接并管理多个客户端实例 | | MCP Client（客户端） | 由Host创建 | 与特定MCP Server建立一对一连接，负责协议通信和能力协商 | | MCP Server（服务器） | 提供特定数据源或工具功能 | 响应客户端的请求，执行具体操作 | ### 1.2 JSON-RPC 2.0通信基础 MCP协议基于JSON-RPC 2.0消息格式，支持请求、响应和通知三种消息类型[ref_1]。JSON-RPC 2.0是一种轻量级远程过程调用协议，具有以下核心特性[ref_2]： - **简单轻量**：基于JSON格式，协议结构简洁 - **传输协议无关性**：可在HTTP、WebSocket、Stdio等多种传输层上运行 - **严格编码要求**：使用UTF-8编码确保字符集兼容性 - **批量请求支持**：允许在单个消息中发送多个请求 ## 2. MCP服务器搭建环境准备 ### 2.1 开发环境配置 首先需要配置Python开发环境，以下是具体步骤： ```bash # 下载uv包管理工具（推荐用于MCP项目） # macOS/Linux curl -LsSf https://astral.sh/uv/install.sh | sh # Windows系统 # curl -LsSf https://astral.sh/uv/install.ps1 | powershell ``` 创建并激活虚拟环境： ```bash # 创建虚拟环境 uv venv # 激活虚拟环境 # macOS/Linux source .venv/bin/activate # Windows # .venv\Scripts\activate ``` ### 2.2 依赖包安装 安装MCP协议核心库和必要的依赖： ```bash # 安装MCP核心库及CLI工具 uv add "mcp[cli]" httpx # 安装其他可能需要的依赖 uv add pydantic loguru ``` ## 3. 实现MCP服务器核心代码 ### 3.1 基础服务器结构 创建一个完整的MCP服务器示例，支持工具调用和资源访问： ```python #!/usr/bin/env python3 """ 基于MCP协议的本地服务器实现 支持JSON-RPC 2.0通信协议 """ import asyncio import json from typing import Any, Dict, List, Optional from mcp import MCPServer, mcp from mcp.types import Tool, TextContent, Resource class LocalMCPServer(MCPServer): """本地MCP服务器实现""" def __init__(self): super().__init__() self._tools = self._initialize_tools() self._resources = self._initialize_resources() def _initialize_tools(self) -> List[Tool]: """初始化服务器提供的工具列表""" return [ Tool( name="calculate", description="执行数学计算", inputSchema={ "type": "object", "properties": { "expression": { "type": "string", "description": "数学表达式，如: 2+3*4" } }, "required": ["expression"] } ), Tool( name="get_time_info", description="获取时间信息", inputSchema={ "type": "object", "properties": { "timezone": { "type": "string", "description": "时区，如: Asia/Shanghai" } } } ) ] def _initialize_resources(self) -> List[Resource]: """初始化服务器提供的资源列表""" return [ Resource( uri="file:///server_info", name="server_info", description="服务器基本信息", mimeType="text/plain" ) ] @mcp.tool() async def calculate(self, expression: str) -> str: """ 执行数学计算工具 Args: expression: 数学表达式 Returns: 计算结果字符串 """ try: # 安全地评估数学表达式 result = eval(expression, {"__builtins__": None}, {}) return f"计算结果: {expression} = {result}" except Exception as e: return f"计算错误: {str(e)}" @mcp.tool() async def get_time_info(self, timezone: str = "UTC") -> str: """ 获取时间信息工具 Args: timezone: 时区名称 Returns: 时间信息字符串 """ from datetime import datetime import pytz try: tz = pytz.timezone(timezone) current_time = datetime.now(tz) return f"{timezone} 当前时间: {current_time.strftime('%Y-%m-%d %H:%M:%S %Z')}" except Exception as e: return f"时区错误: {str(e)}" @mcp.resource("file:///server_info") async def get_server_info(self) -> List[TextContent]: """获取服务器信息资源""" return [ TextContent( type="text", text="MCP本地服务器信息:\n- 协议版本: MCP 1.0\n- 通信协议: JSON-RPC 2.0\n- 支持工具: 计算器、时间查询\n- 状态: 运行中" ) ] async def initialize(self) -> None: """服务器初始化""" await super().initialize() print("MCP服务器初始化完成") async def list_tools(self) -> List[Tool]: """返回可用工具列表""" return self._tools async def list_resources(self) -> List[Resource]: """返回可用资源列表""" return self._resources async def main(): """主函数 - 启动MCP服务器""" # 创建服务器实例 server = LocalMCPServer() # 使用Stdio传输启动服务器 print("启动MCP服务器 (Stdio传输)...") await mcp.run(server, transport='stdio') # 如需使用SSE传输，取消注释下面这行 # await mcp.run(server, transport='sse') if __name__ == "__main__": # 运行服务器 asyncio.run(main()) ``` ### 3.2 配置服务器参数 创建服务器配置文件 `.env`： ```bash # MCP服务器环境配置 SERVER_NAME=local_mcp_server SERVER_VERSION=1.0.0 LOG_LEVEL=INFO # 如需外部API集成，可在此添加API密钥 # WEATHER_API_KEY=your_api_key_here # FINANCIAL_API_KEY=your_financial_api_key ``` ## 4. 服务器启动与测试 ### 4.1 启动MCP服务器 保存上述代码为 `mcp_server.py`，然后启动服务器： ```bash # 激活虚拟环境 source .venv/bin/activate # 启动MCP服务器 uv run mcp_server.py ``` ### 4.2 测试服务器功能 创建测试客户端脚本来验证服务器功能： ```python #!/usr/bin/env python3 """ MCP服务器测试客户端 """ import asyncio import json import subprocess import sys async def test_mcp_server(): """测试MCP服务器功能""" # 启动服务器进程 server_process = subprocess.Popen( [sys.executable, "mcp_server.py"], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True ) try: # 测试工具列表查询 list_tools_request = { "jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {} } # 发送请求 server_process.stdin.write(json.dumps(list_tools_request) + "\n") server_process.stdin.flush() # 读取响应 response = server_process.stdout.readline() print("工具列表响应:", response) # 测试计算工具 calculate_request = { "jsonrpc": "2.0", "id": 2, "method": "tools/call", "params": { "name": "calculate", "arguments": { "expression": "2+3*4" } } } server_process.stdin.write(json.dumps(calculate_request) + "\n") server_process.stdin.flush() calculate_response = server_process.stdout.readline() print("计算工具响应:", calculate_response) finally: # 清理进程 server_process.terminate() server_process.wait() if __name__ == "__main__": asyncio.run(test_mcp_server()) ``` ## 5. 集成到开发环境 ### 5.1 配置Claude Desktop 对于macOS系统，配置Claude Desktop使用本地MCP服务器： ```json // ~/Library/Application Support/Claude/claude_desktop_config.json { "mcpServers": { "local_mcp_server": { "command": "uv", "args": [ "--directory", "/absolute/path/to/your/mcp/server", "run", "mcp_server.py" ] } } } ``` ### 5.2 配置Cursor IDE 在Cursor设置中添加MCP服务器： ```json { "mcpServers": { "local_mcp_server": { "command": "uv", "args": [ "--directory", "/absolute/path/to/your/mcp/server", "run", "mcp_server.py" ] } } } ``` ## 6. 安全配置与最佳实践 ### 6.1 安全措施实现 在MCP服务器中实施安全防护： ```python import re from mcp import MCPServer, mcp class SecureMCPServer(MCPServer): """带有安全措施的MCP服务器""" def __init__(self): super().__init__() self._allowed_expressions = re.compile(r'^[0-9+\-*/().\s]+$') @mcp.tool() async def safe_calculate(self, expression: str) -> str: """ 安全数学计算 - 防止代码注入 Args: expression: 数学表达式 Returns: 计算结果 """ # 输入验证 if not self._allowed_expressions.match(expression): return "错误: 包含不安全的字符" try: # 限制表达式复杂度 if len(expression) > 100: return "错误: 表达式过长" # 安全计算 result = eval(expression, {"__builtins__": None}, {}) return f"安全计算结果: {result}" except Exception as e: return f"计算错误: {str(e)}" ``` ### 6.2 性能优化建议 ```python import time from functools import lru_cache class OptimizedMCPServer(MCPServer): """性能优化的MCP服务器""" def __init__(self): super().__init__() self._request_count = 0 self._start_time = time.time() @lru_cache(maxsize=100) def _cached_calculation(self, expression: str) -> float: """缓存常用计算结果""" return eval(expression, {"__builtins__": None}, {}) async def get_server_stats(self) -> Dict[str, Any]: """获取服务器统计信息""" uptime = time.time() - self._start_time return { "requests_processed": self._request_count, "uptime_seconds": uptime, "requests_per_second": self._request_count / uptime if uptime > 0 else 0 } ``` ## 7. 故障排除与调试 ### 7.1 常见问题解决 | 问题现象 | 可能原因 | 解决方案 | |---------|---------|---------| | 服务器启动失败 | 依赖包缺失 | 运行 `uv add "mcp[cli]"` 安装依赖 | | JSON-RPC通信错误 | 消息格式不正确 | 验证JSON格式，确保符合JSON-RPC 2.0规范[ref_3] | | 工具调用超时 | 处理逻辑复杂 | 优化工具实现，添加超时机制 | | 权限拒绝 | 文件访问权限 | 检查服务器对相关资源的访问权限 | ### 7.2 日志调试配置 添加详细日志记录以便调试： ```python import logging from mcp import MCPServer, mcp # 配置日志 logging.basicConfig( level=logging.DEBUG, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) class LoggedMCPServer(MCPServer): """带有详细日志的MCP服务器""" def __init__(self): super().__init__() self.logger = logging.getLogger(__name__) @mcp.tool() async def calculate(self, expression: str) -> str: self.logger.info(f"计算工具调用: {expression}") try: result = eval(expression, {"__builtins__": None}, {}) self.logger.info(f"计算成功: {expression} = {result}") return f"结果: {result}" except Exception as e: self.logger.error(f"计算错误: {e}") return f"错误: {str(e)}" ``` 通过以上完整实现，您可以成功搭建一个基于MCP协议、支持JSON-RPC 2.0通信的本地MCP服务器。该服务器提供了标准化的工具调用接口，能够与各种MCP客户端（如Claude Desktop、Cursor等）无缝集成，为AI应用提供丰富的外部能力扩展。