当前位置：首页 > news >正文

AI Agent实战一：MCP协议从入门到实践

news 2026/5/4 15:14:25

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"]}}}