MCP 自定义服务器开发入门指南 —— Python FastMCP 篇

MCP 教程 , , ,

MCP 自定义服务器开发入门指南 —— Python FastMCP 篇

整理日期： 2026-06-04
话题来源： 聊天记录延伸（MCP 系列教程的实战进阶篇）

简介

之前几篇文章已经详细介绍了如何使用和安装现有的 MCP 服务器。但在实际工作中，你经常会遇到 “现有的 MCP 服务器不能满足需求” 的情况——你需要将自己的内部 API、专有工具、本地数据库或自定义工作流暴露给 AI 助手。

这时候就需要自己动手开发 MCP 服务器。

本教程将使用 Python FastMCP（MCP 官方 Python SDK 提供的高阶 API）带领你从零创建一个完整可用的 MCP 服务器，并将其连接到 Claude Code 或 Hermes Agent。

学习目标

完成本教程后，你将能够：

  • 搭建 Python FastMCP 项目环境
  • 使用装饰器定义 Tools（工具）、Resources（资源）和 Prompts（提示模板）
  • 在本地测试 MCP 服务器
  • 将自定义 MCP 服务器接入到 Claude Code / Hermes Agent
  • 理解 stdio 和 HTTP 两种传输方式

前置知识

要求 说明
Python >= 3.11 FastMCP 利用 async/await 和类型注解
了解 MCP 基本概念 推荐先阅读 claude-code-mcp-使用教程.md
有基本的 Python 开发经验 熟悉函数定义、类型注解、async/await

一、环境搭建

1.1 安装 MCP Python SDK

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# 创建项目目录
mkdir my-mcp-server && cd my-mcp-server

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

# 安装 MCP SDK（内置 FastMCP 高阶 API）
pip install mcp

# 如果需要 HTTP 传输，还需要 uvicorn
pip install uvicorn

# 验证安装
python -c "from mcp.server.fastmcp import FastMCP; print('FastMCP ready ✅')"

1.2 项目结构

1
2
3
4
5
my-mcp-server/
├── .venv/                  # 虚拟环境
├── server.py               # 主服务器文件
├── pyproject.toml          # 项目配置
└── README.md               # 使用说明

1.3 理解 FastMCP

FastMCP 是 MCP Python SDK 提供的高阶 API，它用 Python 装饰器和类型注解大幅简化了 MCP 服务器的开发。相比直接使用底层 mcp.server.Server，FastMCP 的特点：

特性 FastMCP 底层 Server API
定义工具 @mcp.tool() 装饰器 手动注册 Tool 对象
定义资源 @mcp.resource() 装饰器 手动注册 ResourceTemplate
定义提示模板 @mcp.prompt() 装饰器 手动构建 Prompt 消息
类型安全 自动基于类型注解校验参数 需要手动校验
错误处理 内置异常捕获 需要自行捕获
传输方式 stdio / HTTP 自动切换 需要手动配置

二、第一个 MCP 服务器：Hello World

创建一个基础的 MCP 服务器，包含一个简单的工具和一个资源。

2.1 编写服务器代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
# server.py
from mcp.server.fastmcp import FastMCP

# 创建 MCP 服务器实例
# 名称会显示在 Claude Code/Hermes 的 MCP 列表中
mcp = FastMCP("MyFirstServer")

@mcp.tool()
def greet(name: str) -> str:
    """向用户问好"""
    return f"你好，{name}！欢迎使用自定义 MCP 服务器。 👋"

@mcp.resource("greeting://{name}")
def get_greeting_resource(name: str) -> str:
    """返回个性化的问候资源"""
    return f"Hello, {name}! This is a custom MCP resource."

if __name__ == "__main__":
    # 以 stdio 方式运行（默认）
    mcp.run(transport="stdio")

2.2 测试服务器

1
2
3
4
5
# 方法一：使用 mcp 命令行工具的 dev 模式（推荐调试用）
python -m mcp dev server.py

# 方法二：直接运行，测试 stdio 通信
echo '{"jsonrpc":"2.0","method":"tools/list","params":{},"id":1}' | python server.py

mcp dev 命令会启动一个交互式开发环境，自动连接你的 MCP 服务器，让你可以直接在终端中测试工具：

1
2
3
4
5
6
7
8
MCP Development Server
Connected to MyFirstServer

