FastMCP：在几分钟内构建 MCP 服务器 (2026)

FastMCP：在几分钟内构建 MCP 服务器 (2026)

Model Context Protocol (MCP) 是将 AI 助手连接到外部工具和数据的行业新兴标准。但是从零开始构建 MCP 服务器可能非常乏味——处理传输、序列化和协议细节不应该占用你的宝贵时间。

FastMCP 解决了这个问题。它是一个高级 Python 框架，让你能像构建 Flask 或 FastAPI 应用一样，通过装饰器构建功能齐全的 MCP 服务器。本指南将展示如何从零开始，在几分钟内完成一个可运行的 MCP 服务器。

什么是 FastMCP？

FastMCP 是一个开源 Python 库，它用开发者友好的 API 包装了 MCP SDK。你不再需要手动处理协议消息和传输层，只需使用装饰器编写简单的 Python 函数。

特性	原生 MCP SDK	FastMCP
服务器设置	约 30 行样板代码	2 行
工具 (Tool) 定义	手动编写 Schema + 处理器	@mcp.tool() 装饰器
资源 (Resource) 定义	手动编写 URI + 处理器	@mcp.resource() 装饰器
类型处理	手动编写 JSON Schema	根据类型提示 (Type Hints) 自动生成
测试	手动模拟传输	内置测试客户端
代码行数 (简单服务器)	80-100 行	15-20 行

安装

pip install fastmcp

FastMCP 需要 Python 3.10 或更高版本。

构建你的第一个 MCP 服务器

以下是一个只需 15 行代码的完整 MCP 服务器：

# server.py
from fastmcp import FastMCP

mcp = FastMCP("my-tools")

@mcp.tool()
def add(a: int, b: int) -> int:
    """将两个数字相加。"""
    return a + b

@mcp.tool()
def get_weather(city: str) -> str:
    """获取指定城市的当前天气。"""
    # 在生产环境中，这里应调用真实的天气 API
    return f"{city} 的天气是 72°F，晴朗。"

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

就是这样。FastMCP 会自动：

• 根据你的类型提示生成 JSON Schema。
• 处理 MCP 协议（工具列表、工具调用、错误处理）。
• 设置 stdio 传输，以兼容 Claude Code、Claude Desktop 和其他 MCP 客户端。

运行服务器

python server.py

连接到 Claude Code

将其添加到你的 .mcp.json 或 Claude Code 配置中：

{
  "mcpServers": {
    "my-tools": {
      "command": "python",
      "args": ["/path/to/server.py"]
    }
  }
}

或者使用 Claude Code CLI：

claude mcp add my-tools python /path/to/server.py

核心概念

工具 (Tools)

工具是 AI 可以调用的函数。它们执行操作并返回结果。

from fastmcp import FastMCP
import httpx

mcp = FastMCP("dev-tools")

@mcp.tool()
def search_npm(query: str, limit: int = 5) -> str:
    """在 npm 注册表中搜索软件包。"""
    response = httpx.get(
        f"https://registry.npmjs.org/-/v1/search",
        params={"text": query, "size": limit}
    )
    results = response.json()["objects"]
    lines = []
    for pkg in results:
        p = pkg["package"]
        lines.append(f"- {p['name']} (v{p['version']}): {p.get('description', 'No description')}")
    return "\n".join(lines)

@mcp.tool()
def run_sql(query: str, database: str = "main.db") -> str:
    """针对 SQLite 数据库执行只读 SQL 查询。"""
    import sqlite3
    conn = sqlite3.connect(database)
    conn.execute("PRAGMA query_only = ON")
    cursor = conn.execute(query)
    columns = [desc[0] for desc in cursor.description] if cursor.description else []
    rows = cursor.fetchall()
    conn.close()

    if not rows:
        return "No results."

    header = " | ".join(columns)
    separator = " | ".join(["---"] * len(columns))
    data = "\n".join([" | ".join(str(cell) for cell in row) for row in rows])
    return f"{header}\n{separator}\n{data}"

