MCPModel Context Protocol是2024-2025年最火的AI协议之一。这篇文章从原理到实战带你彻底搞懂它。前言如果你关注AI Agent领域你一定听说过MCP。Anthropic在2024年底推出了这个协议短短几个月就被各大AI工具采用——Claude Desktop、Cursor、Kiro、Windsurf等等都支持了MCP。但大多数人对MCP的理解停留在它能让AI调用外部工具。这篇文章我会深入讲解MCP到底解决了什么问题、它的架构设计有多精妙、以及如何自己开发一个MCP Server。一、MCP解决了什么问题没有MCP之前的痛点假设你要让AI Agent能访问GitHub、查数据库、读文件你需要Agent ←自定义适配器→ GitHub API Agent ←自定义适配器→ MySQL Agent ←自定义适配器→ 文件系统 Agent ←自定义适配器→ 邮件服务 ...每接入一个新能力就要写一套定制代码。不同的AI平台ChatGPT Plugin、Claude Tool Use还有不同的接口规范。MCP的解法标准化协议Agent ←统一MCP协议→ GitHub MCP Server → MySQL MCP Server → 文件系统 MCP Server → 邮件 MCP Server → 任何MCP Server...类比MCP之于AI就像USB之于电脑。有了USB标准键盘、鼠标、U盘、打印机都能即插即用。有了MCP标准任何外部能力都能即插即用地给AI使用。二、MCP的核心架构三个角色┌──────────┐ MCP协议 ┌──────────────┐ │ Client │ ←─────────────→ │ Server │ │ (AI应用) │ │ (能力提供者) │ └──────────┘ └──────────────┘ │ │ │ │ Claude GitHub API Cursor 数据库 Kiro 文件系统 自己的应用 任何服务三种核心能力能力说明示例Tools可调用的函数查询数据库、发送邮件Resources可读取的数据文件内容、API响应Prompts预定义的提示模板代码审查模板、翻译模板通信方式1. stdio标准输入输出 - Client启动Server进程 - 通过stdin/stdout通信 - 适合本地工具 2. SSEServer-Sent Events - 基于HTTP - Server可以推送消息 - 适合远程服务 3. HTTPStreamable HTTP - 最新的传输方式 - 支持流式响应 - 生产环境推荐三、开发一个MCP ServerPython我们来做一个实用的MCP Server查询项目的待办事项和Bug列表连接GitHub Issues。3.1 环境准备# 安装MCP SDKpipinstallmcp[cli]# 或者用uv推荐uvaddmcp[cli]3.2 基础Server结构# github_issues_server.pyimportasyncioimporthttpxfrommcp.serverimportServerfrommcp.server.stdioimportstdio_serverfrommcp.typesimport(Tool,TextContent,Resource,ResourceTemplate)# 创建Server实例serverServer(github-issues)# GitHub配置GITHUB_TOKENyour-github-tokenGITHUB_OWNERyour-usernameGITHUB_REPOyour-repoBASE_URLhttps://api.github.comHEADERS{Authorization:ftoken{GITHUB_TOKEN},Accept:application/vnd.github.v3json}3.3 定义Toolsserver.list_tools()asyncdeflist_tools()-list[Tool]:列出所有可用工具return[Tool(namelist_issues,description获取GitHub仓库的Issue列表可按状态和标签筛选,inputSchema{type:object,properties:{state:{type:string,enum:[open,closed,all],description:Issue状态默认open,default:open},labels:{type:string,description:按标签筛选多个标签用逗号分隔如: bug,priority-high},limit:{type:integer,description:返回数量限制默认10,default:10}}}),Tool(namecreate_issue,description创建一个新的GitHub Issue,inputSchema{type:object,properties:{title:{type:string,description:Issue标题},body:{type:string,description:Issue正文内容支持Markdown},labels:{type:array,items:{type:string},description:标签列表}},required:[title]}),Tool(namesearch_issues,description搜索Issue支持关键词和高级搜索语法,inputSchema{type:object,properties:{query:{type:string,description:搜索关键词}},required:[query]})]server.call_tool()asyncdefcall_tool(name:str,arguments:dict)-list[TextContent]:执行工具调用asyncwithhttpx.AsyncClient()asclient:ifnamelist_issues:returnawait_list_issues(client,arguments)elifnamecreate_issue:returnawait_create_issue(client,arguments)elifnamesearch_issues:returnawait_search_issues(client,arguments)else:return[TextContent(typetext,textf未知工具:{name})]asyncdef_list_issues(client:httpx.AsyncClient,args:dict)-list[TextContent]:获取Issue列表params{state:args.get(state,open),per_page:args.get(limit,10),sort:updated,direction:desc}iflabelsinargs:params[labels]args[labels]responseawaitclient.get(f{BASE_URL}/repos/{GITHUB_OWNER}/{GITHUB_REPO}/issues,headersHEADERS,paramsparams)ifresponse.status_code!200:return[TextContent(typetext,textfAPI请求失败:{response.status_code})]issuesresponse.json()# 格式化输出result_lines[f##{GITHUB_OWNER}/{GITHUB_REPO}Issues ({args.get(state,open)})\n]forissueinissues:labels, .join([l[name]forlinissue.get(labels,[])])label_strf [{labels}]iflabelselseresult_lines.append(f- #{issue[number]}{issue[title]}{label_str}\nf 状态:{issue[state]}| 创建者:{issue[user][login]}| f更新:{issue[updated_at][:10]})ifnotissues:result_lines.append(没有找到符合条件的Issue)return[TextContent(typetext,text\n.join(result_lines))]asyncdef_create_issue(client:httpx.AsyncClient,args:dict)-list[TextContent]:创建Issuedata{title:args[title],body:args.get(body,),labels:args.get(labels,[])}responseawaitclient.post(f{BASE_URL}/repos/{GITHUB_OWNER}/{GITHUB_REPO}/issues,headersHEADERS,jsondata)ifresponse.status_code201:issueresponse.json()return[TextContent(typetext,textf✅ Issue创建成功!\nf编号: #{issue[number]}\nf标题:{issue[title]}\nf链接:{issue[html_url]})]else:return[TextContent(typetext,textf❌ 创建失败:{response.text})]asyncdef_search_issues(client:httpx.AsyncClient,args:dict)-list[TextContent]:搜索Issuequeryf{args[query]}repo:{GITHUB_OWNER}/{GITHUB_REPO}responseawaitclient.get(f{BASE_URL}/search/issues,headersHEADERS,params{q:query,per_page:10})ifresponse.status_code!200:return[TextContent(typetext,textf搜索失败:{response.status_code})]dataresponse.json()resultsdata.get(items,[])result_lines[f## 搜索结果: {args[query]} (共{data[total_count]}条)\n]foriteminresults:result_lines.append(f- #{item[number]}{item[title]}({item[state]}))return[TextContent(typetext,text\n.join(result_lines))]3.4 定义Resourcesserver.list_resources()asyncdeflist_resources()-list[Resource]:列出可用资源return[Resource(urifgithub://{GITHUB_OWNER}/{GITHUB_REPO}/readme,name项目README,description项目的README.md文件内容,mimeTypetext/markdown)]server.read_resource()asyncdefread_resource(uri:str)-str:读取资源ifuri.endswith(/readme):asyncwithhttpx.AsyncClient()asclient:responseawaitclient.get(f{BASE_URL}/repos/{GITHUB_OWNER}/{GITHUB_REPO}/readme,headers{**HEADERS,Accept:application/vnd.github.raw})ifresponse.status_code200:returnresponse.textreturnREADME未找到returnf未知资源:{uri}3.5 启动Serverasyncdefmain():启动MCP Serverasyncwithstdio_server()as(read_stream,write_stream):awaitserver.run(read_stream,write_stream,server.create_initialization_options())if__name____main__:asyncio.run(main())四、在AI工具中使用在Kiro/Cursor中配置// .kiro/settings/mcp.json{mcpServers:{github-issues:{command:python,args:[./github_issues_server.py],env:{GITHUB_TOKEN:your-token}}}}配置后AI就能自动调用你的MCP Server来管理GitHub Issues了。使用效果用户帮我看看项目里有哪些bug还没修 AI思考我需要查看GitHub Issues中标记为bug的未关闭Issue AI调用list_issues(stateopen, labelsbug) AI回答项目当前有3个未修复的Bug... 用户把上周五讨论的登录超时问题记录一个Issue AI调用create_issue(title登录页面token过期后未自动跳转, body..., labels[bug, auth]) AI回答✅ 已创建Issue #47...五、进阶实用的MCP Server 想法Server类型功能实用场景数据库查询执行SQL、查看表结构让AI帮你写查询日志分析读取/搜索日志AI帮你排查线上问题K8s管理查看Pod状态、查日志运维助手内部文档搜索Confluence/Notion知识库问答监控面板查询Prometheus指标智能告警分析CI/CD触发Pipeline、查看构建状态部署助手六、开发MCP Server的坑坑1工具描述决定AI是否能正确调用# ❌ 描述不清楚Tool(namequery,description查询数据)# ✅ 描述清楚参数和返回值Tool(namequery_database,description执行SQL查询并返回结果。支持SELECT语句。返回格式为JSON数组每个元素为一行数据。最多返回100行超出会被截断。,inputSchema{...})坑2返回结果太长导致上下文爆炸# ❌ 返回整个表的数据asyncdef_query(sql):resultsdb.execute(sql)returnstr(results)# 可能几MB# ✅ 截断 摘要asyncdef_query(sql):resultsdb.execute(sql)iflen(results)50:summaryf共{len(results)}行显示前50行:\nresultsresults[:50]returnsummaryformat_table(results)坑3错误处理不当导致Server崩溃# ❌ 异常未捕获Server直接crashserver.call_tool()asyncdefcall_tool(name,args):resultawaitrisky_operation()# 如果抛异常Server挂了return[TextContent(typetext,textresult)]# ✅ 始终catch异常返回友好错误server.call_tool()asyncdefcall_tool(name,args):try:resultawaitrisky_operation()return[TextContent(typetext,textresult)]exceptExceptionase:return[TextContent(typetext,textf操作失败:{str(e)})]坑4没有做超时控制# 外部API可能hang住要设超时asyncwithhttpx.AsyncClient(timeout10.0)asclient:responseawaitclient.get(url)七、MCP的未来MCP正在快速演进认证与授权OAuth 2.1支持安全调用远程MCP ServerServer发现类似npm registry搜索和安装MCP ServerElicitationServer主动向用户提问获取信息多模态支持图像、音频等非文本内容可以预见MCP会成为AI Agent生态的基础设施层。现在学习和开发MCP Server就像2015年写RESTful API一样是未来的基本功。写在最后MCP的精妙之处在于它的简单——协议本身非常轻量但它解决了一个巨大的问题让AI Agent的能力可以标准化、可组合、可复用。如果你是后端开发者强烈建议把你日常用到的内部工具封装成MCP Server。这不仅能提升你自己的效率还能让整个团队的AI工具链受益。觉得有帮助的话点个赞有问题评论区讨论下篇写多Agent协作的架构设计~