> greet(name="世界")
你好，世界！欢迎使用自定义 MCP 服务器。 👋

> list tools
- greet(name: str) -> str: 向用户问好

2.3 连接到 Claude Code

1
2
3
4
# 项目级配置（创建 .mcp.json）
claude mcp add my-first-server -s project -- python /path/to/server.py

# 或直接在 .mcp.json 中配置

.mcp.json 的内容：

1
2
3
4
5
6
7
8
{
  "mcpServers": {
    "my-first-server": {
      "command": "python",
      "args": ["/absolute/path/to/server.py"]
    }
  }
}

验证连接：

1
2
claude mcp list
# 输出应该显示 my-first-server 为 connected 状态

测试对话：

1
2
> 调用 greet 工具，传入 name=开发者
你好，开发者！欢迎使用自定义 MCP 服务器。 👋

三、核心概念详解

3.1 Tools（工具）

Tools 是 可执行的函数，AI 助手可以调用它们来完成具体任务。每个 Tool 可以接受参数并返回结果。

定义方式：

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
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("ToolDemo")

@mcp.tool()
def add(a: int, b: int) -> int:
    """计算两个数的和"""
    return a + b

@mcp.tool()
def get_weather(city: str, unit: str = "celsius") -> dict:
    """获取指定城市的天气信息
    
    Args:
        city: 城市名称（中文或英文）
        unit: 温度单位，celsius（摄氏）或 fahrenheit（华氏）
    """
    # 这里只是示例，实际应该调用天气 API
    return {
        "city": city,
        "temperature": 25 if unit == "celsius" else 77,
        "condition": "晴",
        "humidity": 60
    }

@mcp.tool()
async def fetch_data(url: str) -> str:
    """异步获取网页内容（展示 async 支持）"""
    import httpx
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        return response.text[:500]  # 返回前 500 字符

Tool 的关键特性：

特性 说明
参数类型推断 基于 Python 类型注解自动生成 JSON Schema
默认值 有默认值的参数在 AI 侧变为可选参数
文档字符串 用作工具的描述和参数说明，写清楚很关键
同步/异步 同时支持同步和 async 函数
返回值 自动序列化为 JSON

⚠️ 重要：工具的描述（docstring）直接影响 AI 是否以及如何调用它。建议格式：

  • 第一行：一句话概括功能
  • 空一行后：详细说明使用场景和注意事项
  • 参数说明（可选）

3.2 Resources（资源）

Resources 是 只读的数据源，AI 助手可以通过 URI 模式访问它们。适合暴露配置文件、文档、数据库查询结果等。

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
@mcp.resource("config://app")
def get_config() -> str:
    """返回应用配置"""
    return """
database:
  host: localhost
  port: 5432
  name: myapp
"""

@mcp.resource("docs://{topic}")
def get_documentation(topic: str) -> str:
    """返回指定主题的文档"""
    docs = {
        "mcp": "Model Context Protocol 是一种让 AI 与工具通信的标准协议...",
        "fastmcp": "FastMCP 是 MCP Python SDK 的高阶 API...",
        "claude-code": "Claude Code 是 Anthropic 的 AI 编程助手..."
    }
    return docs.get(topic, f"未找到关于 '{topic}' 的文档")

@mcp.resource("users://{user_id}/profile")
def get_user_profile(user_id: int) -> dict:
    """返回用户信息（展示参数化资源）"""
    # 模拟数据库查询
    return {
        "id": user_id,
        "name": f"User_{user_id}",
        "role": "developer",
        "joined": "2026-01-15"
    }

Resource URI 模式：

1
scheme://path/{parameter}
示例 URI 匹配规则
config://app 固定资源，无参数
docs://mcp 动态资源，topic="mcp"
users://42/profile 多参数资源，user_id=42

3.3 Prompts（提示模板）

Prompts 是 预定义的提示模板，AI 助手可以用它们来引导用户完成特定任务。

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
@mcp.prompt()
def code_review(language: str = "python") -> str:
    """生成代码审查提示模板"""
    return f"""你是一位经验丰富的 {language} 代码审查专家。请审查以下代码：

## 审查要求

1. **代码质量**：检查命名规范、代码结构、可读性
2. **性能问题**：识别潜在的性能瓶颈
3. **安全隐患**：检查常见的安全漏洞
4. **最佳实践**：是否符合 {language} 的最佳实践

请按以下格式输出审查结果：

| 问题类型 | 位置 | 严重程度 | 建议 |
|---------|------|---------|------|
| ... | ... | 高/中/低 | ... |

请开始审查："""

