FastMCP:在几分钟内构建 MCP 服务器 (2026)
为 AI 助手创建自定义 MCP 服务器的最快方式
开始使用 Hypereal 构建
通过单个 API 访问 Kling、Flux、Sora、Veo 等。免费积分开始,扩展到数百万。
无需信用卡 • 10万+ 开发者 • 企业级服务
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 服务器的技巧
- 从小处着手。 先构建一两个工具,进行测试,然后扩展。
- 利用类型提示。 FastMCP 会自动从你的类型提示生成 JSON Schema。
- 编写高质量的 Docstrings。 AI 会阅读工具描述来决定何时使用它们。
- 优雅处理错误。 将错误信息作为字符串返回,而不是抛出异常。
- 保持工具功能单一。 一个工具应该做好一件事。让 AI 来组合它们。
- 使用环境变量。 通过 MCP 配置中的
env传递 API 密钥和配置。
总结
FastMCP 消除了构建 MCP 服务器时的样板代码,让你专注于实际功能。无论你是在封装 API、连接数据库,还是构建自定义开发者工具,都能在几分钟内将想法转化为可用的 MCP 服务器。
如果你的 MCP 工具需要生成媒体内容——图像、视频、音频或头像——免费试用 Hypereal AI —— 提供 35 个额度,无需信用卡。其 REST API 可以轻松集成到任何 MCP 服务器中,实现 AI 驱动的媒体生成。
