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

AI辅助创作 | 专栏《2026 AI编程效率革命》第07篇前言MCPModel 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 pydantic2.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 Server3.1 天气查询MCP服务# server/weather_server.pyimporthttpxfrommcp.serverimportServerfrommcp.typesimportTool,TextContent# 创建MCP Server实例serverServer(weather-service)server.list_tools()asyncdeflist_tools()-list[Tool]:声明服务提供的工具列表return[Tool(nameget_weather,description获取指定城市的当前天气信息,inputSchema{type:object,properties:{city:{type:string,description:城市名称如北京、上海},unit:{type:string,enum:[celsius,fahrenheit],description:温度单位默认摄氏度,default:celsius}},required:[city]}),Tool(nameget_forecast,description获取未来3天天气预报,inputSchema{type:object,properties:{city:{type:string,description:城市名称}},required:[city]})]server.call_tool()asyncdefcall_tool(name:str,arguments:dict)-list[TextContent]:处理工具调用请求ifnameget_weather:returnawaitget_weather(arguments[city],arguments.get(unit,celsius))elifnameget_forecast:returnawaitget_forecast(arguments[city])else:raiseValueError(f未知工具:{name})asyncdefget_weather(city:str,unit:strcelsius)-list[TextContent]:查询天气使用模拟数据# 实际项目中调用天气APIweather_data{北京:{temp:22,humidity:45,condition:晴},上海:{temp:25,humidity:72,condition:多云},深圳:{temp:30,humidity:80,condition:阵雨},}dataweather_data.get(city,{temp:20,humidity:50,condition:未知})ifunitfahrenheit:data[temp]data[temp]*9/532result(f{city}当前天气\nf️ 温度:{data[temp]}{°Cifunitcelsiuselse°F}\nf 湿度:{data[humidity]}%\nf️ 状况:{data[condition]})return[TextContent(typetext,textresult)]asyncdefget_forecast(city:str)-list[TextContent]:获取天气预报forecast(f{city}未来3天预报\nf明天: 晴18-26°C\nf后天: 多云17-24°C\nf大后天: 小雨15-22°C)return[TextContent(typetext,textforecast)]# 启动服务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 serverServer(file-manager)WORKSPACEPath(./workspace)server.list_tools()asyncdeflist_tools()-list[Tool]:return[Tool(namelist_files,description列出指定目录下的文件,inputSchema{type:object,properties:{path:{type:string,description:目录路径相对于workspace}},required:[path]}),Tool(nameread_file,description读取文件内容,inputSchema{type:object,properties:{path:{type:string,description:文件路径}},required:[path]}),Tool(namewrite_file,description写入文件,inputSchema{type:object,properties:{path:{type:string},content:{type:string}},required:[path,content]}),Tool(namesearch_files,description按名称搜索文件,inputSchema{type:object,properties:{pattern:{type:string,description:搜索关键词}},required:[pattern]})]server.call_tool()asyncdefcall_tool(name:str,arguments:dict)-list[TextContent]:ifnamelist_files:targetWORKSPACE/arguments[path]files[str(f.relative_to(WORKSPACE))forfintarget.rglob(*)iff.is_file()]return[TextContent(typetext,textjson.dumps(files[:50],indent2))]elifnameread_file:targetWORKSPACE/arguments[path]contenttarget.read_text(encodingutf-8)return[TextContent(typetext,textcontent)]elifnamewrite_file:targetWORKSPACE/arguments[path]target.parent.mkdir(parentsTrue,exist_okTrue)target.write_text(arguments[content],encodingutf-8)return[TextContent(typetext,textf已写入:{arguments[path]})]elifnamesearch_files:results[str(f.relative_to(WORKSPACE))forfinWORKSPACE.rglob(f*{arguments[pattern]}*)iff.is_file()]return[TextContent(typetext,textjson.dumps(results[:20],indent2))]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|NoneNoneself.available_tools:list[]asyncdefconnect(self,server_command:list[str]):连接到MCP Serverserver_paramsStdioServerParameters(commandserver_command[0],argsserver_command[1:],)self.stdio_contextstdio_client(server_params)read_stream,write_streamawaitself.stdio_context.__aenter__()self.sessionClientSession(read_stream,write_stream)awaitself.session.__aenter__()awaitself.session.initialize()# 获取可用工具列表resultawaitself.session.list_tools()self.available_toolsresult.toolsprint(f已连接可用工具:{[t.namefortinself.available_tools]})asyncdefcall_tool(self,tool_name:str,arguments:dict):调用工具ifnotself.session:raiseRuntimeError(未连接到MCP Server)resultawaitself.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():clientMCPClient()# 连接天气服务awaitclient.connect([python,server/weather_server.py])# 调用工具resultawaitclient.call_tool(get_weather,{city:北京})print(result[0].text)resultawaitclient.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_clientMCPClient()toolasyncdefquery_weather(city:str)-str:查询指定城市的天气信息resultawaitmcp_client.call_tool(get_weather,{city:city})returnresult[0].texttoolasyncdefget_forecast(city:str)-str:获取城市未来3天天气预报resultawaitmcp_client.call_tool(get_forecast,{city:city})returnresult[0].textasyncdefrun_agent():# 连接MCP服务awaitmcp_client.connect([python,server/weather_server.py])# 创建LangChain AgentllmChatOpenAI(modelgpt-5.5,temperature0)tools[query_weather,get_forecast]promptChatPromptTemplate.from_messages([(system,你是一个天气助手可以查询天气和预报。用中文回复。),(human,{input}),(placeholder,{agent_scratchpad}),])agentcreate_tool_calling_agent(llm,tools,prompt)executorAgentExecutor(agentagent,toolstools,verboseTrue)# 测试resultawaitexecutor.ainvoke({input:北京和上海今天天气怎么样明天呢})print(result[output])awaitmcp_client.disconnect()if__name____main__:asyncio.run(run_agent())六、在Cursor中使用MCPCursor原生支持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等多语言SDKPython、TypeScript、Java、Go总结MCP协议通过标准化AI与工具的交互方式解决了重复集成的问题。本文从协议原理讲到实战代码覆盖了MCP的核心概念Tools/Resources/PromptsMCP Server开发实战MCP客户端开发与LangChain Agent集成在Cursor中使用MCP下一篇我们将深入多Agent协作系统的搭建利用MCP协议构建更复杂的AI应用。免责声明本文为AI辅助创作内容代码示例仅供学习参考。MCP协议规范可能随版本更新而变化请以官方文档为准。实际项目中请注意API Key安全和数据隐私保护。专栏《2026 AI编程效率革命》| 第07篇发布日期2026-05-03