@mcp.prompt()
def database_query(table_name: str) -> str:
    """生成数据库查询提示模板"""
    return f"""你是一位 SQL 专家。我需要查询 {table_name} 表。

请帮我：
1. 先描述 {table_name} 表可能的字段结构
2. 生成常用的查询 SQL（SELECT、INSERT、UPDATE、DELETE）
3. 给出索引建议
4. 提供优化建议

表名：{table_name}
"""

四、实战案例：构建文件系统工具服务器

让我们构建一个真正有用的服务器——提供文件搜索和内容分析功能。

4.1 完整代码

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
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
# file_tools_server.py
import os
import json
from pathlib import Path
from typing import Optional
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("FileTools")

# 安全配置：限制可访问的目录
ALLOWED_DIRS = [
    os.path.expanduser("~/projects"),
    os.path.expanduser("~/documents"),
]


def _is_path_allowed(file_path: str) -> bool:
    """检查文件路径是否在允许的范围内"""
    abs_path = os.path.abspath(os.path.expanduser(file_path))
    for allowed in ALLOWED_DIRS:
        allowed_abs = os.path.abspath(allowed)
        if abs_path.startswith(allowed_abs):
            return True
    return False


@mcp.tool()
def search_in_files(
    directory: str,
    pattern: str,
    file_glob: str = "*",
    max_results: int = 20
) -> str:
    """在指定目录中搜索文件内容
    
    Args:
        directory: 要搜索的目录（必须在允许范围内）
        pattern: 搜索关键词
        file_glob: 文件通配符，如 *.py, *.md, *.json
        max_results: 最大返回结果数
    """
    if not _is_path_allowed(directory):
        return f"❌ 错误：目录 '{directory}' 不在允许范围内"
    
    results = []
    search_dir = Path(os.path.expanduser(directory))
    
    for file_path in search_dir.rglob(file_glob):
        if not file_path.is_file():
            continue
        try:
            content = file_path.read_text(encoding="utf-8", errors="ignore")
            if pattern.lower() in content.lower():
                results.append(str(file_path))
                if len(results) >= max_results:
                    break
        except Exception:
            continue
    
    if not results:
        return f"未找到包含 '{pattern}' 的文件"
    
    return f"找到 {len(results)} 个包含 '{pattern}' 的文件：\n" + "\n".join(results)


@mcp.tool()
def analyze_codebase(directory: str) -> str:
    """分析代码库的基本统计信息
    
    Args:
        directory: 项目目录路径
    """
    if not _is_path_allowed(directory):
        return f"❌ 错误：目录 '{directory}' 不在允许范围内"
    
    stats = {}
    total_lines = 0
    total_files = 0
    
    for file_path in Path(directory).rglob("*"):
        if not file_path.is_file():
            continue
        ext = file_path.suffix.lower()
        if ext in [".py", ".js", ".ts", ".jsx", ".tsx", ".java", ".go", ".rs", ".md", ".json", ".yaml", ".yml", ".toml", ".css", ".html"]:
            stats[ext] = stats.get(ext, 0) + 1
            try:
                lines = len(file_path.read_text(encoding="utf-8", errors="ignore").splitlines())
                total_lines += lines
                total_files += 1
            except Exception:
                continue
    
    result = f"## 📊 代码库统计\n\n"
    result += f"**目录：** {directory}\n"
    result += f"**总文件数：** {total_files}\n"
    result += f"**总代码行数：** {total_lines}\n\n"
    result += "### 按文件类型分布\n\n"
    result += "| 扩展名 | 文件数 | 占比 |\n"
    result += "|--------|--------|------|\n"
    
    for ext, count in sorted(stats.items(), key=lambda x: -x[1]):
        pct = count / total_files * 100
        result += f"| {ext} | {count} | {pct:.1f}% |\n"
    
    return result


