当前位置：首页 > news >正文

从零构建 MCP Server：协议原理 + 完整实战

news 2026/3/26 22:28:58

🚀 从零构建 MCP Server：协议原理 + 完整实战

一篇真正写给后端 / 架构 / AI 工程师的 MCP 深度实战指南不是“会用”，而是“知道它为什么这样设计”

写在前面：为什么你必须认真了解 MCP？

过去一年，大模型真正的瓶颈已经不在「推理能力」，而在于：

❌ 无法访问实时数据
❌ 无法安全调用企业内部系统
❌ 无法被工程化治理

MCP（Model Context Protocol）正是为了解决这些问题而生。

MCP 不是一个 SDK 它是 「大模型与真实世界之间的协议层」

本文将带你从 0 到 1 实现一个 MCP Server，并深入拆解它背后的设计思想。

我来带你从零开始构建一个MCP（Model Context Protocol）Server，深入理解MCP背后的技术原理。

一、MCP核心概念

1.1 什么是MCP？

MCP是一个开放协议，允许大模型安全地与外部工具和数据源交互，解决模型知识的局限性问题。

1.2 MCP架构

┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐ │ LLM Client │───▶│ MCP Server │───▶│ 外部工具/数据源 │ │ (如 Claude.app) │◀───│ (我们实现的) │◀───│ (如数据库/API) │ └─────────────────┘ └─────────────────┘ └─────────────────┘

二、动手实现MCP Server

2.1 环境搭建

# 创建项目目录 mkdir mcp-server-tutorial cd mcp-server-tutorial # 初始化Python环境 python -m venv venv source venv/bin/activate # Linux/Mac # venv\Scripts\activate # Windows # 安装依赖 pip install pydantic jsonschema

2.2 基础MCP Server实现

server.py

#!/usr/bin/env python3 """ MCP Server 基础实现 理解MCP协议的核心原理 """ import json import sys import asyncio from typing import Dict, List, Any, Optional from enum import Enum from dataclasses import dataclass import logging # 配置日志 logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class MessageType(Enum): """MCP消息类型""" REQUEST = "request" RESPONSE = "response" NOTIFICATION = "notification" @dataclass class McpMessage: """MCP消息基类""" jsonrpc: str = "2.0" def to_dict(self) -> Dict[str, Any]: """转换为字典""" raise NotImplementedError class RequestMessage(McpMessage): """请求消息""" def __init__(self, method: str, params: Dict[str, Any] = None, id: Any = None): self.method = method self.params = params or {} self.id = id def to_dict(self) -> Dict[str, Any]: return { "jsonrpc": self.jsonrpc, "method": self.method, "params": self.params, "id": self.id } class ResponseMessage(McpMessage): """响应消息""" def __init__(self, id: Any, result: Any = None, error: Dict[str, Any] = None): self.id = id self.result = result self.error = error def to_dict(self) -> Dict[str, Any]: response = { "jsonrpc": self.jsonrpc, "id": self.id } if self.error: response["error"] = self.error else: response["result"] = self.result return response class Tool: """MCP工具定义""" def __init__(self, name: str, description: str, input_schema: Dict[str, Any]): self.name = name self.description = description self.input_schema = input_schema def to_dict(self) -> Dict[str, Any]: return { "name": self.name, "description": self.description, "inputSchema": self.input_schema } class BaseMcpServer: """基础MCP Server实现""" def __init__(self, name: str, version: str): self.name = name self.version = version self.tools: Dict[str, Tool] = {} self.handlers: Dict[str, callable] = { "initialize": self._handle_initialize, "tools/list": self._handle_tools_list, "tools/call": self._handle_tools_call, "ping": self._handle_ping } def register_tool(self, tool: Tool) -> None: """注册工具""" self.tools[tool.name] = tool def register_handler(self, method: str, handler: callable) -> None: """注册自定义处理器""" self.handlers[method] = handler async def handle_message(self, message_str: str) -> str: """处理传入的MCP消息""" try: message_data = json.loads(message_str) message_type = self._get_message_type(message_data) if message_type == MessageType.REQUEST: return await self._handle_request(message_data) else: logger.warning(f"未处理的消息类型: {message_type}") return "" except json.JSONDecodeError: error_response = ResponseMessage( id=None, error={"code": -32700, "message": "Parse error"} ) return json.dumps(error_response.to_dict()) def _get_message_type(self, message: Dict[str, Any]) -> MessageType: """判断消息类型""" if "method" in message: return MessageType.REQUEST elif "result" in message or "error" in message: return MessageType.RESPONSE else: return MessageType.NOTIFICATION async def _handle_request(self, request_data: Dict[str, Any]) -> str: """处理请求""" method = request_data.get("method") request_id = request_data.get("id") if method in self.handlers: try: params = request_data.get("params", {}) result = await self.handlers[method](params) if request_id is not None: response = ResponseMessage(id=request_id, result=result) return json.dumps(response.to_dict()) except Exception as e: logger.error(f"处理请求 {method} 时出错: {e}") error_response = ResponseMessage( id=request_id, error={"code": -32603, "message": str(e)} ) return json.dumps(error_response.to_dict()) else: error_response = ResponseMessage( id=request_id, error={"code": -32601, "message": f"Method not found: {method}"} ) return json.dumps(error_response.to_dict()) async def _handle_initialize(self, params: Dict[str, Any]) -> Dict[str, Any]: """处理初始化请求""" logger.info("收到初始化请求") return { "protocolVersion": "2024-11-05", "capabilities": { "tools": {} }, "serverInfo": { "name": self.name, "version": self.version }

查看全文

http://www.jsqmd.com/news/347260/