MCP 协议实战：告别硬编码，用 Model Context Protocol 让你的 AI 工具即插即用

引言2025 年底Anthropic 推出了 Model Context ProtocolMCP—— 一个让 AI 模型与外部工具、数据源交互的开放标准。在此之前让 LLM 调用外部工具是一件极其碎片化的事情。OpenAI 有 Function CallingAnthropic 有 Tool UseGoogle 有 Function Declaration——每种格式的 JSON Schema 都不同每次接入新工具都要写胶水代码。一个接了 5 个外部 API 的 AI 应用可能有 500 行代码是在做格式转换。MCP 改变了这一切。它定义了一套统一的协议Server 暴露能力Client 发现并调用LLM 只需要理解标准的 MCP 接口。一、MCP 架构三个核心概念MCP 的设计极其简洁只有三个核心概念┌─────────────┐ MCP Protocol ┌─────────────┐ │ MCP Client │ ◄──────────────────► │ MCP Server │ │ (Host App) │ │ (Tool/Data) │ └─────────────┘ └─────────────┘ │ │ │ 1. 发现: tools/list │ │ 2. 调用: tools/call │ │ 3. 资源: resources/read │MCP Server— 提供工具和资源的服务端。一个 Server 可以暴露数据库查询、文件操作、API 调用等能力。MCP Client— 嵌入在 AI 应用如 Claude Desktop、自定义 Agent中的客户端负责发现 Server 的能力并调用它们。Transport— Server 和 Client 之间的通信方式。支持stdio本地进程通信和HTTP SSE远程通信两种模式。二、构建你的第一个 MCP Server依赖pip install mcp2.1 最简单的 MCP Server天气预报工具# weather_server.py import asyncio import httpx from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import Tool, TextContent # 1. 创建 MCP Server 实例 server Server(weather-service) # 2. 注册工具列表 server.list_tools() async def list_tools() - list[Tool]: 告诉 Client 这个 Server 能做什么 return [ Tool( nameget_weather, description获取指定城市的实时天气信息, inputSchema{ type: object, properties: { city: { type: string, description: 城市名称英文如 Beijing, Tokyo } }, required: [city] } ), Tool( namecompare_weather, description比较两个城市的天气差异, inputSchema{ type: object, properties: { city_a: {type: string}, city_b: {type: string} }, required: [city_a, city_b] } ) ] # 3. 实现工具调用 server.call_tool() async def call_tool(name: str, arguments: dict) - list[TextContent]: 处理来自 Client 的工具调用请求 async with httpx.AsyncClient() as client: if name get_weather: city arguments[city] # 调用免费的天气 API resp await client.get( fhttps://wttr.in/{city}?formatj1, timeout10 ) data resp.json() current data[current_condition][0] return [TextContent( typetext, textf{city} 当前天气:\n f 温度: {current[temp_C]}°C\n f 湿度: {current[humidity]}%\n f 天气: {current[weatherDesc][0][value]}\n f 风速: {current[windspeedKmph]} km/h )] elif name compare_weather: city_a arguments[city_a] city_b arguments[city_b] # 并发查询两个城市 resp_a, resp_b await asyncio.gather( client.get(fhttps://wttr.in/{city_a}?formatj1), client.get(fhttps://wttr.in/{city_b}?formatj1) ) temp_a resp_a.json()[current_condition][0][temp_C] temp_b resp_b.json()[current_condition][0][temp_C] diff abs(float(temp_a) - float(temp_b)) return [TextContent( typetext, textf温度对比:\n f {city_a}: {temp_a}°C\n f {city_b}: {temp_b}°C\n f 温差: {diff}°C\n f {温差较大注意衣物调整 if diff 10 else 温度相近} )] # 4. 启动 Serverstdio 模式 async def main(): async with stdio_server() as (read_stream, write_stream): await server.run( read_stream, write_stream, server.create_initialization_options() ) if __name__ __main__: asyncio.run(main())2.2 配置到 Claude Desktop将 Server 注册到 Claude Desktop 只需要一行 JSON 配置{ mcpServers: { weather: { command: python, args: [/path/to/weather_server.py], description: 实时天气查询服务 } } }配置文件位置 -Windows:%APPDATA%/Claude/claude_desktop_config.json-macOS:~/Library/Application Support/Claude/claude_desktop_config.json-Linux:~/.config/Claude/claude_desktop_config.json重启 Claude Desktop 后点击工具图标锤子就能看到你的 Server。AI 会自动发现工具并在需要时调用。三、进阶数据库 MCP Server真实项目中最常见的需求是让 LLM 查询业务数据库。下面构建一个 MySQL MCP Server# db_server.py import asyncio import aiomysql from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import Tool, TextContent server Server(mysql-explorer) # 数据库连接池 pool None server.list_tools() async def list_tools() - list[Tool]: return [ Tool( namelist_tables, description列出数据库中的所有表, inputSchema{ type: object, properties: {}, required: [] } ), Tool( namedescribe_table, description查看表结构字段名、类型、注释, inputSchema{ type: object, properties: { table_name: { type: string, description: 表名 } }, required: [table_name] } ), Tool( namerun_safe_query, description执行只读 SQL 查询SELECT / SHOW / DESCRIBE, inputSchema{ type: object, properties: { query: { type: string, description: SQL 查询语句仅限只读 }, limit: { type: integer, default: 20, description: 返回行数上限 } }, required: [query] } ) ] server.call_tool() async def call_tool(name: str, arguments: dict) - list[TextContent]: global pool if pool is None: pool await aiomysql.create_pool( hostlocalhost, port3306, userreadonly_user, passwordyour_password, dbyour_database, minsize1, maxsize5 ) async with pool.acquire() as conn: async with conn.cursor(aiomysql.DictCursor) as cur: if name list_tables: await cur.execute(SHOW TABLES) tables [row[fTables_in_your_database] for row in await cur.fetchall()] return [TextContent( typetext, textf数据库包含 {len(tables)} 张表:\n \n.join(f • {t} for t in tables) )] elif name describe_table: table arguments[table_name] # 参数化防止注入 await cur.execute(DESCRIBE {}.format(table.replace(, ))) columns await cur.fetchall() lines [f表 {table} 结构:] for col in columns: lines.append( f {col[Field]:20s} {col[Type]:15s} f{NOT NULL if col[Null] NO else NULL:10s} f{col.get(Comment, )} ) return [TextContent(typetext, text\n.join(lines))] elif name run_safe_query: query arguments[query].strip() limit arguments.get(limit, 20) # 安全拦截只允许只读操作 dangerous [INSERT, UPDATE, DELETE, DROP, ALTER, CREATE, TRUNCATE, REPLACE] first_word query.split()[0].upper() if first_word in dangerous: return [TextContent( typetext, textf⛔ 禁止执行写操作: {first_word} )] # 自动追加 LIMIT if LIMIT not in query.upper(): query f{query.rstrip(;)} LIMIT {limit} await cur.execute(query) rows await cur.fetchall() if not rows: return [TextContent(typetext, text查询返回空结果)] # 格式化输出为 Markdown 表格 headers list(rows[0].keys()) lines [| | .join(headers) |] lines.append(| |.join([---] * len(headers)) |) for row in rows[:limit]: vals [str(v)[:50] for v in row.values()] lines.append(| | .join(vals) |) lines.append(f\n*共 {len(rows)} 行显示前 {min(len(rows), limit)} 行*) return [TextContent(typetext, text\n.join(lines))] async def main(): async with stdio_server() as (read_stream, write_stream): await server.run(read_stream, write_stream, server.create_initialization_options()) if __name__ __main__: asyncio.run(main())关键安全设计- 数据库连接使用只读账号readonly_user - 写入操作被白名单拦截只允许 SELECT/SHOW/DESCRIBE - 查询结果自动 LIMIT 限制防止返回海量数据 - 表名使用参数化处理防止 SQL 注入四、远程 MCP Server通过 HTTP 暴露服务对于需要跨机器访问的场景MCP 支持 HTTP SSE 传输# remote_server.py from mcp.server import Server from mcp.server.http import create_http_server import uvicorn server Server(remote-api-gateway) server.list_tools() async def list_tools(): return [ Tool( namecall_internal_api, description调用公司内部 API, inputSchema{ type: object, properties: { endpoint: {type: string}, params: {type: object} }, required: [endpoint] } ) ] server.call_tool() async def call_tool(name: str, arguments: dict): # ... 工具实现 pass # 创建 HTTP 应用 app create_http_server(server) if __name__ __main__: uvicorn.run(app, host0.0.0.0, port8800)Client 端通过 HTTP URL 连接# remote_client.py from mcp.client import Client from mcp.client.http import http_client async def main(): async with http_client(http://mcp.internal.company.com:8800) as (read, write): client Client(my-app, read, write) # 发现远程工具 tools await client.list_tools() print(f发现 {len(tools)} 个远程工具) # 调用远程工具 result await client.call_tool( call_internal_api, {endpoint: /v1/users, params: {status: active}} ) print(result) import asyncio asyncio.run(main())五、MCP 生态生产级 Server 推荐MCP 发布不到一年社区已经沉淀出了一批生产可用的 ServerServer功能推荐场景anthropic/mcp-server-puppeteer浏览器自动化Web 抓取、E2E 测试anthropic/mcp-server-filesystem文件系统操作代码分析、文档处理mcp-server-postgresPostgreSQL 查询数据探索、报表生成mcp-server-githubGitHub API 集成PR 审查、Issue 管理mcp-server-brave-search搜索引擎集成实时信息检索mcp-server-slackSlack 消息管理团队协作自动化安装社区 Server 只需一行# 以 Brave Search MCP Server 为例 npx anthropic/mcp-server-brave-search --api-key YOUR_KEY # Python 版本 pip install mcp-server-brave-search python -m mcp_server_brave_search --api-key YOUR_KEY配置后在 Claude Desktop 中添加即可使用——AI 会自动发现搜索工具。这就是 MCP 的核心理念工具开发者只需实现一次 Server所有兼容 MCP 的 AI 应用都能即插即用。结语MCP 解决了一个 AI 工程领域的基础设施问题碎片化的工具集成。在 MCP 之前每接入一个新工具就要写一套新的胶水代码。在 MCP 之后工具开发者只需实现一个 Server所有兼容 MCP 的 AI 应用自动发现并使用。对于 AI 应用开发者来说MCP 带来的最大变化是你不再需要为不同模型适配不同的 Function Calling 格式。你的工具只写一次剩下的交给协议。现在去构建你的第一个 MCP Server 吧。本文代码基于 MCP Python SDK v1.x (2026)。MCP 协议规范见 modelcontextprotocol.io。