AI Agent实战一:MCP协议从入门到实践
2026/5/4 15:02:28 网站建设 项目流程

AI辅助创作 | 专栏《2026 AI编程效率革命》第07篇

前言

MCP(Model Context Protocol)是Anthropic在2024年底推出的开放协议,旨在标准化AI模型与外部工具、数据源的交互方式。到2026年,MCP已经成为AI Agent开发的事实标准协议。本文将从零开始,带你理解MCP协议的核心概念,并通过实战代码搭建一个完整的MCP服务。


一、MCP协议是什么?

1.1 为什么需要MCP?

在MCP出现之前,AI工具集成面临以下问题:

  • 每个AI应用都要为每个工具写单独的集成代码
  • 工具接口没有统一标准,重复造轮子
  • 模型切换时工具链需要重写

MCP的核心理念:像USB-C统一接口一样,统一AI与工具的连接方式

1.2 MCP架构

┌─────────────┐ MCP协议 ┌─────────────┐ │ AI应用 │ ◄──────────────► │ MCP Server │ │ (Host) │ │ (工具提供方) │ │ 如: Cursor │ │ │ └─────────────┘ └──────┬───────┘ │ ┌──────┴───────┐ │ 外部资源 │ │ 数据库/API │ │ 文件系统等 │ └──────────────┘

MCP定义了三种核心能力:

能力说明示例
Tools可调用的函数查询数据库、发送邮件
Resources可读取的数据文件内容、数据库记录
Prompts预定义提示词代码审查模板

二、MCP开发环境搭建

2.1 安装MCP SDK

# 安装MCP Python SDKuvaddmcp# 安装官方示例依赖uvaddhttpx pydantic

2.2 项目结构

mcp-demo/ ├── pyproject.toml ├── .env ├── server/ │ ├── __init__.py │ ├── weather_server.py # 天气查询MCP服务 │ ├── file_server.py # 文件管理MCP服务 │ └── db_server.py # 数据库查询MCP服务 ├── client/ │ ├── __init__.py │ └── mcp_client.py # MCP客户端 └── tests/ └── test_server.py

三、实战:创建你的第一个MCP Server

3.1 天气查询MCP服务

# server/weather_server.pyimporthttpxfrommcp.serverimportServerfrommcp.typesimportTool,TextContent# 创建MCP Server实例server=Server("weather-service")@server.list_tools()asyncdeflist_tools()->list[Tool]:"""声明服务提供的工具列表"""return[Tool(name="get_weather",description="获取指定城市的当前天气信息",inputSchema={"type":"object","properties":{"city":{"type":"string","description":"城市名称,如'北京'、'上海'"},"unit":{"type":"string","enum":["celsius","fahrenheit"],"description":"温度单位,默认摄氏度","default":"celsius"}},"required":["city"]}),Tool(name="get_forecast",description="获取未来3天天气预报",inputSchema={"type":"object","properties":{"city":{"type":"string","description":"城市名称"}},"required":["city"]})]@server.call_tool()asyncdefcall_tool(name:str,arguments:dict)->list[TextContent]:"""处理工具调用请求"""ifname=="get_weather":returnawaitget_weather(arguments["city"],arguments.get("unit","celsius"))elifname=="get_forecast":returnawaitget_forecast(arguments["city"])else:raiseValueError(f"未知工具:{name}")asyncdefget_weather(city:str,unit:str="celsius")->list[TextContent]:"""查询天气(使用模拟数据)"""# 实际项目中调用天气APIweather_data={"北京":{"temp":22,"humidity":45,"condition":"晴"},"上海":{"temp":25,"humidity":72,"condition":"多云"},"深圳":{"temp":30,"humidity":80,"condition":"阵雨"},}data=weather_data.get(city,{"temp":20,"humidity":50,"condition":"未知"})ifunit=="fahrenheit":data["temp"]=data["temp"]*9/5+32result=(f"📍{city}当前天气\n"f"🌡️ 温度:{data['temp']}{'°C'ifunit=='celsius'else'°F'}\n"f"💧 湿度:{data['humidity']}%\n"f"🌤️ 状况:{data['condition']}")return[TextContent(type="text",text=result)]asyncdefget_forecast(city:str)->list[TextContent]:"""获取天气预报"""forecast=(f"📍{city}未来3天预报\n"f"明天: 晴,18-26°C\n"f"后天: 多云,17-24°C\n"f"大后天: 小雨,15-22°C")return[TextContent(type="text",text=forecast)]# 启动服务if__name__=="__main__":importasynciofrommcp.server.stdioimportstdio_serverasyncdefmain():asyncwithstdio_server()as(read_stream,write_stream):awaitserver.run(read_stream,write_stream)asyncio.run(main())

