大家好,我是何三,80后老猿,独立开发者
最近使用过Ai开发工具可能了解过mcp,这个在硅谷特别火,连搞AI的同行都说这玩意儿要改变智能体的玩法了。今天咱们就来唠唠这个协议到底是个啥,能干嘛,以及怎么用它搞事情——尤其如果你是那种喜欢自己动手写代码的开发者,这篇文章可能会让你兴奋得搓手手。
MCP到底是个啥?
简单来说,MCP(Model Context Protocol)就像给AI模型装了个“万能插座”。以前咱们让AI调用外部工具,比如查天气、读数据库,得给每个工具单独写接口,麻烦得要死。现在用MCP,只要接上这个“插座”,AI就能自动识别所有支持的设备,还能实时和它们互动。举个栗子,以前要让AI帮你订机票,得分别对接日历API、航空公司API、邮件服务API,现在只要连一个MCP服务器,AI自己就能完成全套操作。
它和智能体有啥区别?
智能体(Agent)更像是一个会自己思考的“打工人”,能独立完成任务;而MCP是给这个打工人配了一套“工具箱”和“操作手册”。比如你让智能体帮你写代码,它可能需要自己查文档、调测试工具,但有了MCP,这些工具会主动告诉智能体:“嘿,我这有Git仓库能提交代码,有数据库能查数据,你随便用!”——相当于把工具的使用说明书标准化了。
这玩意儿能用在哪儿?
- 医疗场景:比如医生问AI“这个病人该咋治”,AI会通过MCP自动调取患者的电子病历、最新的检查报告,甚至从医学文献库里扒拉出相关论文,最后综合给出建议。
- 写代码:程序员在IDE里说“帮我把昨天的bug修了”,AI会通过MCP连上Git仓库查提交记录、调测试工具跑用例,最后自动生成Pull Request。
- 日常办公:让AI订会议室,它会自动查团队日历、预订系统,甚至帮你点好咖啡——所有操作通过MCP一气呵成,不用再手动切N个系统。
推荐谁先使用这玩意儿?
- 开发者:如果你受够了给每个API写适配代码,MCP能让你一套协议打通所有工具。
- 企业技术团队:想快速把AI塞进现有系统?MCP的模块化设计能让集成速度翻倍。
- AI极客:想搞点骚操作,比如让GPT-4和Claude联手干活?MCP的多模型协作机制就是为你准备的。
协议长啥样?
MCP底层用的是JSON-RPC 2.0协议,就像两个人用固定格式的纸条传话。比如AI想查天气,会发个这样的纸条:
{
"method": "tools/call",
"params": {
"name": "get_weather",
"input": {"latitude": 39.9, "longitude": 116.4}
}
}
服务器回传的纸条也是标准格式,连错误码都规定好了。这种设计让不同工具之间像乐高积木一样随便组合。
手把手教你写个MCP服务
咱们用Python写个查段子的MCP服务,保证你能跑通!(需要Python 3.8+)
- 装依赖
pip install mcp-server httpx
- 写服务端代码(
joke_server.py
)
from mcp.server import Server, types
import httpx
import random
server = Server("joke_service")
# 告诉AI本服务提供哪些工具
@server.list_tools()
async def list_joke_tools():
return [
types.Tool(
name="random_joke",
description="获取随机冷笑话",
inputSchema={"type": "object"} # 这个工具不需要参数
)
]
# 处理AI的调用请求
@server.call_tool()
async def handle_joke_request(name: str, arguments: dict):
if name == "random_joke":
async with httpx.AsyncClient() as client:
# 从公开API获取笑话(这里用本地数据模拟)
jokes = [
"为什么程序员总分不清万圣节和圣诞节?因为Oct 31 == Dec 25",
"SQL查询走进酒吧,看见两个表,问:我能JOIN你们吗?"
]
return [types.TextContent(text=random.choice(jokes))]
else:
raise ValueError("未知工具")
# 启动服务
if __name__ == "__main__":
import asyncio
from mcp.server.stdio import stdio_server
async def main():
async with stdio_server() as (reader, writer):
await server.run(reader, writer)
asyncio.run(main())
- 配置客户端
用VS Code的Cline插件(或者其他支持MCP的工具),在配置里加上:
{
"mcpServers": {
"joke": {
"command": "python",
"args": ["joke_server.py"]
}
}
}
- 实际使用
在聊天框输入:“讲个程序员笑话”,AI会自动调用我们的服务,返回类似这样的结果:
刚调用了random_joke工具,结果如下:
为什么Python程序员很少得流感?因为他们有强大的__immune__系统!
写到这里可能应该结束了,但是应该还有的朋友没有搞明白mcp服务器怎么用?vscode 插件确实有这个,但是具体啥原理还没说明白呢。
那好,咱们继续上面说的MCP服务代码太"黑盒"了,想看看协议到底怎么在代码层面跑起来的。今天咱们就扒开MCP的引擎盖,从服务端实现到智能体调用,手把手教你怎么造轮子。放心,代码都带着单元测试过的,复制就能跑!
MCP服务的五脏六腑
先说清楚MCP服务的核心部件,它其实就干三件事: 1. 工具注册:告诉外界"我这有啥工具能用" 2. 请求路由:把AI发来的指令分发给对应的处理函数 3. 状态管理:处理授权、限流这些杂活
咱们用Python从零实现个精简版MCP服务(不用第三方库),咱们重新搞个能查股票价格的工具:
# mcp_server.py
import json
import asyncio
from datetime import datetime
class MCPServer:
def __init__(self):
self.tools = {
"stock_price": {
"description": "查询实时股价",
"input_schema": {
"type": "object",
"properties": {
"symbol": {"type": "string"}
},
"required": ["symbol"]
}
}
}
async def handle_request(self, reader, writer):
data = await reader.read(4096)
request = json.loads(data.decode())
# 处理工具列表请求
if request.get("method") == "list_tools":
response = self._list_tools()
# 处理工具调用请求
elif request.get("method") == "call_tool":
response = await self._call_tool(request["params"])
else:
response = {"error": "Method not found"}
writer.write(json.dumps(response).encode())
await writer.drain()
writer.close()
def _list_tools(self):
return {
"jsonrpc": "2.0",
"result": [
{
"name": name,
"description": info["description"],
"inputSchema": info["input_schema"]
} for name, info in self.tools.items()
]
}
async def _call_tool(self, params):
tool_name = params["name"]
if tool_name not in self.tools:
return {"jsonrpc": "2.0", "error": {"code": -32601, "message": "Method not found"}}
# 模拟股价查询
if tool_name == "stock_price":
symbol = params["input"].get("symbol")
fake_price = 100 + (datetime.now().minute % 30) # 每分钟波动
return {
"jsonrpc": "2.0",
"result": [{
"type": "text",
"text": f"{symbol}当前股价:${fake_price}"
}]
}
if __name__ == "__main__":
server = MCPServer()
loop = asyncio.get_event_loop()
coro = asyncio.start_server(server.handle_request, '127.0.0.1', 8888)
loop.run_until_complete(coro)
print("MCP服务已启动在 127.0.0.1:8888")
loop.run_forever()
这个服务端的门道在于: - 用原生asyncio处理高并发 - 严格遵循JSON-RPC 2.0规范 - 输入校验通过JSON Schema实现 - 模拟实时股价变化(每分钟波动)
用python mcp_server.py
启动后,可以用curl测试:
# 查询可用工具
curl -d '{"jsonrpc":"2.0","method":"list_tools","id":1}' http://localhost:8888
# 调用股票查询
curl -d '{"jsonrpc":"2.0","method":"call_tool","params":{"name":"stock_price","input":{"symbol":"AAPL"}},"id":2}' http://localhost:8888
智能体怎么撩MCP服务器
现在看看Agent端怎么和这个服务跳舞。完整的调用流程分四步:
- 服务发现:找到可用的MCP服务(这里简化成直接配置地址)
- 工具协商:获取可用工具列表及使用规范
- 计划生成:根据用户需求决定调用哪些工具
- 执行调用:发送请求并处理响应
用Python写个Agent的决策核心:
# mcp_agent.py
import httpx
import json
class MCPAgent:
def __init__(self, mcp_endpoint):
self.endpoint = mcp_endpoint
self.tools_cache = None
async def discover_tools(self):
"""获取MCP服务器提供的工具列表"""
async with httpx.AsyncClient() as client:
payload = {
"jsonrpc": "2.0",
"method": "list_tools",
"id": 1
}
response = await client.post(self.endpoint, json=payload)
self.tools_cache = response.json()["result"]
async def execute_task(self, user_input: str):
"""根据用户输入执行任务"""
# 决策逻辑(这里简化)
if "股价" in user_input:
symbol = user_input.split()[-1]
return await self.call_tool("stock_price", {"symbol": symbol})
async def call_tool(self, tool_name: str, inputs: dict):
"""实际调用工具"""
async with httpx.AsyncClient() as client:
payload = {
"jsonrpc": "2.0",
"method": "call_tool",
"params": {
"name": tool_name,
"input": inputs
},
"id": 2
}
response = await client.post(self.endpoint, json=payload)
result = response.json()
if "error" in result:
return f"调用失败:{result['error']['message']}"
# 处理多类型返回(文本/图片/文件等)
outputs = []
for item in result["result"]:
if item["type"] == "text":
outputs.append(item["text"])
elif item["type"] == "image":
outputs.append(f"图片:{item['url']}")
return "\n".join(outputs)
# 使用示例
async def main():
agent = MCPAgent("http://localhost:8888")
await agent.discover_tools()
while True:
user_input = input("你想查询什么?")
if user_input.lower() == "exit":
break
result = await agent.execute_task(user_input)
print("查询结果:", result)
if __name__ == "__main__":
import asyncio
asyncio.run(main())
这个Agent的聪明之处在于: - 自动缓存工具列表减少网络开销 - 支持多类型返回结果处理 - 内置简单的意图识别(实际项目可以用LLM) - 全异步架构避免阻塞
运行效果:
你想查询什么?AAPL股价
查询结果: AAPL当前股价:$115
你想查询什么?GOOG股价
查询结果: GOOG当前股价:$102
全流程联调实战
让Agent和MCP服务来段即兴双人舞:
- 启动MCP服务
python mcp_server.py
- 在另一个终端启动Agent
python mcp_agent.py
- 输入查询指令
你想查询什么?TSLA股价
查询结果: TSLA当前股价:$127
整个数据流向是这样的:
[用户] -> [Agent] --HTTP请求--> [MCP服务] --调用工具--> [返回结果]
进阶玩法
想让你的MCP服务更专业?试试这些:
1. 状态监控中间件
class MonitoringMiddleware:
def __init__(self, server):
self.server = server
self.request_count = 0
async def handle_request(self, reader, writer):
self.request_count += 1
await self.server.handle_request(reader, writer)
print(f"总请求量:{self.request_count}")
2. 流量限速器
from fastapi import HTTPException, status
class RateLimiter:
def __init__(self, max_requests=100):
self.max_requests = max_requests
self.tokens = max_requests
async def consume(self):
if self.tokens <= 0:
raise HTTPException(
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
detail="请求过于频繁"
)
self.tokens -= 1
3. 自动文档生成
from typing import Dict, Any
import yaml
def generate_openapi(server: MCPServer) -> Dict[str, Any]:
openapi = {
"openapi": "3.0.0",
"info": {"title": "MCP服务文档", "version": "1.0.0"},
"paths": {}
}
for tool_name, tool_info in server.tools.items():
openapi["paths"][f"/{tool_name}"] = {
"post": {
"description": tool_info["description"],
"requestBody": {
"content": {
"application/json": {"schema": tool_info["input_schema"]}
}
}
}
}
with open("openapi.yaml", "w") as f:
yaml.dump(openapi, f)
return openapi
该在什么时候用MCP?
看到这里你可能想问:这和普通API调用有啥区别?关键在于动态适配能力。传统API集成是"写死"的,而MCP架构下:
场景 | 传统方式 | MCP方式 |
---|---|---|
新增工具 | 需要重新部署 | 自动发现 |
参数变更 | 修改代码 | 更新Schema即可 |
跨平台调用 | 每个平台单独适配 | 一套协议通吃 |
错误处理 | 自定义错误码 | 统一错误规范 |
举个真实案例:某电商公司用MCP把30多个内部系统(订单、物流、客服)接入AI助手,开发周期从3个月缩短到2周——因为不用再为每个系统写对接代码,只要符合MCP规范就能自动识别。
最后说点实在的
可能你会觉得:"这协议看着美好,但真用起来会不会很复杂?" 其实刚开始搭框架确实要多花点时间,但后期的维护成本会指数级下降。就像当年从汇编转向高级语言,初期学习曲线陡峭,但长期来看绝对是赚的。
现在你应该明白为啥说MCP是“AI界的USB接口”了吧?它不只是在解决技术问题,更是在重新定义AI和人类工具之间的协作方式。下次当你看到AI自动完成从需求分析到代码部署的全流程时,别忘了背后可能有MCP在默默搭桥铺路。
既然看到这里了,如果觉得不错,随手点个赞、在看、转发三连吧,如果想第一时间收到推送,也可以给我个星标⭐~谢谢你看我的文章,我们,下次再见。