@mcp.resource("file://{path}")
def read_file_resource(path: str) -> str:
    """读取指定文件的内容（只读资源）"""
    if not _is_path_allowed(path):
        return f"❌ 错误：文件 '{path}' 不在允许范围内"
    
    file_path = Path(os.path.expanduser(path))
    if not file_path.exists():
        return f"❌ 错误：文件 '{path}' 不存在"
    if not file_path.is_file():
        return f"❌ 错误：'{path}' 不是文件"
    
    try:
        content = file_path.read_text(encoding="utf-8", errors="ignore")
        return f"```\n{content}\n```"
    except Exception as e:
        return f"❌ 读取失败：{e}"


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

4.2 测试与使用

1
2
3
4
5
# 开发模式测试
python -m mcp dev file_tools_server.py

# 添加到 Claude Code
claude mcp add file-tools -s user -- python /path/to/file_tools_server.py

在 Claude Code 中测试：

1
2
3
> 帮我分析 ~/projects/myapp 代码库的统计信息
> 在 ~/documents 中搜索包含 "MCP" 的 .md 文件
> 读取 src/main.py 的内容

五、资源（Resources）高级用法

5.1 带查询参数的资源

1
2
3
4
5
6
7
8
@mcp.resource("data://search/{query}")
def search_resource(query: str) -> str:
    """搜索资源——注意：query 来自 URI 路径"""
    results = [
        {"title": f"结果 1：关于 {query}", "url": f"https://example.com/search?q={query}"},
        {"title": f"结果 2：{query} 的最佳实践", "url": f"https://example.com/best-practices"},
    ]
    return json.dumps(results, ensure_ascii=False, indent=2)

5.2 二进制资源（图片处理）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
@mcp.resource("image://{path}/metadata")
def get_image_metadata(path: str) -> str:
    """获取图片文件的元数据信息"""
    from PIL import Image
    img_path = Path(os.path.expanduser(path))
    if not img_path.exists():
        return f"❌ 文件不存在：{path}"
    
    with Image.open(img_path) as img:
        return json.dumps({
            "filename": img_path.name,
            "size_bytes": img_path.stat().st_size,
            "format": img.format,
            "mode": img.mode,
            "width": img.width,
            "height": img.height,
            "aspect_ratio": round(img.width / img.height, 2)
        }, ensure_ascii=False, indent=2)

六、传输方式详解

6.1 stdio 传输（本地、默认）

1
2
# 默认使用 stdio
mcp.run()  # 等价于 mcp.run(transport="stdio")

stdio 传输适合：

  • 本地使用的个人工具
  • 不需要网络暴露的内部服务
  • 对安全性要求高的场景

6.2 HTTP 传输（远程服务）

1
2
3
4
5
6
7
8
9
10
11
12
13
# server_http.py
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("RemoteServer")

@mcp.tool()
def remote_echo(message: str) -> str:
    """远程回显工具"""
    return f"[远程] 收到消息：{message}"

if __name__ == "__main__":
    # 使用 HTTP 传输，监听 8000 端口
    mcp.run(transport="http", host="0.0.0.0", port=8000)

启动 HTTP 服务器：

1
2
3
4
5
6
# 安装依赖
pip install uvicorn

# 启动
python server_http.py
# 输出：MCP HTTP server running on http://0.0.0.0:8000

在远程 Claude Code 中连接：

1
claude mcp add remote-server --transport http -- http://your-server:8000/mcp

6.3 传输方式对比

维度 stdio HTTP
通信方式 子进程 stdin/stdout HTTP/SSE
部署位置 与 AI CLI 同一机器 可部署到远程服务器
多客户端 仅本地 AI CLI 可同时服务多个客户端
启动速度 即时 需先启动服务器
安全性 文件系统隔离 需要网络访问控制
适用场景 个人工具、本地专用 团队共享、云服务集成

七、连接 Hermes Agent

在 Hermes Agent 中配置自定义 MCP 服务器的语法与 Claude Code 类似，但通过 YAML 配置：

1
2
3
4
5
6
7
8
9
10
11
12
# ~/.hermes/config.yaml
mcp_servers:
  # stdio 传输
  file-tools:
    command: "python"
    args: ["/path/to/file_tools_server.py"]
    timeout: 60

  # HTTP 传输  
  remote-server:
    url: "http://localhost:8000/mcp"
    timeout: 30

配置完成后重启 Hermes Agent，工具会自动注册，Hermes 会像使用内置工具一样使用你自定义的 MCP 工具。

八、调试技巧

