大家好,我是何三,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+)

  1. 装依赖
pip install mcp-server httpx
  1. 写服务端代码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())
  1. 配置客户端
    用VS Code的Cline插件(或者其他支持MCP的工具),在配置里加上:
{
  "mcpServers": {
    "joke": {
      "command": "python",
      "args": ["joke_server.py"]
    }
  }
}
  1. 实际使用
    在聊天框输入:“讲个程序员笑话”,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端怎么和这个服务跳舞。完整的调用流程分四步:

  1. 服务发现:找到可用的MCP服务(这里简化成直接配置地址)
  2. 工具协商:获取可用工具列表及使用规范
  3. 计划生成:根据用户需求决定调用哪些工具
  4. 执行调用:发送请求并处理响应

用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服务来段即兴双人舞:

  1. 启动MCP服务
python mcp_server.py
  1. 在另一个终端启动Agent
python mcp_agent.py
  1. 输入查询指令
你想查询什么?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在默默搭桥铺路。

既然看到这里了,如果觉得不错,随手点个赞、在看、转发三连吧,如果想第一时间收到推送,也可以给我个星标⭐~谢谢你看我的文章,我们,下次再见。

公众号二维码