自己写一个 MCP Server:从协议到实战的完整指南

用 Python FastMCP 3.x 从零搭建一个文件管理 MCP Server,覆盖协议概念、代码实现、远程部署和踩坑实录。代码完整可运行,不是伪代码。

MCP Hub 上现成的 Server 有几百个,但真正落到自己手上——公司内部 API、私有数据源、定制化工具链——没有一个现成的能直接用。这篇文章带你用 Python FastMCP 3.x 从零写一个文件管理 MCP Server,跑通本地调试、远程部署的完整链路。

MCP 协议核心概念速览

三个角色

角色职责例子
Host发起连接的 LLM 应用Claude Desktop、Cursor、VS Code Copilot
ClientHost 内部与 Server 通信的组件Claude 内置的 MCP Client
Server提供工具、数据、提示词的服务你写的那个

关系很简单:Host 包含 Client,Client 连接 Server。

三个核心能力

  • Tools:让 LLM 调用函数(计算、查数据库、操作文件)
  • Resources:让 LLM 读取数据(配置文件、数据库内容)
  • Prompts:预定义的提示词模板,复用对话模式

两种传输方式

方式场景协议
stdio本地运行,Host 直接启动你的进程JSON-RPC over stdin/stdout
Streamable HTTP远程部署,Host 通过 HTTP 连接HTTP POST + SSE

MCP ≠ Function Calling

这是最大的概念误区。Function Calling 是单个模型厂商的私有接口(比如 OpenAI 的 tool_use),而 MCP 是一个开放协议——任何 LLM、任何工具,只要遵循协议规范就能互通。你可以把它理解为「工具调用界的 USB 接口」。

实战:搭建文件管理 MCP Server

环境准备

1
2
3
4
5
6
7
8
9
# Python >= 3.10
python3 --version

# 创建虚拟环境
python3 -m venv mcp-env
source mcp-env/bin/activate

# 安装 FastMCP(当前最新 3.4.x,别装旧版)
pip install fastmcp

注意:网上大量教程还在用 FastMCP 2.x 的 API(@mcp.server.tool()FastMCP.from_tools() 等),这些在 3.x 已经全部改了。如果你照抄旧代码,会直接报错。

完整代码

创建 file_server.py

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
import os
import json
from pathlib import Path
from fastmcp import FastMCP

mcp = FastMCP(name="FileManager")

@mcp.tool
def list_files(dir_path: str) -> str:
    """列出指定目录下的文件和子目录。
    
    Args:
        dir_path: 要列出的目录路径
    """
    path = Path(dir_path).resolve()
    if not path.exists():
        return json.dumps({"error": f"路径不存在: {dir_path}"}, ensure_ascii=False)
    if not path.is_dir():
        return json.dumps({"error": f"不是目录: {dir_path}"}, ensure_ascii=False)
    
    entries = []
    for item in sorted(path.iterdir()):
        entry = {
            "name": item.name,
            "type": "dir" if item.is_dir() else "file",
            "size": item.stat().st_size if item.is_file() else None,
        }
        entries.append(entry)
    
    return json.dumps({
        "path": str(path),
        "count": len(entries),
        "entries": entries,
    }, ensure_ascii=False, indent=2)


@mcp.tool
def read_file(file_path: str) -> str:
    """读取文件内容(文本文件,限制 10MB)。
    
    Args:
        file_path: 文件路径
    """
    path = Path(file_path).resolve()
    if not path.exists():
        return json.dumps({"error": f"文件不存在: {file_path}"}, ensure_ascii=False)
    if not path.is_file():
        return json.dumps({"error": f"不是文件: {file_path}"}, ensure_ascii=False)
    if path.stat().st_size > 10 * 1024 * 1024:
        return json.dumps({"error": "文件超过 10MB 限制"}, ensure_ascii=False)
    
    try:
        content = path.read_text(encoding="utf-8")
        return json.dumps({
            "path": str(path),
            "size": path.stat().st_size,
            "content": content,
        }, ensure_ascii=False)
    except UnicodeDecodeError:
        return json.dumps({"error": "不是文本文件或编码不支持"}, ensure_ascii=False)


@mcp.tool
def search_files(pattern: str, directory: str = ".") -> str:
    """在指定目录下按文件名模式搜索文件。
    
    Args:
        pattern: 搜索模式,支持通配符(如 *.py、test_*)
        directory: 搜索的根目录,默认为当前目录
    """
    path = Path(directory).resolve()
    if not path.exists():
        return json.dumps({"error": f"目录不存在: {directory}"}, ensure_ascii=False)
    
    matches = list(path.glob(f"**/{pattern}"))
    results = [
        {
            "path": str(m.relative_to(path)),
            "type": "dir" if m.is_dir() else "file",
            "size": m.stat().st_size if m.is_file() else None,
        }
        for m in sorted(matches)[:100]  # 限制最多 100 条
    ]
    
    return json.dumps({
        "pattern": pattern,
        "directory": str(path),
        "count": len(results),
        "results": results,
    }, ensure_ascii=False, indent=2)


@mcp.resource("resource://cwd")
def get_cwd() -> str:
    """返回当前工作目录信息。"""
    return json.dumps({
        "cwd": os.getcwd(),
        "user": os.environ.get("USER", "unknown"),
        "home": str(Path.home()),
    }, ensure_ascii=False, indent=2)


if __name__ == "__main__":
    mcp.run()

