MCP 协议与 AI Agent 开发实战：从对话到执行，构建多工具协作智能体-平芜编程栈

MCP 协议与 AI Agent 开发实战：从对话到执行，构建多工具协作智能体

引言

2026年，AI 产业的竞争焦点已从"谁的模型更聪明"转向"谁能连接更多真实业务系统"。在这一转变中，由 Anthropic 发起的 MCP（Model Context Protocol，模型上下文协议）已成为事实上的行业标准。OpenAI、Google、华为等头部厂商相继支持 MCP，国内钉钉、飞书、用友等企业级平台也密集发布了原生 MCP Server。本文将深入剖析 MCP 协议的核心架构，并提供可直接运行的实战代码，带你从零构建一个具备多工具协作能力的 AI Agent。

一、为什么传统 Function Calling 在 2026 年"不够用"了？

过去两年，开发者习惯了为每个大模型单独编写 Function Calling 适配层。但在面对企业级复杂场景时，这种"点对点"集成模式暴露出三大致命缺陷：

1.1 N×M 集成地狱

如果有 N 个 AI 应用和 M 个业务系统，就需要开发 N×M 个适配器。每新增一个系统，所有 AI 应用都要重新开发对接代码。一个企业接入 10 个内部系统，光适配代码就可能耗费 2-3 名工程师数月时间。

1.2 上下文割裂

传统工具调用只能返回结构化数据，无法将文件的元信息、数据库的 Schema、API 的语义描述作为富上下文动态注入模型的推理窗口。这意味着模型在调用工具时缺乏对工具能力的完整理解。

1.3 安全与权限失控

工具调用往往直接绑定 API Key，缺乏细粒度的用户级权限代理。Agent 一旦获得权限，便拥有了系统的"上帝视角"，极易引发数据泄露。

二、MCP 核心架构：AI 时代的 USB-C 接口

MCP 采用 Client-Server 架构，将"AI 应用"与"数据/工具源"彻底解耦。AI 应用只需实现一个 MCP Client，即可无缝接入全球任何符合规范的 MCP Server，实现真正的"即插即用"。

2.1 架构概览

[AI Application (Host)] │──> 内置 MCP Client ▼ [MCP Protocol Layer] <──> JSON-RPC 2.0 / stdio / SSE │ ├──> [MCP Server A: 本地文件系统] ──> Resources(文件内容), Tools(读写操作) ├──> [MCP Server B: PostgreSQL] ──> Resources(表结构), Tools(SQL查询) └──> [MCP Server C: 飞书/钉钉] ──> Prompts(消息模板), Tools(发送消息)

2.2 三个核心角色

Host（宿主应用）：运行 AI 模型的应用，如 Claude Desktop、VS Code 插件、自建 ChatBot
MCP Client：嵌入在 Host 中，负责与 MCP Server 建立连接、发送请求
MCP Server：提供具体能力的服务端，暴露 Resources（数据）、Tools（操作）和 Prompts（模板）

2.3 通信协议

MCP 基于 JSON-RPC 2.0 协议，支持两种传输方式：

stdio：通过标准输入输出通信，适合本地工具
SSE（Server-Sent Events）：通过 HTTP 流式通信，适合远程服务

三、实战：构建基于 MCP 的安全数据分析 Agent

以下是一个完整的实战示例，展示如何使用 Python 搭建一个连接本地 SQLite 数据库的 Agent。重点不在于"让 AI 查数据"，而在于展示工具注册、结构化输出校验、以及人机协同确认这三个工程化核心要素。

3.1 环境准备

pipinstallmcp openai pydantic

3.2 定义严格的输出结构

防止模型幻觉的关键是使用 Pydantic 定义严格的输出结构：

frompydanticimportBaseModel,FieldfromtypingimportList,OptionalclassQueryPlan(BaseModel):"""查询计划 - 模型必须先输出计划再执行"""reasoning:str=Field(description="分析用户意图的推理过程")sql_query:str=Field(description="要执行的 SQL 查询语句")risk_level:str=Field(description="风险评估: low/medium/high")requires_approval:bool=Field(description="是否需要人工审批")classQueryResult(BaseModel):"""查询结果"""success:booldata:Optional[List[dict]]=Nonerow_count:int=0error:Optional[str]=None

3.3 构建 MCP Server

importasyncioimportjsonimportsqlite3frommcp.server.fastmcpimportFastMCPfromopenaiimportAsyncOpenAI# 初始化 MCP Servermcp=FastMCP("数据分析Agent")# 初始化 OpenAI 客户端client=AsyncOpenAI()# 数据库连接池DB_PATH="analytics.db"@mcp.tool()asyncdefexecute_sql_query(query:str)->dict:"""执行只读 SQL 查询并返回结果"""# 安全检查：只允许 SELECT 语句query_upper=query.strip().upper()ifnotquery_upper.startswith("SELECT"):return{"error":"仅允许 SELECT 查询","success":False}# 禁止危险操作dangerous_keywords=["DROP","DELETE","UPDATE","INSERT","ALTER","CREATE"]forkeywordindangerous_keywords:ifkeywordinquery_upper:return{"error":f"不允许{keyword}操作","success":False}try:conn=sqlite3.connect(DB_PATH)conn.row_factory=sqlite3.Row cursor=conn.cursor()cursor.execute(query)rows=[dict(row)forrowincursor.fetchall()]conn.close()return{"success":True,"data":rows,"row_count":len(rows)}exceptExceptionase:return{"success":False,"error":str(e)}@mcp.tool()asyncdefget_table_schema(table_name:str)->dict:"""获取指定表的 Schema 信息"""conn=sqlite3.connect(DB_PATH)cursor=conn.cursor()cursor.execute(f"PRAGMA table_info({table_name})")columns=[dict(row)forrowincursor.fetchall()]conn.close()return{"table":table_name,"columns":columns}@mcp.resource("db://schema")asyncdefget_all_schemas()->str:"""获取所有表的 Schema（作为资源暴露）"""conn=sqlite3.connect(DB_PATH)cursor=conn.cursor()cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")tables=[row[0]forrowincursor.fetchall()]schemas=[]fortableintables:cursor.execute(f"PRAGMA table_info({table})")columns=cursor.fetchall()col_desc=", ".join([f"{c[1]}({c[2]})"forcincolumns])schemas.append(f"表{table}:{col_desc}")conn.close()return"\n".join(schemas)