资源 (Resources)

资源公开 AI 可以读取的数据。它们通过 URI 进行标识。

@mcp.resource("config://app")
def get_app_config() -> str:
    """返回应用程序配置。"""
    import json
    with open("config.json") as f:
        return json.dumps(json.load(f), indent=2)

@mcp.resource("file://{path}")
def read_file(path: str) -> str:
    """从项目目录读取文件。"""
    with open(path) as f:
        return f.read()

@mcp.resource("db://schema")
def get_database_schema() -> str:
    """返回数据库模式。"""
    import sqlite3
    conn = sqlite3.connect("main.db")
    cursor = conn.execute(
        "SELECT sql FROM sqlite_master WHERE type='table'"
    )
    schemas = [row[0] for row in cursor.fetchall()]
    conn.close()
    return "\n\n".join(schemas)

提示词 (Prompts)

提示词是 AI 可以使用的可重用提示词模板。

@mcp.prompt()
def code_review(code: str, language: str = "python") -> str:
    """生成代码审查提示。"""
    return f"""请审查以下 {language} 代码：
1. 安全漏洞
2. 性能问题
3. 代码风格和最佳实践
4. 潜在漏洞

代码内容：
```{language}
{code}

请提供具体且可操作的反馈，并附带修复代码示例。"""

@mcp.prompt() def explain_error(error_message: str, stack_trace: str = "") -> str: """生成错误解释提示。""" prompt = f"请解释此错误并建议修复方案：\n\n错误信息：{error_message}" if stack_trace: prompt += f"\n\n堆栈跟踪：\n{stack_trace}" return prompt

## 实践示例：GitHub MCP 服务器

这是一个更完整的示例——一个提供 GitHub 仓库工具的 MCP 服务器：