代码要点

  1. @mcp.tool 装饰器把普通函数变成 MCP Tool,函数名就是工具名,docstring 自动变成工具描述
  2. 参数用 Python type hints(strint),FastMCP 自动生成 JSON Schema
  3. @mcp.resource("resource://cwd") 暴露一个只读数据源,Client 可以按 URI 请求
  4. 所有返回值统一用 JSON 字符串,LLM 解析起来更可靠
  5. 路径用 Path.resolve() 处理,避免相对路径歧义

测试运行

1
2
3
4
5
# 启动 Server(stdio 模式,等待 JSON-RPC 输入)
python file_server.py

# 用 MCP Inspector 调试(需要 Node.js)
npx @modelcontextprotocol/inspector python file_server.py

MCP Inspector 会打开一个 Web 界面,你可以可视化地调用每个 Tool、查看 Schema、检查返回值。

集成到 Claude Desktop

编辑 ~/Library/Application Support/Claude/claude_desktop_config.json(macOS)或对应路径:

1
2
3
4
5
6
7
8
9
{
  "mcpServers": {
    "file-manager": {
      "command": "python",
      "args": ["/绝对路径/file_server.py"],
      "cwd": "/你的工作目录"
    }
  }
}

重启 Claude Desktop,你的文件管理工具就出现在工具列表里了。

从本地到远程:部署你的 MCP Server

stdio → Streamable HTTP

本地开发用 stdio(简单直接),生产环境需要暴露 HTTP 接口。FastMCP 3.x 内置支持:

1
2
3
# 只改最后一行
if __name__ == "__main__":
    mcp.run(transport="streamable-http", host="0.0.0.0", port=8080)

或者用命令行直接启动:

1
fastmcp run file_server.py --transport streamable-http --port 8080

认证

远程部署必须加认证。MCP 规范推荐 OAuth 2.1(2025 年 6 月 spec 确定)。不要用裸 API Key——明文传输不安全。

FastMCP 3.x 支持通过中间件集成 OAuth:

1
2
3
4
from fastmcp import FastMCP

mcp = FastMCP(name="FileManager", auth_config=...)
# 具体配置参见 FastMCP 官方文档的 Authentication 章节

Docker 容器化

1
2
3
4
5
6
7
8
9
FROM python:3.12-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY file_server.py .

EXPOSE 8080
CMD ["fastmcp", "run", "file_server.py", "--transport", "streamable-http", "--host", "0.0.0.0", "--port", "8080"]
1
2
docker build -t file-mcp-server .
docker run -p 8080:8080 file-mcp-server

踩坑实录:我开发 MCP Server 时遇到的 5 个坑

坑 1:FastMCP 版本陷阱

FastMCP 从 1.x → 2.x → 3.x 快速迭代,网上 90% 的教程用的都是过时 API。典型的坑:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
# ❌ FastMCP 2.x 写法(3.x 报错)
from fastmcp import FastMCP, Context

@mcp.tool()
async def my_tool(ctx: Context) -> str:
    ...

# ✅ FastMCP 3.x 写法
@mcp.tool
def my_tool() -> str:
    ...

教训:写代码前先 pip show fastmcp 确认版本,再去查对应版本文档。

坑 2:把 MCP 当 Function Calling 写

MCP 不只是「换个方式调用函数」。Function Calling 是单次请求-响应,MCP 是持久化的能力暴露——你的 Server 一直活着,Client 随时可以调用任何 Tool、读取任何 Resource。设计思路应该是「我这个 Server 能提供什么能力」,而不是「我这个函数怎么被调用」。

坑 3:Schema 不精确,LLM 瞎调

参数的 type hints 和 docstring 直接决定 LLM 理解你工具的准确度。下面是反面教材:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
# ❌ LLM 不知道 limit 是什么范围、offset 从 0 还是 1 开始
def query(page, size):
    """查询数据"""
    ...

# ✅ 类型、范围、含义都写清楚
def query_users(
    page: int = 1, 
    size: int = 20,
) -> str:
    """分页查询用户列表。

    Args:
        page: 页码,从 1 开始
        size: 每页条数,最大 100
    """
    ...

坑 4:调试困难

MCP 用 JSON-RPC over stdio,不像 REST API 可以直接 curl 测试。你没法手动构造 JSON-RPC 消息发到 stdin。

解决方案:用 MCP Inspector(npx @modelcontextprotocol/inspector),它提供完整的可视化调试界面。或者用 FastMCP Client 写测试脚本:

1
2
3
4
5
6
7
8
9
from fastmcp import Client

async def test():
    async with Client("file_server.py") as client:
        result = await client.call_tool("list_files", {"dir_path": "/tmp"})
        print(result)

import asyncio
asyncio.run(test())

坑 5:Node 版本不兼容

如果你用 TypeScript SDK 开发 MCP Server,需要 Node >= 18。低版本会出现诡异的 ERR_MODULE_NOT_FOUNDSyntaxError。如果你用 npx @modelcontextprotocol/inspector 调试 Python Server,也需要确保 Node 版本够新。

1
2
3
4
5
6
# 检查版本
node --version  # 需要 >= 18

# 推荐用 nvm 管理
nvm install 20
nvm use 20

总结:什么时候自己写 vs 用现成的

场景建议
标准 API(GitHub、Slack、数据库)用现成 Server(MCP Hub 上找)
内部工具、私有 API、定制逻辑自己写
需要特定认证流程自己写
简单的文件/数据操作自己写(几十行代码搞定)

MCP Server 开发本身不难——FastMCP 一个装饰器搞定工具定义,type hints 自动生成 Schema。难的是理解协议设计理念(能力暴露 vs 函数调用)和避开版本陷阱。

推荐资源

下一篇我会聊 MCP + Cursor 的进阶玩法——怎么让你的 MCP Server 真正融入日常开发工作流。