3.4 构建 Agent 主循环

importjsonfromtypingimportList,Dict,AnyclassDataAnalysisAgent:def__init__(self):self.conversation_history:List[Dict[str,Any]]=[]self.approval_queue:List[Dict]=[]asyncdefprocess_query(self,user_message:str)->str:"""处理用户查询的主入口"""# 1. 获取可用工具和资源tools=awaitself._get_available_tools()schema=awaitself._get_database_schema()# 2. 构建系统提示词system_prompt=f"""你是一个数据分析助手。你可以使用以下工具：{tools}当前数据库 Schema：{schema}重要规则： 1. 在执行任何 SQL 查询前，先输出查询计划 2. 查询计划必须包含风险评估 3. 高风险查询需要用户确认后才能执行 4. 只允许 SELECT 查询 5. 结果以表格或自然语言呈现"""# 3. 调用模型进行规划messages=[{"role":"system","content":system_prompt},{"role":"user","content":user_message}]response=awaitclient.chat.completions.create(model="gpt-4o",messages=messages,tools=tools,tool_choice="auto")# 4. 处理工具调用returnawaitself._handle_response(response)asyncdef_handle_response(self,response)->str:"""处理模型响应，执行工具调用"""message=response.choices[0].messageifmessage.tool_calls:results=[]fortool_callinmessage.tool_calls:tool_name=tool_call.function.name tool_args=json.loads(tool_call.function.arguments)# 执行工具iftool_name=="execute_sql_query":result=awaitexecute_sql_query(tool_args["query"])eliftool_name=="get_table_schema":result=awaitget_table_schema(tool_args["table_name"])else:result={"error":f"未知工具:{tool_name}"}results.append({"tool":tool_name,"result":result})# 将结果反馈给模型生成最终回复returnawaitself._generate_final_response(results)returnmessage.contentor"无法处理该请求"asyncdef_generate_final_response(self,tool_results:List[dict])->str:"""基于工具执行结果生成最终回复"""context=json.dumps(tool_results,ensure_ascii=False,indent=2)response=awaitclient.chat.completions.create(model="gpt-4o",messages=[{"role":"system","content":"基于以下工具执行结果，用自然语言回答用户问题。"},{"role":"user","content":f"工具执行结果：\n{context}"}])returnresponse.choices[0].message.content

3.5 运行 Agent

asyncdefmain():agent=DataAnalysisAgent()# 示例查询queries=["上个月销售额最高的10个产品是什么？","统计各地区的用户活跃度分布","分析用户留存率的变化趋势",]forqueryinqueries:print(f"\n用户:{query}")result=awaitagent.process_query(query)print(f"Agent:{result}")if__name__=="__main__":asyncio.run(main())

四、MCP 协议的高级特性

4.1 动态工具发现

MCP Server 可以在运行时动态注册和注销工具，无需重启 Agent：

@mcp.tool()asyncdefregister_new_tool(name:str,description:str,parameters:dict):"""动态注册新工具"""# 实现动态工具注册逻辑pass

4.2 资源订阅

MCP 支持资源变更通知，当数据库 Schema 或文件内容发生变化时，Server 可以主动通知 Client：

@mcp.resource("db://changes")asyncdefwatch_schema_changes():"""监听数据库 Schema 变更"""# 实现变更监听逻辑pass

4.3 权限代理

MCP 支持用户级权限代理，Agent 以当前用户的身份访问资源，而不是使用服务账号的全局权限：

@mcp.tool()asyncdefaccess_user_data(user_id:str,resource:str):"""以指定用户身份访问资源"""# 验证用户权限# 以用户身份执行操作pass

五、MCP 生态现状与未来

截至2026年7月，MCP 生态已经相当丰富：

官方 SDK：Python、TypeScript、Java、Go 等主流语言都有官方 MCP SDK
预构建 Server：文件系统、数据库（PostgreSQL、MySQL、SQLite）、搜索引擎、云服务（AWS、GCP、Azure）等都有现成的 MCP Server
平台集成：Claude Desktop、VS Code、Cursor、Windsurf 等主流 AI 工具都已支持 MCP
企业级 Server：钉钉、飞书、用友、Salesforce、SAP 等企业平台已发布原生 MCP Server

MCP 正在成为 AI 世界的"HTTP 协议"——它不关心你用什么模型、什么工具，只定义了一套标准的通信方式。这种标准化带来的网络效应，将加速 AI Agent 从"单兵作战"迈向"万物互联"的新纪元。