8.1 查看 MCP 通信日志

1
2
3
4
# 方法一：使用 mcp dev 命令
python -m mcp dev server.py

# 方法二：添加日志输出

8.2 在服务器中打印调试信息

1
2
3
4
5
6
7
8
9
10
11
12
13
# debug_server.py
import sys
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("DebugServer")

@mcp.tool()
def debug_tool(param1: str, param2: int = 42) -> str:
    """调试工具"""
    # 注意：print 会污染 stdio 通信
    # 调试信息必须输出到 stderr
    print(f"[DEBUG] 收到参数: param1={param1}, param2={param2}", file=sys.stderr)
    return f"处理完成: {param1} x {param2}"

8.3 常见错误排查

问题 原因 解决
Tool not found 装饰器未正确注册 确认使用了 @mcp.tool() 而不是 @mcp.tool
JSON serialization error 返回值不可序列化 确保返回 str、dict、list 等 JSON 兼容类型
Connection refused 配置路径错误 检查 .mcp.json 中的命令路径是否绝对路径
ModuleNotFoundError 虚拟环境未激活 使用 venv 中 Python 的绝对路径
Permission denied 文件权限不足 检查文件是否有执行权限

九、最佳实践与安全指南

9.1 安全注意事项

1
2
3
4
5
6
7
8
9
10
11
# ✅ 好的实践：路径白名单
ALLOWED_PATHS = ["/home/user/projects", "/home/user/data"]

def validate_path(path: str) -> bool:
    return any(path.startswith(allowed) for allowed in ALLOWED_PATHS)

# ❌ 不好的实践：直接使用用户输入
@mcp.tool()
def delete_file(path: str) -> str:  # 危险！没有路径校验
    os.remove(path)
    return f"已删除 {path}"

9.2 工具设计原则

  1. 单一职责：每个工具只做一件事
  2. 清晰的命名：search_files、get_weather 比 do_stuff 好得多
  3. 详细的文档：工具描述直接影响 AI 的使用效果
  4. 参数校验：永远不要信任 AI 传入的参数
  5. 返回结构化数据：优先返回 dict/JSON 而非纯文本
  6. 异步支持：IO 密集操作使用 async 版本

9.3 项目组织模板

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
my-mcp-server/
├── server.py              # 入口文件
├── tools/                 # 工具分组
│   ├── __init__.py
│   ├── file_tools.py
│   ├── web_tools.py
│   └── database_tools.py
├── resources/             # 资源定义
│   ├── __init__.py
│   ├── config_resources.py
│   └── data_resources.py
├── prompts/               # 提示模板
│   ├── __init__.py
│   └── review_prompts.py
├── pyproject.toml
└── README.md
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# server.py 入口示例
from mcp.server.fastmcp import FastMCP
from tools.file_tools import register_file_tools
from tools.web_tools import register_web_tools
from resources.config_resources import register_config_resources

mcp = FastMCP("MyAppServer")

# 注册各模块的工具和资源
register_file_tools(mcp)
register_web_tools(mcp)
register_config_resources(mcp)

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

十、总结

通过本教程，你已经学会了使用 Python FastMCP 构建自定义 MCP 服务器的完整流程：

知识点 掌握程度
FastMCP 环境搭建 ✅
使用 @mcp.tool() 定义工具 ✅
使用 @mcp.resource() 定义资源 ✅
使用 @mcp.prompt() 定义提示模板 ✅
stdio 和 HTTP 传输方式 ✅
安全路径校验 ✅
连接 Claude Code 和 Hermes Agent ✅
调试和错误排查 ✅

快速参考：10 分钟搭建你的第一个 MCP 服务器

1
2
3
mkdir my-server && cd my-server
python3 -m venv .venv && source .venv/bin/activate
pip install mcp
1
2
3
4
5
6
7
8
9
10
# server.py
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("MyServer")

@mcp.tool()
def ping() -> str:
    return "pong"

if __name__ == "__main__":
    mcp.run(transport="stdio")
1
python -m mcp dev server.py

下一步，请阅读 《MCP 自定义服务器开发进阶指南》，深入学习错误处理、流式输出、高级工具模式、HTTP 部署和性能优化等内容。

本文基于 MCP Python SDK 官方文档和社区最佳实践整理。

CaoZHProgrammer

学习使我快乐。