大模型 MCP 本质原理：从协议到代码实现

1. 引言为什么需要 MCP大模型本身无法直接访问外部数据、调用工具或操作系统资源。传统做法是为每个应用单独编写工具调用逻辑导致重复开发、协议割裂、维护成本高。MCPModel Context Protocol模型上下文协议正是为了解决这一问题而诞生的开放标准它为大模型与外部工具、数据源之间定义了一套统一的通信协议。MCP 的核心价值在于一次接入处处可用。开发者只需按照 MCP 规范实现一次工具服务任何支持 MCP 的大模型应用都能直接调用无需为每个模型单独适配。2. MCP 协议架构MCP 采用客户端-服务器架构包含三个核心角色MCP Host大模型应用本身如 Claude Desktop、IDE 插件等负责发起请求并处理结果。MCP Client运行在 Host 内部的协议客户端负责与 Server 建立连接、发送请求、接收响应。MCP Server暴露工具、资源和提示词的独立服务可以是本地进程也可以是远程 HTTP 服务。三者之间的关系可以用下图表示flowchart LR A[大模型应用 Host] -- B[MCP Client] B --|JSON-RPC 2.0| C[MCP Server] C -- D[本地文件系统] C -- E[数据库] C -- F[外部 API]3. 传输层与消息格式MCP 协议基于JSON-RPC 2.0作为消息格式传输层支持两种模式stdio客户端与服务器通过标准输入输出进行通信适用于本地进程。Streamable HTTP通过 HTTP 进行通信适用于远程服务。一条典型的 MCP 请求消息结构如下{ jsonrpc: 2.0, id: 1, method: tools/call, params: { name: get_weather, arguments: { city: 北京 } } }响应消息结构如下{ jsonrpc: 2.0, id: 1, result: { content: [ { type: text, text: 北京今天晴气温 25°C } ] } }4. 核心原语工具、资源与提示词MCP 定义了三种核心原语分别对应不同的能力维度原语作用典型方法工具Tools可被模型调用的函数执行具体操作tools/list、tools/call资源Resources向模型暴露只读数据如文件、数据库记录resources/list、resources/read提示词Prompts预定义的提示模板引导模型完成特定任务prompts/list、prompts/get其中工具是最常用的原语。模型通过 tools/list 发现可用工具再通过 tools/call 调用具体工具并获取结果。5. 从零实现一个 MCP Server下面我们使用 Python 和官方 SDK 从零实现一个完整的 MCP Server。首先安装依赖pip install mcp创建一个简单的文件读取工具from mcp.server.fastmcp import FastMCP 创建 MCP Server 实例 mcp FastMCP(FileServer) mcp.tool() def read_file(path: str) - str: 读取指定路径的文件内容 with open(path, r, encodingutf-8) as f: return f.read() mcp.tool() def list_files(directory: str) - list[str]: 列出目录下的所有文件 import os return os.listdir(directory) if name main: mcp.run(transportstdio)上面的代码通过mcp.tool()装饰器将普通函数暴露为 MCP 工具。FastMCP 会自动处理 JSON-RPC 消息的编解码、协议握手和工具注册。6. 实现 MCP Client 并调用工具接下来实现一个 MCP Client连接上面的 Server 并调用工具import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def main(): # 配置服务器启动参数 server_params StdioServerParameters( commandpython, args[file_server.py] ) # 建立 stdio 连接 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(可用工具:, [t.name for t in tools.tools]) # 调用工具 result await session.call_tool( list_files, arguments{directory: .} ) print(目录内容:, result.content) asyncio.run(main())运行客户端后可以看到它成功发现并调用了 Server 暴露的工具。这就是 MCP 最基本的完整闭环。7. 深入协议初始化握手与能力协商MCP 连接建立后客户端和服务器首先要进行初始化握手交换协议版本和能力信息。握手过程如下sequenceDiagram participant C as Client participant S as Server C-S: initialize (协议版本, 客户端能力) S--C: initialize 响应 (服务器能力) C-S: initialized 通知 C-S: tools/list S--C: 工具列表 C-S: tools/call S--C: 工具结果初始化请求的核心字段包括protocolVersion客户端支持的协议版本号。capabilities客户端支持的能力如工具、资源、提示词等。clientInfo客户端名称和版本。服务器在响应中返回自己支持的协议版本和能力。如果版本不兼容双方需要协商或拒绝连接。8. 进阶带鉴权的远程 MCP Server生产环境中MCP Server 通常以 HTTP 方式部署并需要鉴权。下面实现一个基于 FastAPI 的远程 MCP Serverfrom mcp.server.fastmcp import FastMCP from mcp.server.sse import SseServerTransport from fastapi import FastAPI from fastapi.responses import StreamingResponse import uvicorn mcp FastMCP(RemoteServer) mcp.tool() def get_user_info(user_id: str) - dict: 根据用户 ID 查询用户信息 # 实际项目中这里会查询数据库 return {id: user_id, name: 张三, level: VIP} 创建 FastAPI 应用 app FastAPI() sse SseServerTransport(/messages) app.post(/messages) async def handle_message(request: Request): 处理客户端发来的 JSON-RPC 消息 async with sse.connect_sse(request.scope, request.receive, request._send) as streams: await mcp.run(streams[0], streams[1], mcp.create_initialization_options()) app.get(/sse) async def handle_sse(request: Request): SSE 端点用于建立事件流连接 async def event_generator(): async with sse.connect_sse(request.scope, request.receive, request._send) as streams: await mcp.run(streams[0], streams[1], mcp.create_initialization_options()) return StreamingResponse(event_generator(), media_typetext/event-stream) if name main: uvicorn.run(app, host0.0.0.0, port8000)远程部署时鉴权通常通过 HTTP 头传递 Token 实现。客户端在连接时携带 Authorization 头服务器在消息处理前校验身份。9. 实战让大模型通过 MCP 操作数据库下面实现一个完整的实战案例让大模型通过 MCP 查询和操作 SQLite 数据库。首先创建数据库工具import sqlite3 from mcp.server.fastmcp import FastMCP mcp FastMCP(DatabaseServer) DB_PATH app.db def get_connection(): conn sqlite3.connect(DB_PATH) conn.row_factory sqlite3.Row return conn mcp.tool() def query(sql: str) - list[dict]: 执行 SQL 查询语句返回查询结果 conn get_connection() try: cursor conn.execute(sql) rows cursor.fetchall() return [dict(row) for row in rows] finally: conn.close() mcp.tool() def execute(sql: str) - str: 执行 SQL 写操作INSERT/UPDATE/DELETE conn get_connection() try: conn.execute(sql) conn.commit() return 执行成功 except Exception as e: return f执行失败: {str(e)} finally: conn.close() if name main: mcp.run(transportstdio)初始化数据库并插入测试数据sqlite3 app.db CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT, age INTEGER); sqlite3 app.db INSERT INTO users (name, age) VALUES (张三, 28), (李四, 32);现在大模型应用可以通过 MCP Client 连接这个 Server用自然语言让模型生成 SQL 并查询数据库。模型先调用 query 工具了解表结构再根据用户问题生成查询语句。10. 总结与最佳实践MCP 的本质可以概括为用统一的 JSON-RPC 协议把大模型与外部世界连接起来。它通过工具、资源和提示词三种原语实现了能力发现、调用和结果返回的标准化。在实际项目中建议遵循以下最佳实践工具粒度适中每个工具只做一件事参数设计清晰便于模型理解。提供详细描述工具和参数的描述直接影响模型调用的准确性。做好错误处理工具内部异常要转换为可读的错误信息返回给模型。注意安全边界对工具调用做权限控制避免模型执行危险操作。合理设计超时长时间运行的工具要设置超时避免阻塞模型响应。掌握 MCP 协议的本质你就能为大模型应用构建出强大、可扩展的工具生态让模型真正成为连接业务系统的智能入口。