为什么要自己写 MCP Server
社区已有大量现成的 MCP Server(GitHub、Slack、PostgreSQL 等),直接拿来用就行。但以下场景需要你自己写:
- 内部工具集成:你公司自建的 API、内部数据库、私有部署的服务,没有现成 Server 可用
- 自定义工作流:你需要把多个步骤封装成一个 Tool,比如"检查代码质量 + 跑测试 + 生成报告"一站式完成
- 数据安全与隔离:敏感数据不能经过第三方 Server,需要在本地或私有环境运行
一句话判断:凡是 LLM 需要调用但你不想让第三方经手的操作,都值得写一个自己的 MCP Server。
MCP Server 核心概念速览
MCP Server 可以向客户端暴露三种能力:
| 能力 | 说明 | 类比 | 谁触发 |
|---|---|---|---|
| Tools | 模型可调用的函数 | API endpoint | LLM 自动调用 |
| Resources | 可读取的结构化数据 | GET 请求 | 应用主动拉取 |
| Prompts | 预设的提示模板 | 快捷指令 | 用户选择触发 |
本文聚焦 Tools,因为它是让 LLM "做事" 的核心机制,也是 90% 的实际需求所在。
Tool 的工作流程
1. Server 注册 Tool(名称、描述、参数 schema)
2. Client 通过 tools/list 发现可用 Tool
3. LLM 根据上下文决定调用某个 Tool
4. Client 发送 tools/call 请求
5. Server 执行并返回结果
6. LLM 拿到结果,继续推理
关键点:Tool 的描述(description)决定了 LLM 是否会正确地调用它。 描述写得好,模型用得准;描述模糊,模型会乱调或漏调。
环境准备
Python(推荐入门路径)
# 安装 uv(Python 包管理器,比 pip 快 10-100x)
curl -LsSf https://astral.sh/uv/install.sh | sh
# 创建项目
uv init snippet-server
cd snippet-server
uv venv && source .venv/bin/activate
# 安装 MCP SDK
uv add "mcp[cli]"
要求:Python 3.10+,MCP SDK 1.2.0+。
TypeScript(适合前端/全栈开发者)
mkdir snippet-server && cd snippet-server
npm init -y
npm install @modelcontextprotocol/sdk zod@3
npm install -D @types/node typescript
要求:Node.js 16+。
从零开始:构建一个实用的 MCP Server
我们要构建一个 Snippet Manager:管理代码片段的工具服务器。它提供三个 Tool:
save_snippet:保存代码片段,带标题和标签search_snippets:按关键词或标签搜索片段list_snippets:列出所有片段或按标签过滤
这是一个比天气 Demo 更实用的例子——你日常开发中真的会用它来管理代码片段。
第一步:创建 Server 实例
# snippet_server.py
import json
from pathlib import Path
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("snippet-manager")
SNIPPETS_FILE = Path.home() / ".snippets.json"
def _load_snippets() -> list[dict]:
if not SNIPPETS_FILE.exists():
return []
return json.loads(SNIPPETS_FILE.read_text(encoding="utf-8"))
def _save_snippets(snippets: list[dict]) -> None:
SNIPPETS_FILE.write_text(
json.dumps(snippets, ensure_ascii=False, indent=2),
encoding="utf-8",
)
FastMCP 是 Python MCP SDK 提供的高层封装。它会自动根据类型注解和 docstring 生成 Tool 的 inputSchema,你不需要手写 JSON Schema。
第二步:定义 Tools
@mcp.tool()
def save_snippet(title: str, code: str, tags: str = "") -> str:
"""Save a code snippet with title and optional tags.
Args:
title: A short descriptive title for the snippet
code: The code content to save
tags: Comma-separated tags for categorization (e.g. "python,async,http")
"""
snippets = _load_snippets()
for s in snippets:
if s["title"] == title:
s["code"] = code
s["tags"] = [t.strip() for t in tags.split(",") if t.strip()]
_save_snippets(snippets)
return f"Updated snippet: {title}"
snippets.append({
"title": title,
"code": code,
"tags": [t.strip() for t in tags.split(",") if t.strip()],
})
_save_snippets(snippets)
return f"Saved snippet: {title}"
@mcp.tool()
def search_snippets(query: str) -> str:
"""Search snippets by keyword in title, code, or tags.
Args:
query: Search keyword to match against title, code content, or tags
"""
snippets = _load_snippets()
query_lower = query.lower()
results = [
s for s in snippets
if query_lower in s["title"].lower()
or query_lower in s["code"].lower()
or any(query_lower in t.lower() for t in s["tags"])
]
if not results:
return f"No snippets found matching '{query}'"
output = []
for s in results:
tag_str = ", ".join(s["tags"]) if s["tags"] else "no tags"
output.append(f"## {s['title']} [{tag_str}]\n{s['code']}")
return f"Found {len(results)} snippet(s):\n\n" + "\n\n".join(output)
@mcp.tool()
def list_snippets(tag: str = "") -> str:
"""List all saved snippets, optionally filtered by tag.
Args:
tag: Optional tag to filter snippets by
"""
snippets = _load_snippets()
if tag:
tag_lower = tag.lower()
snippets = [
s for s in snippets
if any(tag_lower == t.lower() for t in s["tags"])
]
if not snippets:
if tag:
return f"No snippets found with tag '{tag}'"
return "No snippets saved yet. Use save_snippet to add one."
lines = []
for s in snippets:
tag_str = ", ".join(s["tags"]) if s["tags"] else "no tags"
lines.append(f"- {s['title']} [{tag_str}]")
header = f"Snippets" + (f" tagged '{tag}'" if tag else "")
return f"{header} ({len(snippets)}):\n" + "\n".join(lines)
注意几个关键设计:
- docstring 即 schema:FastMCP 会从 docstring 的
Args:部分提取参数描述,自动生成inputSchema - 类型注解即验证:
title: str会生成{"type": "string"}的 schema;tags: str = ""会标记为非必需参数 - 错误处理用文本返回:Tool 不应该抛异常,而是返回描述性错误文本让 LLM 理解发生了什么
第三步:运行 Server
def main():
mcp.run(transport="stdio")
if __name__ == "__main__":
main()
启动:
uv run snippet_server.py
Server 会通过 stdio 监听 JSON-RPC 消息。此时你不会看到任何输出,这是正常的——stdio 模式下 Server 通过 stdin/stdout 与客户端通信,不能往 stdout 打印任何东西。
FastMCP 如何自动生成 Tool 定义
这是 FastMCP 最省心的地方。以上面的 save_snippet 为例,FastMCP 自动生成的 Tool 定义等价于:
{
"name": "save_snippet",
"description": "Save a code snippet with title and optional tags.",
"inputSchema": {
"type": "object",
"properties": {
"title": {
"type": "string",
"description": "A short descriptive title for the snippet"
},
"code": {
"type": "string",
"description": "The code content to save"
},
"tags": {
"type": "string",
"description": "Comma-separated tags for categorization (e.g. \"python,async,http\")",
"default": ""
}
},
"required": ["title", "code"]
}
}
你写 Python 函数,FastMCP 帮你搞定协议层。类型映射关系:
| Python 类型 | JSON Schema 类型 |
|---|---|
str |
string |
int |
integer |
float |
number |
bool |
boolean |
list[str] |
array[string] |
dict |
object |
接入 Claude Code / Cursor
Server 写好了,需要让客户端知道它的存在。
Claude Desktop / Claude Code
编辑配置文件:
# macOS
code ~/Library/Application\ Support/Claude/claude_desktop_config.json
# 或 Claude Code 使用项目级配置
# .mcp.json 放在项目根目录
配置内容:
{
"mcpServers": {
"snippet-manager": {
"command": "uv",
"args": [
"--directory",
"/ABSOLUTE/PATH/TO/snippet-server",
"run",
"snippet_server.py"
]
}
}
}
command字段填uv的完整路径(用which uv获取),--directory指向项目目录,必须是绝对路径。
Cursor
在项目根目录创建 .cursor/mcp.json:
{
"mcpServers": {
"snippet-manager": {
"command": "uv",
"args": [
"--directory",
"/ABSOLUTE/PATH/TO/snippet-server",
"run",
"snippet_server.py"
]
}
}
}
Claude Code(项目级配置)
在项目根目录创建 .mcp.json:
{
"mcpServers": {
"snippet-manager": {
"command": "uv",
"args": [
"--directory",
"/ABSOLUTE/PATH/TO/snippet-server",
"run",
"snippet_server.py"
],
"env": {}
}
}
}
配置完成后重启客户端。在 Claude Desktop 中输入消息时,工具图标旁应该能看到你的 Server 名称。
调试技巧
第一原则:绝不要往 stdout 写东西
stdio 模式下,stdout 是 JSON-RPC 通信通道。往 stdout 写任何内容都会破坏协议消息,导致客户端断连。
import sys
# 错误 - 会破坏 MCP 通信
print("Server started")
# 正确 - 写到 stderr
print("Server started", file=sys.stderr)
# 正确 - 用 logging(默认输出到 stderr)
import logging
logging.basicConfig(level=logging.DEBUG)
logging.info("Server started")
TypeScript 同理:
// 错误
console.log("Server started");
// 正确
console.error("Server started");
使用 MCP Inspector 测试
在接入客户端之前,先用 Inspector 验证 Server 是否正常工作:
npx @modelcontextprotocol/inspector uv --directory /path/to/snippet-server run snippet_server.py
Inspector 提供可视化界面,可以:
- 查看
tools/list返回的所有 Tool 定义 - 手动调用
tools/call验证输入输出 - 检查 schema 是否正确生成
详细的 Inspector 使用方法参见 MCP Inspector 调试实战。
常见错误排查
| 现象 | 原因 | 解决 |
|---|---|---|
| 客户端看不到 Tool | Server 启动失败或 schema 错误 | 用 Inspector 独立测试 |
| 调用后无响应 | Tool 函数内部抛异常 | 在函数内 try/except,返回错误文本 |
| Server 启动后立即退出 | 依赖缺失或语法错误 | 单独运行 uv run snippet_server.py 看报错 |
| 中文参数乱码 | 编码问题 | 文件读写指定 encoding="utf-8" |
用 CLI 模式快速验证
# 列出所有 Tool
npx @modelcontextprotocol/inspector --cli uv run snippet_server.py --method tools/list
# 调用 save_snippet
npx @modelcontextprotocol/inspector --cli uv run snippet_server.py \
--method tools/call \
--tool-name save_snippet \
--tool-arg title="hello" \
--tool-arg code='print("hello")'
进阶:Remote Server + HTTP 传输
stdio 适合本地使用。如果你的 Server 需要被多个客户端远程访问(比如团队共享),需要使用 HTTP 传输。
三种传输方式对比
| 传输 | 场景 | 启动方式 | 客户端配置 |
|---|---|---|---|
| stdio | 本地单机 | 客户端启动进程 | command + args |
| SSE(旧版) | 远程服务 | Server 自行启动 HTTP | url 字段 |
| streamable-http(推荐) | 远程服务 | Server 自行启动 HTTP | url 字段 |
切换到 HTTP 传输
只需改一行启动代码:
mcp.run(transport="streamable-http", host="0.0.0.0", port=8080)
客户端配置也相应改变:
{
"mcpServers": {
"snippet-manager": {
"url": "http://your-server:8080/mcp"
}
}
}
OAuth 认证
远程 Server 暴露在公网时,需要认证保护。MCP 支持标准 OAuth 2.1 流程:
- Server 声明需要认证(返回 401 + WWW-Authenticate header)
- Client 引导用户到授权页面
- 用户授权后,Client 拿到 access token
- 后续请求携带 Bearer token
如果你用 Cloudflare Workers / Vercel 等平台部署 MCP Server,平台通常提供内置的认证方案,不需要自己实现完整的 OAuth 流程。
大多数个人开发者和团队场景下,stdio 模式就够用了。只有在"多用户远程访问"或"Server 部署在云端"时才需要考虑 HTTP 传输和认证。
完整代码参考
# snippet_server.py
import json
from pathlib import Path
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("snippet-manager")
SNIPPETS_FILE = Path.home() / ".snippets.json"
def _load_snippets() -> list[dict]:
if not SNIPPETS_FILE.exists():
return []
return json.loads(SNIPPETS_FILE.read_text(encoding="utf-8"))
def _save_snippets(snippets: list[dict]) -> None:
SNIPPETS_FILE.write_text(
json.dumps(snippets, ensure_ascii=False, indent=2),
encoding="utf-8",
)
@mcp.tool()
def save_snippet(title: str, code: str, tags: str = "") -> str:
"""Save a code snippet with title and optional tags.
Args:
title: A short descriptive title for the snippet
code: The code content to save
tags: Comma-separated tags for categorization (e.g. "python,async,http")
"""
snippets = _load_snippets()
for s in snippets:
if s["title"] == title:
s["code"] = code
s["tags"] = [t.strip() for t in tags.split(",") if t.strip()]
_save_snippets(snippets)
return f"Updated snippet: {title}"
snippets.append({
"title": title,
"code": code,
"tags": [t.strip() for t in tags.split(",") if t.strip()],
})
_save_snippets(snippets)
return f"Saved snippet: {title}"
@mcp.tool()
def search_snippets(query: str) -> str:
"""Search snippets by keyword in title, code, or tags.
Args:
query: Search keyword to match against title, code content, or tags
"""
snippets = _load_snippets()
query_lower = query.lower()
results = [
s for s in snippets
if query_lower in s["title"].lower()
or query_lower in s["code"].lower()
or any(query_lower in t.lower() for t in s["tags"])
]
if not results:
return f"No snippets found matching '{query}'"
output = []
for s in results:
tag_str = ", ".join(s["tags"]) if s["tags"] else "no tags"
output.append(f"## {s['title']} [{tag_str}]\n{s['code']}")
return f"Found {len(results)} snippet(s):\n\n" + "\n\n".join(output)
@mcp.tool()
def list_snippets(tag: str = "") -> str:
"""List all saved snippets, optionally filtered by tag.
Args:
tag: Optional tag to filter snippets by
"""
snippets = _load_snippets()
if tag:
tag_lower = tag.lower()
snippets = [
s for s in snippets
if any(tag_lower == t.lower() for t in s["tags"])
]
if not snippets:
if tag:
return f"No snippets found with tag '{tag}'"
return "No snippets saved yet. Use save_snippet to add one."
lines = []
for s in snippets:
tag_str = ", ".join(s["tags"]) if s["tags"] else "no tags"
lines.append(f"- {s['title']} [{tag_str}]")
header = f"Snippets" + (f" tagged '{tag}'" if tag else "")
return f"{header} ({len(snippets)}):\n" + "\n".join(lines)
def main():
mcp.run(transport="stdio")
if __name__ == "__main__":
main()
调试清单
Q: FastMCP 支持异步函数吗?
支持。@mcp.tool() 同时接受同步和异步函数。如果你的 Tool 需要发起 HTTP 请求或做 IO 操作,用 async def 即可:
@mcp.tool()
async def fetch_api_data(url: str) -> str:
"""Fetch data from a URL."""
async with httpx.AsyncClient() as client:
response = await client.get(url)
return response.text
Q: Tool 的返回值只能是字符串吗?
不是。FastMCP 支持返回字符串(自动包装为 text content)、图片(Image 对象)等。但 95% 的场景返回字符串就够了——LLM 本质上处理的就是文本。
Q: 如何在 Tool 中传递环境变量(API Key 等)?
在客户端配置的 env 字段中设置,Tool 代码中通过 os.environ 读取:
{
"mcpServers": {
"my-server": {
"command": "uv",
"args": ["run", "server.py"],
"env": {
"API_KEY": "sk-xxx"
}
}
}
}
import os
api_key = os.environ.get("API_KEY")
Q: 多个 MCP Server 可以同时运行吗?
可以。在客户端配置的 mcpServers 中添加多个条目即可,每个 Server 独立进程、独立通信。Claude Code 和 Cursor 都支持同时连接多个 Server。
Q: 本地开发时如何热重载?
stdio 模式下没有原生热重载。开发时可以用 uvicorn 或 watchfiles 配合 HTTP 传输模式;或者每次改代码后重启客户端(Claude Code 用 /mcp 命令可重新连接)。