3.2 文件管理MCP服务

# server/file_server.pyimportosimportjsonfrompathlibimportPathfrommcp.serverimportServerfrommcp.typesimportTool,TextContent server=Server("file-manager")WORKSPACE=Path("./workspace")@server.list_tools()asyncdeflist_tools()->list[Tool]:return[Tool(name="list_files",description="列出指定目录下的文件",inputSchema={"type":"object","properties":{"path":{"type":"string","description":"目录路径(相对于workspace)"}},"required":["path"]}),Tool(name="read_file",description="读取文件内容",inputSchema={"type":"object","properties":{"path":{"type":"string","description":"文件路径"}},"required":["path"]}),Tool(name="write_file",description="写入文件",inputSchema={"type":"object","properties":{"path":{"type":"string"},"content":{"type":"string"}},"required":["path","content"]}),Tool(name="search_files",description="按名称搜索文件",inputSchema={"type":"object","properties":{"pattern":{"type":"string","description":"搜索关键词"}},"required":["pattern"]})]@server.call_tool()asyncdefcall_tool(name:str,arguments:dict)->list[TextContent]:ifname=="list_files":target=WORKSPACE/arguments["path"]files=[str(f.relative_to(WORKSPACE))forfintarget.rglob("*")iff.is_file()]return[TextContent(type="text",text=json.dumps(files[:50],indent=2))]elifname=="read_file":target=WORKSPACE/arguments["path"]content=target.read_text(encoding="utf-8")return[TextContent(type="text",text=content)]elifname=="write_file":target=WORKSPACE/arguments["path"]target.parent.mkdir(parents=True,exist_ok=True)target.write_text(arguments["content"],encoding="utf-8")return[TextContent(type="text",text=f"已写入:{arguments['path']}")]elifname=="search_files":results=[str(f.relative_to(WORKSPACE))forfinWORKSPACE.rglob(f"*{arguments['pattern']}*")iff.is_file()]return[TextContent(type="text",text=json.dumps(results[:20],indent=2))]raiseValueError(f"未知工具:{name}")if__name__=="__main__":importasynciofrommcp.server.stdioimportstdio_serverasyncdefmain():asyncwithstdio_server()as(read_stream,write_stream):awaitserver.run(read_stream,write_stream)asyncio.run(main())

四、MCP客户端开发

4.1 连接MCP Server

