MCP协议解析:JSON-RPC 2.0驱动LLM连接工具与数据,赋能AI Agent开发
摘要:MCP协议:用JSON-RPC 2.0给LLM装上“神经突触”,Agent开发就靠它了想让你的AI Agent真正“动起来”,而不是只会聊天?关键在于连接。MCP(Model Context Protocol)协议,就是那个能让LLM像人脑一样,通过“神经突触”连接各种工具、数据和服务的底层框架。它基于JSON-RPC 2.0,轻量、高效,是构建下一代AI自动化系统的核心。一、MCP的四大核...

MCP协议:用JSON-RPC 2.0给LLM装上“神经突触”,Agent开发就靠它了
想让你的AI Agent真正“动起来”,而不是只会聊天?关键在于连接。MCP(Model Context Protocol)协议,就是那个能让LLM像人脑一样,通过“神经突触”连接各种工具、数据和服务的底层框架。它基于JSON-RPC 2.0,轻量、高效,是构建下一代AI自动化系统的核心。
一、MCP的四大核心能力:不止是API调用
很多人把MCP简单理解为“给模型调API的协议”,这太小看它了。它的四大能力,构成了一个完整的Agent执行闭环:
- 工具发现与调用:Agent能动态发现Server提供了哪些工具(比如“发送邮件”、“查询数据库”),并像调用函数一样调用它们。这解决了“模型不知道自己能干什么”的问题。
- 资源访问与订阅:Agent可以读取Server提供的文件、数据库记录等资源,甚至订阅其变化。比如,一个监控Agent可以订阅服务器日志资源,一有错误就自动处理。
- 提示词模板管理:Server可以提供预置的、高质量的提示词模板。Agent调用工具时,自动套用最佳实践模板,极大提升输出质量的稳定性。
- 采样控制与人机协同:这是安全关键。Agent在执行敏感操作(如删除文件、大额支付)前,可以通过MCP协议将决策“采样”回给人类用户确认,实现可控的自动化。
商业价值点:这四大能力意味着,你可以将任何现有软件服务(CRM、ERP、内部工具)快速封装成MCP Server,瞬间让成千上万的AI Agent具备操作该服务的能力,实现服务的“AI化”分发。
二、四层架构:清晰分工,易于扩展
MCP的架构设计非常清晰,分为四层,开发者可以按需介入:
| 层级 | 名称 | 职责 | 类比 |
|---|---|---|---|
| L1 | 传输层 | 负责客户端与服务端之间的原始通信。支持stdio(本地进程通信)和HTTP + SSE(远程流式通信)。 | 神经网络的“轴突”,传递电信号。 |
| L2 | 协议层 | 定义消息格式。核心是JSON-RPC 2.0,规定了请求、响应、通知的结构。 | 神经信号的“编码规则”。 |
| L3 | 能力层 | 实现四大核心能力(工具、资源、提示词、采样)的具体逻辑。 | 不同脑区的“功能分区”(视觉区、语言区)。 |
| L4 | 应用层 | 开发者编写的业务逻辑,比如“查询天气工具”、“数据库连接器”。 | 具体的“思维活动”或“行为”。 |
工程价值:这种分层让调试变得简单。通信问题查L1/L2,工具调用失败查L3/L4,逻辑清晰,不会一团乱麻。
三、实战:开发一个生产级“数据库查询”MCP Server
理论说再多,不如写个例子。我们来构建一个能让AI Agent查询MySQL数据库的MCP Server。这里用Python和官方的mcp库。
1. 安装与基础结构
pip install mcp mysql-connector-python2. 核心代码实现 (db_server.py)
import json
from typing import Any
import mysql.connector
from mcp.server import Server
from mcp.types import Tool, TextContent
import mcp.server.stdio
# 初始化MCP Server,给它一个名字
server = Server("mysql-query-server")
# 定义“工具”:这里是一个执行SQL查询的工具
@server.tool("query_database")
async def query_database(sql: str) -> list[TextContent]:
"""执行只读SQL查询并返回JSON格式结果。"""
try:
# 安全警告:生产环境务必使用参数化查询防止SQL注入!
conn = mysql.connector.connect(

host="localhost",
user="your_user",
password="your_password",
database="your_db"
)
cursor = conn.cursor(dictionary=True)
cursor.execute(sql)
results = cursor.fetchall()
cursor.close()
conn.close()
# 将结果转为JSON字符串返回
return [TextContent(type="text", text=json.dumps(results, ensure_ascii=False))]
except Exception as e:
return [TextContent(type="text", text=f"查询出错: {str(e)}")]
# 定义“资源”:这里提供数据库表结构作为资源,让Agent了解数据库
@server.resource("schema://tables")
async def get_schema() -> str:
"""返回数据库所有表的结构信息。"""
# ... (连接数据库,查询information_schema的代码,此处简化)
schema_info = {"users": ["id", "name", "email"], "orders": ["id", "user_id", "amount"]}
return json.dumps(schema_info)
# 主函数:以stdio模式启动Server
if __name__ == "__main__":
mcp.server.stdio.run_server(server)3. 部署与集成步骤
- 启动Server:在终端运行
python db_server.py。它会通过stdin/stdout等待MCP客户端的连接。 在AI客户端(如Claude Desktop)中配置:在Claude的设置文件中添加:
{ "mcpServers": { "mysql-db": { "command": "python", "args": ["/path/to/db_server.py"] } } }- 测试:现在你可以对Claude说:“帮我查一下用户表里有多少条记录。” Claude会发现
query_database工具,生成SQLSELECT COUNT(*) FROM users,通过MCP调用你的Server,最终把结果告诉你。
可复制路径:将这个Server部署在一台固定服务器上,改用HTTP + SSE传输模式(修改run_server方式),你就能将这个数据库查询能力通过一个URL提供给所有支持MCP的AI客户端和Agent框架,实现一次开发,多处调用。
四、下一步行动清单
- 动手跑通:复制上面的代码,用你自己的一个简单数据库(甚至SQLite)跑通整个流程。感受一下Agent自动调用工具的神奇。
- 封装你的服务:想想你手头有什么重复性工作(比如整理报表、回复固定格式邮件),把它封装成一个MCP Server的工具。
- 探索高级模式:试试“资源订阅”功能。写一个监控本地某个文件变化的Server,让Agent在文件更新时自动触发分析任务。
- 发布与商业化:将你开发的通用MCP Server(比如“PDF解析器”、“社交媒体发布器”)打包,在龙虾官网(yitb.com)或类似平台上作为插件发布。这是将你的技术能力直接产品化、货币化的最短路径。
MCP不是未来,它已经是现在。掌握它,你就掌握了构建下一代智能自动化系统的“连接器”密码。