返回技术博客

MCP 协议实战指南:从原理到自己动手写一个 MCP Server

上一篇聊了 Agent 和 ChatBot 的区别,核心结论是:Agent 能做事,因为它能调用工具。

但紧接着就有一个问题:工具怎么接进来?

你写了一个查天气的函数,Claude 怎么知道这个函数存在?怎么知道要传什么参数?怎么拿到返回结果?如果你想把同一个工具给 Cursor 和 GitHub Copilot 也用上,难道要针对每个平台各写一套?

MCP(Model Context Protocol) 就是来解决这个问题的。

用一个类比秒懂 MCP

如果你记得 USB 出现之前的日子。打印机用并口,鼠标用 PS/2,扫描仪用 SCSI,每种设备一种接口。USB 出来之后,一个口搞定所有设备。

MCP 干的是同样的事,只不过连接的不是硬件和电脑,而是工具和 AI Agent。

写一次工具,Claude 能用,Cursor 能用,Copilot 也能用。

MCP 长什么样

架构很经典,客户端-服务器模式:

+---------------+     JSON-RPC     +---------------+
|  MCP Client   | <--------------> |  MCP Server   |
|  (Agent 端)   |                  |  (工具端)     |
+---------------+                  +---------------+
       |                                  |
       v                                  v
   Claude Code                      数据库/API/文件
   Cursor                           Slack/钉钉
   其他 Agent                       你能想到的任何东西

Client 是 Agent 那边的,负责问"你有什么工具"和"帮我调一下这个工具"。Server 是工具那边的,负责回答"我有这些工具"和"调完了,结果在这"。

中间的通信用 JSON-RPC 2.0,传输层支持两种:stdio(Agent 直接启动你的进程)和 HTTP+SSE(工具作为独立服务运行)。个人用 stdio 就行,简单省事。

一个 MCP Server 要做什么

说到底就三件事:

声明工具。 告诉 Agent 你有哪些工具、每个工具干什么、需要什么参数。这些信息要写得清楚,Agent 是靠这些描述来决定什么时候调用什么工具的。

执行调用。 Agent 决定用某个工具时,会传过来工具名和参数。你负责执行具体逻辑(查数据库、调 API、读文件),然后把结果返回。

返回结果。 结果以结构化的格式回去,Agent 看到结果后决定下一步做什么,可能直接回复用户,也可能基于结果再调另一个工具。

除了工具(Tools),MCP Server 还能提供资源(Resources,Agent 可以读取的数据)和提示模板(Prompts,预定义的提示词)。但最常用的就是 Tools,下面的实战也聚焦在这个上面。

实战:用 Python 写一个天气查询 MCP Server

直接上代码。

先装依赖:

pip install mcp httpx

完整代码:

import json
import httpx
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent

app = Server("weather-server")


@app.list_tools()
async def list_tools():
    return [
        Tool(
            name="get_weather",
            description="获取指定城市的当前天气信息",
            inputSchema={
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称,如 Beijing、Shanghai"
                    }
                },
                "required": ["city"]
            }
        )
    ]


@app.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "get_weather":
        city = arguments["city"]
        async with httpx.AsyncClient() as client:
            resp = await client.get(
                f"https://wttr.in/{city}",
                params={"format": "j1"}
            )
            data = resp.json()

        current = data["current_condition"][0]
        result = {
            "city": city,
            "temperature": current["temp_C"] + "°C",
            "humidity": current["humidity"] + "%",
            "description": current["weatherDesc"][0]["value"],
            "wind": current["windspeedKmph"] + " km/h"
        }
        return [TextContent(
            type="text",
            text=json.dumps(result, ensure_ascii=False, indent=2)
        )]

    raise ValueError(f"Unknown tool: {name}")


async def main():
    async with stdio_server() as (read_stream, write_stream):
        await app.run(read_stream, write_stream)

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

代码不长,我把关键点说一下。