```python
# github_mcp.py
from fastmcp import FastMCP
import httpx
import os

mcp = FastMCP("github-tools")

GITHUB_TOKEN = os.environ.get("GITHUB_TOKEN", "")
HEADERS = {"Authorization": f"Bearer {GITHUB_TOKEN}"} if GITHUB_TOKEN else {}

@mcp.tool()
def search_repos(query: str, limit: int = 5) -> str:
    """搜索 GitHub 仓库。"""
    response = httpx.get(
        "https://api.github.com/search/repositories",
        params={"q": query, "per_page": limit},
        headers=HEADERS
    )
    repos = response.json().get("items", [])
    lines = []
    for repo in repos:
        lines.append(
            f"- [{repo['full_name']}]({repo['html_url']}) "
            f"({repo['stargazers_count']} stars): {repo['description'] or 'No description'}"
        )
    return "\n".join(lines) or "未找到仓库。"

@mcp.tool()
def get_repo_readme(owner: str, repo: str) -> str:
    """获取 GitHub 仓库的 README 内容。"""
    response = httpx.get(
        f"https://api.github.com/repos/{owner}/{repo}/readme",
        headers={**HEADERS, "Accept": "application/vnd.github.raw+json"}
    )
    if response.status_code == 200:
        return response.text
    return f"错误：无法获取 README ({response.status_code})"

@mcp.tool()
def list_issues(owner: str, repo: str, state: str = "open", limit: int = 10) -> str:
    """列出 GitHub 仓库的问题 (Issues)。"""
    response = httpx.get(
        f"https://api.github.com/repos/{owner}/{repo}/issues",
        params={"state": state, "per_page": limit},
        headers=HEADERS
    )
    issues = response.json()
    lines = []
    for issue in issues:
        labels = ", ".join([l["name"] for l in issue.get("labels", [])])
        lines.append(f"- #{issue['number']}: {issue['title']} [{labels}]")
    return "\n".join(lines) or "未找到问题。"

@mcp.tool()
def get_file_contents(owner: str, repo: str, path: str, ref: str = "main") -> str:
    """从 GitHub 仓库获取文件内容。"""
    response = httpx.get(
        f"https://api.github.com/repos/{owner}/{repo}/contents/{path}",
        params={"ref": ref},
        headers={**HEADERS, "Accept": "application/vnd.github.raw+json"}
    )
    if response.status_code == 200:
        return response.text
    return f"错误：无法获取文件 ({response.status_code})"

@mcp.resource("github://repos/{owner}/{repo}/info")
def get_repo_info(owner: str, repo: str) -> str:
    """获取 GitHub 仓库的基本信息。"""
    response = httpx.get(
        f"https://api.github.com/repos/{owner}/{repo}",
        headers=HEADERS
    )
    repo_data = response.json()
    return (
        f"名称: {repo_data['full_name']}\n"
        f"星标数: {repo_data['stargazers_count']}\n"
        f"语言: {repo_data['language']}\n"
        f"描述: {repo_data['description']}\n"
        f"URL: {repo_data['html_url']}"
    )

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

注册它：

{
  "mcpServers": {
    "github": {
      "command": "python",
      "args": ["/path/to/github_mcp.py"],
      "env": {
        "GITHUB_TOKEN": "ghp_your_token"
      }
    }
  }
}

测试你的 MCP 服务器

FastMCP 包含一个内置的测试客户端：

# test_server.py
import asyncio
from fastmcp import Client

async def test():
    async with Client("python server.py") as client:
        # 列出可用工具
        tools = await client.list_tools()
        print(f"可用工具: {[t.name for t in tools]}")

        # 调用工具
        result = await client.call_tool("add", {"a": 5, "b": 3})
        print(f"5 + 3 = {result}")

        # 列出资源
        resources = await client.list_resources()
        print(f"可用资源: {[r.uri for r in resources]}")

asyncio.run(test())

FastMCP vs. 原生 MCP SDK

下面是两种方式定义相同工具的对比：

原生 MCP SDK

from mcp.server import Server
from mcp.types import Tool, TextContent
import json

server = Server("my-server")

@server.list_tools()
async def list_tools():
    return [
        Tool(
            name="add",
            description="Add two numbers",
            inputSchema={
                "type": "object",
                "properties": {
                    "a": {"type": "integer", "description": "First number"},
                    "b": {"type": "integer", "description": "Second number"}
                },
                "required": ["a", "b"]
            }
        )
    ]

@server.call_tool()
async def call_tool(name, arguments):
    if name == "add":
        result = arguments["a"] + arguments["b"]
        return [TextContent(type="text", text=str(result))]

FastMCP

from fastmcp import FastMCP

mcp = FastMCP("my-server")

@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two numbers."""
    return a + b

FastMCP 版本只需 4 行，而原生版本需要 25 行以上。对于拥有大量工具的服务器，这种差异会急剧放大。

构建 MCP 服务器的技巧

1. 从小处着手。 先构建一两个工具，进行测试，然后扩展。
2. 利用类型提示。 FastMCP 会自动从你的类型提示生成 JSON Schema。
3. 编写高质量的 Docstrings。 AI 会阅读工具描述来决定何时使用它们。
4. 优雅处理错误。 将错误信息作为字符串返回，而不是抛出异常。
5. 保持工具功能单一。 一个工具应该做好一件事。让 AI 来组合它们。
6. 使用环境变量。 通过 MCP 配置中的 env 传递 API 密钥和配置。

总结

FastMCP 消除了构建 MCP 服务器时的样板代码，让你专注于实际功能。无论你是在封装 API、连接数据库，还是构建自定义开发者工具，都能在几分钟内将想法转化为可用的 MCP 服务器。

如果你的 MCP 工具需要生成媒体内容——图像、视频、音频或头像——免费试用 Hypereal AI —— 提供 35 个额度，无需信用卡。其 REST API 可以轻松集成到任何 MCP 服务器中，实现 AI 驱动的媒体生成。