# client/mcp_client.pyimportasynciofrommcpimportClientSession,StdioServerParametersfrommcp.client.stdioimportstdio_clientclassMCPClient:def__init__(self):self.session:ClientSession|None=Noneself.available_tools:list=[]asyncdefconnect(self,server_command:list[str]):"""连接到MCP Server"""server_params=StdioServerParameters(command=server_command[0],args=server_command[1:],)self.stdio_context=stdio_client(server_params)read_stream,write_stream=awaitself.stdio_context.__aenter__()self.session=ClientSession(read_stream,write_stream)awaitself.session.__aenter__()awaitself.session.initialize()# 获取可用工具列表result=awaitself.session.list_tools()self.available_tools=result.toolsprint(f"已连接,可用工具:{[t.namefortinself.available_tools]}")asyncdefcall_tool(self,tool_name:str,arguments:dict):"""调用工具"""ifnotself.session:raiseRuntimeError("未连接到MCP Server")result=awaitself.session.call_tool(tool_name,arguments)returnresult.contentasyncdefdisconnect(self):"""断开连接"""ifself.session:awaitself.session.__aexit__(None,None,None)awaitself.stdio_context.__aexit__(None,None,None)asyncdefmain():client=MCPClient()# 连接天气服务awaitclient.connect(["python","server/weather_server.py"])# 调用工具result=awaitclient.call_tool("get_weather",{"city":"北京"})print(result[0].text)result=awaitclient.call_tool("get_forecast",{"city":"上海"})print(result[0].text)awaitclient.disconnect()if__name__=="__main__":asyncio.run(main())

五、与AI模型集成

5.1 将MCP工具接入LangChain Agent

# integration/langchain_mcp_agent.pyimportasyncioimportjsonfromlangchain_openaiimportChatOpenAIfromlangchain.agentsimportcreate_tool_calling_agent,AgentExecutorfromlangchain_core.toolsimporttoolfromlangchain_core.promptsimportChatPromptTemplatefromclient.mcp_clientimportMCPClient# 创建MCP客户端mcp_client=MCPClient()@toolasyncdefquery_weather(city:str)->str:"""查询指定城市的天气信息"""result=awaitmcp_client.call_tool("get_weather",{"city":city})returnresult[0].text@toolasyncdefget_forecast(city:str)->str:"""获取城市未来3天天气预报"""result=awaitmcp_client.call_tool("get_forecast",{"city":city})returnresult[0].textasyncdefrun_agent():# 连接MCP服务awaitmcp_client.connect(["python","server/weather_server.py"])# 创建LangChain Agentllm=ChatOpenAI(model="gpt-5.5",temperature=0)tools=[query_weather,get_forecast]prompt=ChatPromptTemplate.from_messages([("system","你是一个天气助手,可以查询天气和预报。用中文回复。"),("human","{input}"),("placeholder","{agent_scratchpad}"),])agent=create_tool_calling_agent(llm,tools,prompt)executor=AgentExecutor(agent=agent,tools=tools,verbose=True)# 测试result=awaitexecutor.ainvoke({"input":"北京和上海今天天气怎么样?明天呢?"})print(result["output"])awaitmcp_client.disconnect()if__name__=="__main__":asyncio.run(run_agent())

六、在Cursor中使用MCP

Cursor原生支持MCP协议,配置方法:

// .cursor/mcp.json{"mcpServers":{"weather":{"command":"python","args":["server/weather_server.py"]},"file-manager":{"command":"python","args":["server/file_server.py"]}}}

配置完成后,在Cursor的Agent模式中就可以直接调用这些MCP工具了。


七、MCP生态与社区资源

2026年MCP生态已经相当成熟:

  • MCP Hub:官方工具市场,数百个现成MCP Server
  • 常用MCP Server:数据库(PostgreSQL/MySQL)、GitHub、Slack、Google Drive等
  • 多语言SDK:Python、TypeScript、Java、Go

总结

MCP协议通过标准化AI与工具的交互方式,解决了重复集成的问题。本文从协议原理讲到实战代码,覆盖了:

  1. MCP的核心概念(Tools/Resources/Prompts)
  2. MCP Server开发实战
  3. MCP客户端开发
  4. 与LangChain Agent集成
  5. 在Cursor中使用MCP

下一篇我们将深入多Agent协作系统的搭建,利用MCP协议构建更复杂的AI应用。


免责声明:本文为AI辅助创作内容,代码示例仅供学习参考。MCP协议规范可能随版本更新而变化,请以官方文档为准。实际项目中请注意API Key安全和数据隐私保护。

专栏:《2026 AI编程效率革命》| 第07篇
发布日期:2026-05-03

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询