@app.list_tools() 装饰的函数负责声明工具。这里只有一个 get_weather,入参是城市名。description 字段非常关键,Agent 靠这个来理解你的工具是干什么的。写不清楚,Agent 就不知道什么时候该调用它。

@app.call_tool() 装饰的函数负责执行。收到调用后去 wttr.in 拿天气数据,整理成结构化的 JSON 返回。

最后几行是启动逻辑,用 stdio 方式运行。

TypeScript 版本

同样的功能,TS 写法:

import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import {
  CallToolRequestSchema,
  ListToolsRequestSchema,
} from "@modelcontextprotocol/sdk/types.js";

const server = new Server(
  { name: "weather-server", version: "1.0.0" },
  { capabilities: { tools: {} } }
);

server.setRequestHandler(ListToolsRequestSchema, async () => ({
  tools: [
    {
      name: "get_weather",
      description: "获取指定城市的当前天气信息",
      inputSchema: {
        type: "object" as const,
        properties: {
          city: { type: "string", description: "城市名称" },
        },
        required: ["city"],
      },
    },
  ],
}));

server.setRequestHandler(CallToolRequestSchema, async (request) => {
  if (request.params.name === "get_weather") {
    const city = request.params.arguments?.city as string;
    const resp = await fetch(`https://wttr.in/${city}?format=j1`);
    const data = await resp.json();
    const current = data.current_condition[0];

    return {
      content: [{
        type: "text",
        text: JSON.stringify({
          city,
          temperature: current.temp_C + "°C",
          humidity: current.humidity + "%",
          description: current.weatherDesc[0].value,
        }, null, 2),
      }],
    };
  }
  throw new Error("Unknown tool: " + request.params.name);
});

const transport = new StdioServerTransport();
await server.connect(transport);

风格不同,逻辑完全一样:声明工具、处理调用、返回结果。

接上 Claude

代码写好了,怎么让 Claude 用上?

Claude Code 在项目根目录创建 .mcp.json

{
  "mcpServers": {
    "weather": {
      "command": "python",
      "args": ["weather_server.py"]
    }
  }
}

Claude Desktop 编辑配置文件(macOS 在 ~/Library/Application Support/Claude/claude_desktop_config.json):

{
  "mcpServers": {
    "weather": {
      "command": "python",
      "args": ["/绝对路径/weather_server.py"]
    }
  }
}

重启 Claude,然后对它说"北京今天天气怎么样"。它会自动发现 get_weather 工具,调用它,拿到实时数据,组织成自然语言回复你。整个过程你看不到任何 JSON,只看到一句正常的回答。

进阶:数据库查询 Server

天气查询太玩具了。来个实际点的,让 Agent 查你的数据库:

import sqlite3
import json
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent

app = Server("db-query-server")
DB_PATH = "./data.db"


@app.list_tools()
async def list_tools():
    return [
        Tool(
            name="query",
            description="Execute a read-only SQL query. Only SELECT allowed.",
            inputSchema={
                "type": "object",
                "properties": {
                    "sql": {
                        "type": "string",
                        "description": "SELECT SQL query"
                    }
                },
                "required": ["sql"]
            }
        ),
        Tool(
            name="list_tables",
            description="List all tables in the database",
            inputSchema={
                "type": "object",
                "properties": {}
            }
        )
    ]


@app.call_tool()
async def call_tool(name: str, arguments: dict):
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row

    try:
        if name == "list_tables":
            cursor = conn.execute(
                "SELECT name FROM sqlite_master WHERE type='table'"
            )
            tables = [row["name"] for row in cursor]
            return [TextContent(type="text", text=json.dumps(tables))]

        if name == "query":
            sql = arguments["sql"].strip()
            if not sql.upper().startswith("SELECT"):
                return [TextContent(
                    type="text",
                    text="Error: Only SELECT queries are allowed"
                )]
            cursor = conn.execute(sql)
            rows = [dict(row) for row in cursor.fetchall()]
            return [TextContent(
                type="text",
                text=json.dumps(rows, ensure_ascii=False, indent=2)
            )]
    finally:
        conn.close()

    raise ValueError(f"Unknown tool: {name}")

这个例子有几个设计上的考量。

list_tables 工具的存在是为了让 Agent 先了解数据库结构。没有这个,Agent 就得靠猜来写 SQL,成功率会低很多。好的 MCP Server 设计要考虑 Agent 的工作流,它需要什么信息来做决策,就提供什么工具。

只允许 SELECT 是安全底线。你不会想让 Agent 有能力 DROP TABLE

安全这件事

把工具开放给 AI 这件事,本质上是在授权。安全要从一开始就想清楚。

最小权限。 数据库工具只给 SELECT,文件工具只给读。能不开的权限就不开。

验证输入。 Agent 传过来的参数不能直接信任。类型、长度、范围都要检查。SQL 注入在这里同样适用,Agent 可能构造出你意想不到的输入。

敏感信息不要暴露。 工具的 description 里不要写数据库地址和密码。API Key 通过环境变量传入。日志里别打印敏感参数。

设置超时。 Agent 调工具出了问题可能反复重试。没有超时限制的话,你的服务可能被打挂。

两种传输方式怎么选

stdio HTTP (SSE)
启动方式 Agent 直接启动你的进程 你自己把服务跑起来
适合场景 自己用、本地开发 团队共享、生产部署
配置难度 一个 JSON 搞定 需要处理认证和网络
安全 进程级隔离,天然安全 得自己做认证

个人用或者本地开发,stdio 完全够用。什么时候需要 HTTP?当你想把一个 MCP Server 部署到服务器上,让团队的所有人共用同一套工具的时候。

不想自己写?用现成的

社区已经有很多开箱即用的 MCP Server:

Server 功能 一句话场景
server-filesystem 文件读写 让 Agent 操作你的文件
server-github GitHub API 管理 PR、Issue
server-postgres PostgreSQL 数据分析、报表
server-slack Slack 消息 Agent 往群里发通知
mcp-server-fetch HTTP 请求 抓网页、调外部 API

在 Claude Code 里装一个 GitHub Server 就这一行:

claude mcp add github -- npx @modelcontextprotocol/server-github

装完之后你对 Claude 说"帮我看看这个 repo 最近有什么 open 的 issue",它会直接调 GitHub API 查出来告诉你。

几个常见问题

MCP 和 Function Calling 什么关系?

Function Calling 是模型的能力,模型能输出"我想调用某个函数"的结构化意图。MCP 是在这之上的协议,定义了工具注册、发现、调用、结果返回的完整流程。Function Calling 是嘴巴能说出指令,MCP 是从嘴巴到手的完整神经通路。

MCP Server 支持什么语言?

官方 SDK 有 Python 和 TypeScript。社区有 Go、Rust、Java、C# 等实现。只要能处理 JSON-RPC,任何语言都行。

一个 Agent 能同时连多个 Server 吗?

可以,而且这是推荐用法。一个 Server 管数据库,一个管文件,一个管消息推送。Agent 根据任务需要自动选择用哪个。

MCP 会不会被别的协议替代?

目前看可能性不大。除了 Anthropic 自家产品,Cursor、Windsurf、Cline、GitHub Copilot 都已经支持 MCP。OpenAI 在 2025 年 3 月也宣布支持。事实标准已经形成。

动手试试

如果你是第一次接触 MCP,我建议的路径:

先装几个现成的 Server 体验一下效果,感受 Agent 有了工具之后能做什么。然后挑一个你日常用得多的 API 或数据源,自己写一个 Server 封装起来。代码量真的不大,一个下午就能搞定。

一旦你的 Agent 能查你的数据库、调你的内部 API、操作你的文件系统,你会发现它从一个"很会聊天的助手"变成了一个"真正能帮你干活的同事"。这个体验上的差距比文字描述的要大得多。


上一篇:AI Agent 到底是什么?和 ChatBot 有什么本质区别 下一篇:AI 编程 Agent 进化史:从代码补全到自主开发