自己实现一个 MCP Server:从协议解析到工具注册的完整实战
摘要
本文是 AI Agent 工程化落地实战系列第10篇,将从零开始带领读者实现一个完整的 MCP(Model Context Protocol)Server。文章涵盖 MCP Server 的核心职责、Python SDK 环境搭建、工具注册全流程、参数校验与结果格式化、调试测试方法,以及打包分发实践。通过5个以上完整代码示例和3个 Mermaid 架构图,读者可以深入理解 MCP 协议的请求-响应机制,并独立构建可复用的 MCP Server。本文基于 MCP Python SDK 1.2.x 版本,适用于 Python 3.10+ 环境。
版本声明:本文基于 MCP Python SDK
1.2.1(2025年1月发布),Python 3.10+,操作系统以 macOS/Linux 为主,Windows 用户需注意路径差异。如 SDK 版本更新导致 API 变化,请以官方文档为准。
文章目录
一、MCP Server 的核心职责:注册工具、处理请求、返回结果
在上一篇《MCP协议详解》中,我们深入剖析了 MCP 的通信模型、消息格式和传输层设计。现在,让我们从"理解协议"转向"实现协议"——亲手搭建一个 MCP Server。
1.1 MCP Server 在 Agent 架构中的位置
MCP Server 是 Agent 与外部能力之间的桥梁。它不关心 LLM 如何思考,也不关心用户如何交互,它只做三件事:
- 注册工具:向 MCP Client 声明自己能提供哪些能力
- 处理请求:接收 Client 发来的工具调用请求,执行对应逻辑
- 返回结果:将执行结果按照 MCP 协议格式返回给 Client

这种分层设计带来一个核心优势:工具的实现与 Agent 的推理完全解耦。你写 MCP Server 时,不需要知道哪个 LLM 会调用它、不需要知道用户会问什么问题,只需要定义清楚"我能做什么"和"我需要什么参数"。

图:MCP Server 从环境搭建到工具注册、调试测试、打包发布的完整开发流程
1.2 三大核心职责详解
上图展示了 MCP Server 的完整工作流程。让我们逐个拆解三大职责:
职责一:注册工具(Tool Registration)
MCP Server 在启动时需要向 Client 声明自己支持的工具列表。每个工具的元数据包括:
| 字段 | 类型 | 必填 | 说明 |
|---|---|---|---|
name | string | 是 | 工具唯一标识符,如 get_weather |
description | string | 是 | 工具功能描述,LLM 据此判断是否调用 |
inputSchema | object | 是 | JSON Schema 格式的参数定义 |
职责二:处理请求(Request Handling)
当 LLM 决定调用某个工具时,MCP Client 会通过 JSON-RPC 2.0 协议发送 tools/call 请求。Server 需要解析请求体、提取参数、执行对应函数,并将结果封装为协议规定的响应格式。
职责三:返回结果(Response Formatting)
MCP 协议规定工具调用结果必须包含 content 字段,其中是一个数组,每个元素可以是 text、image 或 resource 类型。这种设计让结果既能承载简单文本,也能承载富媒体内容。
1.3 为什么不用普通的 HTTP API?
你可能会问:Agent 调用工具,为什么不直接用 REST API?为什么要搞一套 MCP 协议?
| 维度 | 普通 HTTP API | MCP Server |
|---|---|---|
| 协议 | HTTP/REST | JSON-RPC 2.0 |
| 发现机制 | 需手动编写文档 | 自动通过 tools/list 发现 |
| 参数定义 | OpenAPI/Swagger | JSON Schema(与 LLM 原生兼容) |
| 传输层 | 仅 HTTP | stdio、SSE、WebSocket |
| 状态管理 | 通常无状态 | 支持有状态会话 |
| 与 LLM 集成 | 需要额外适配层 | 原生设计,零适配 |
MCP 的核心价值在于标准化。就像 HTTP 统一了信息获取方式一样,MCP 统一了 LLM 调用外部工具的方式。写一个 MCP Server,任何支持 MCP 的 Client(Claude Desktop、Cursor、VS Code Copilot 等)都能直接使用。
二、环境准备:Python MCP SDK 安装与项目结构
2.1 技术选型
MCP 官方提供了 TypeScript 和 Python 两个 SDK。本文选择 Python,原因有三:
- Python 是 AI/ML 生态的主流语言,读者熟悉度高
- Python SDK 的 API 设计更直观,适合教学
- 大量现有 AI 工具(LangChain、LlamaIndex)都是 Python 生态
2.2 安装 MCP Python SDK
# 创建项目目录
mkdir my-mcp-server && cd my-mcp-server
# 建议使用虚拟环境
python -m venv .venv
source .venv/bin/activate # Windows: .venv\Scripts\activate
# 安装 MCP Python SDK
pip install mcp>=1.2.0
# 验证安装
python -c "import mcp; print(mcp.__version__)"
代码解释:以上命令完成项目初始化工作。首先创建项目目录并进入,然后使用 Python 内置的 venv 模块创建隔离的虚拟环境,避免依赖污染全局环境。激活虚拟环境后,通过 pip 安装 MCP SDK(版本 1.2.0 及以上),最后通过一行简单的 Python 代码验证 SDK 是否正确安装并查看版本号。如果终端输出 1.2.x,说明环境就绪。
2.3 项目结构设计
一个好的项目结构应该让代码自解释。以下是我们将要构建的 MCP Server 项目结构:
my-mcp-server/
├── pyproject.toml # 项目元数据与依赖
├── README.md # 项目说明
├── src/
│ └── my_mcp_server/
│ ├── __init__.py # 包初始化
│ ├── server.py # MCP Server 主入口
│ ├── tools/
│ │ ├── __init__.py
│ │ ├── calculator.py # 计算器工具
│ │ ├── file_ops.py # 文件操作工具
│ │ └── web_fetch.py # 网页抓取工具
│ └── utils/
│ ├── __init__.py
│ └── validators.py # 参数校验工具
├── tests/
│ ├── test_calculator.py
│ ├── test_file_ops.py
│ └── test_server.py
└── examples/
└── client_demo.py # 客户端调用示例

2.4 依赖配置
创建 pyproject.toml 文件:
[project]
name = "my-mcp-server"
version = "0.1.0"
description = "A custom MCP Server for AI Agent integration"
requires-python = ">=3.10"
dependencies = [
"mcp>=1.2.0",
"pydantic>=2.0.0",
"httpx>=0.25.0",
]
[project.scripts]
my-mcp-server = "my_mcp_server.server:main"
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
代码解释:这是标准的 Python 项目配置文件。name 和 version 定义包的标识信息;requires-python 限定最低 Python 版本为 3.10(MCP SDK 使用了 3.10+ 的类型语法);dependencies 列出运行时依赖——mcp 是核心 SDK,pydantic 用于数据校验,httpx 用于异步 HTTP 请求。[project.scripts] 部分注册了一个命令行入口 my-mcp-server,用户安装后可以直接在终端执行该命令启动 Server。build-system 使用 hatchling 作为构建后端,这是 Python 生态中现代化的构建工具。
三、第一个 MCP Server:Hello World 工具注册
3.1 最小可用 Server
让我们从最简单的例子开始——一个只有单个工具的 MCP Server:
# src/my_mcp_server/server.py
from mcp.server.fastmcp import FastMCP
# 创建 MCP Server 实例
mcp = FastMCP("my-first-server")
# 注册一个简单工具
@mcp.tool()
def say_hello(name: str) -> str:
"""向指定人员打招呼。
Args:
name: 要打招呼的人员姓名
Returns:
一句问候语
"""
return f"Hello, {name}! 这是一条来自 MCP Server 的消息。"
# 启动 Server
if __name__ == "__main__":
mcp.run(transport="stdio")
代码解释:这段代码展示了一个最小可运行的 MCP Server。首先从 MCP SDK 导入 FastMCP 类——这是 SDK 提供的高层 API,封装了底层的 JSON-RPC 通信细节。创建实例时传入 "my-first-server" 作为 Server 名称。核心部分是 @mcp.tool() 装饰器:它将普通的 Python 函数注册为 MCP 工具,自动从函数签名和 docstring 中提取参数定义和描述信息。最后 mcp.run(transport="stdio") 以 stdio 模式启动 Server,意味着它通过标准输入/输出与 Client 通信,这是本地开发最简单的传输方式。
3.2 工具注册的底层机制
当你使用 @mcp.tool() 装饰器时,SDK 在后台做了以下工作:
理解这个流程对于后续调试至关重要。当你发现工具没有被 LLM 正确调用时,最常见的原因就是注册环节出了问题——要么是函数签名不够清晰,要么是 docstring 缺失导致 LLM 无法理解工具用途。
3.3 运行与验证
将上面的代码保存后,直接运行:
python -m my_mcp_server.server
Server 会启动并等待 stdin 输入。此时它看起来什么都没做,但实际上它已经准备好接收 JSON-RPC 消息了。要验证它是否正常工作,我们需要一个 MCP Client——这将在第五节详细讲解。
四、工具实现详解:参数校验、执行逻辑、结果格式
4.1 带参数校验的计算器工具
Hello World 太简单了,让我们实现一个更实用的工具——数学计算器,它支持四则运算并包含完整的参数校验:
# src/my_mcp_server/tools/calculator.py
from mcp.server.fastmcp import FastMCP
from typing import Literal
from pydantic import BaseModel, field_validator
class CalculatorInput(BaseModel):
"""计算器工具的输入参数模型。"""
operation: Literal["add", "subtract", "multiply", "divide"]
a: float
b: float
@field_validator("b")
@classmethod
def check_division_by_zero(cls, v, info):
"""除法运算时检查除数是否为零。"""
if info.data.get("operation") == "divide" and v == 0:
raise ValueError("除数不能为零")
return v
def register_calculator(mcp: FastMCP):
"""向 MCP Server 注册计算器工具。"""
@mcp.tool()
def calculate(operation: Literal["add", "subtract", "multiply", "divide"],
a: float, b: float) -> str:
"""执行基础四则运算。
支持加法、减法、乘法和除法。当执行除法时,
除数 b 不能为零。
Args:
operation: 运算类型,可选值:add, subtract, multiply, divide
a: 第一个操作数
b: 第二个操作数(除法时不能为零)
Returns:
运算结果的字符串表示,格式为 "a op b = result"
Raises:
ValueError: 当除数为零或运算类型无效时
"""
# 参数校验
input_data = CalculatorInput(operation=operation, a=a, b=b)
# 执行运算
ops = {
"add": lambda x, y: x + y,
"subtract": lambda x, y: x - y,
"multiply": lambda x, y: x * y,
"divide": lambda x, y: x / y,
}
result = ops[input_data.operation](input_data.a, input_data.b)
# 格式化结果
symbols = {"add": "+", "subtract": "-", "multiply": "×", "divide": "÷"}
symbol = symbols[input_data.operation]
return f"{input_data.a} {symbol} {input_data.b} = {result}"
代码解释:这个计算器工具展示了 MCP Server 工具开发的完整实践。首先定义了一个 CalculatorInput Pydantic 模型,利用 Literal 类型限定运算类型为四种之一,通过 field_validator 装饰器实现除零检查——当运算为除法且除数为零时,Pydantic 会自动抛出 ValueError。register_calculator 函数接收 MCP 实例,使用 @mcp.tool() 装饰器注册工具。工具函数内部的执行逻辑使用字典映射替代 if-else 链,代码更简洁。最终返回格式化的字符串结果,如 3.0 + 5.0 = 8.0。注意 docstring 的写法——LLM 会阅读这段描述来决定是否调用该工具,所以必须清晰准确地说明工具功能、参数含义和返回值格式。
4.2 文件操作工具:处理复杂返回类型
MCP 工具不仅可以返回文本,还可以返回结构化数据。下面是一个文件读取工具的示例:
# src/my_mcp_server/tools/file_ops.py
import os
from mcp.server.fastmcp import FastMCP
from pathlib import Path
SAFE_BASE_DIR = Path(os.environ.get("MCP_FILE_BASE", os.getcwd())).resolve()
def register_file_ops(mcp: FastMCP):
"""注册文件操作相关工具。"""
@mcp.tool()
def read_file(file_path: str, max_lines: int = 100) -> str:
"""读取指定文本文件的内容。
安全限制:只能读取预设基础目录下的文件。
支持读取代码、配置文件、日志等文本内容。
Args:
file_path: 文件相对路径(相对于基础目录)
max_lines: 最大读取行数,默认100行
Returns:
文件内容字符串,如果文件不存在或超出
安全范围则返回错误信息
"""
target = (SAFE_BASE_DIR / file_path).resolve()
# 安全检查:防止路径穿越攻击
if not str(target).startswith(str(SAFE_BASE_DIR)):
return f"错误:拒绝访问基础目录之外的文件"
if not target.exists():
return f"错误:文件 {file_path} 不存在"
if not target.is_file():
return f"错误:{file_path} 不是文件"
try:
with open(target, "r", encoding="utf-8") as f:
lines = []
for i, line in enumerate(f):
if i >= max_lines:
lines.append(f"... [截断,仅显示前 {max_lines} 行]")
break
lines.append(line.rstrip("\n"))
return "\n".join(lines)
except UnicodeDecodeError:
return f"错误:文件 {file_path} 不是有效的 UTF-8 文本文件"
except Exception as e:
return f"错误:读取文件时发生异常 - {type(e).__name__}: {e}"
@mcp.tool()
def list_directory(dir_path: str = ".") -> str:
"""列出指定目录下的文件和子目录。
Args:
dir_path: 目录相对路径,默认为当前目录
Returns:
目录内容列表,每行一个条目,
目录名后加 / 后缀
"""
target = (SAFE_BASE_DIR / dir_path).resolve()
if not str(target).startswith(str(SAFE_BASE_DIR)):
return "错误:拒绝访问基础目录之外的目录"
if not target.exists():
return f"错误:目录 {dir_path} 不存在"
if not target.is_dir():
return f"错误:{dir_path} 不是目录"
entries = []
for item in sorted(target.iterdir()):
prefix = "📁 " if item.is_dir() else "📄 "
entries.append(f"{prefix}{item.name}")
return "\n".join(entries) if entries else "目录为空"
代码解释:文件操作工具展示了两个关键实践——安全边界和错误处理。SAFE_BASE_DIR 在模块加载时通过环境变量设定基础目录,所有文件操作都被限制在该目录内。read_file 工具使用 Path.resolve() 解析路径后,检查解析后的绝对路径是否以 SAFE_BASE_DIR 开头,这种"前缀检查"是防止路径穿越攻击的标准做法。list_directory 工具返回格式化的目录列表,用 emoji 区分文件和目录,让 LLM 能更直观地理解返回内容。两个工具都包含完整的异常处理链:安全检查 → 存在性检查 → 类型检查 → 读取操作 → 编码处理,每一层都有明确的错误返回,确保 LLM 能理解失败原因并决定下一步操作。
4.3 异步工具:网络请求
MCP SDK 完全支持异步工具,这对于涉及网络 I/O 的场景至关重要:
# src/my_mcp_server/tools/web_fetch.py
import httpx
from mcp.server.fastmcp import FastMCP
async def register_web_fetch(mcp: FastMCP):
"""注册网页抓取工具(异步实现)。"""
@mcp.tool()
async def fetch_url(url: str, timeout_seconds: int = 10) -> str:
"""抓取指定 URL 的页面内容。
支持 HTTP 和 HTTPS 协议,返回页面的
原始文本内容。注意:返回内容可能较大,
建议在 prompt 中指定截断长度。
Args:
url: 要抓取的完整 URL(需包含 http:// 或 https://)
timeout_seconds: 请求超时时间,默认10秒
Returns:
页面文本内容,或错误信息
"""
if not url.startswith(("http://", "https://")):
return "错误:URL 必须以 http:// 或 https:// 开头"
try:
async with httpx.AsyncClient(
timeout=timeout_seconds,
follow_redirects=True,
verify=True,
) as client:
response = await client.get(url)
response.raise_for_status()
# 限制返回大小,避免超出 LLM 上下文
content = response.text[:50000]
if len(response.text) > 50000:
content += f"\n\n[... 内容已截断,原始大小: {len(response.text)} 字符]"
return content
except httpx.TimeoutException:
return f"错误:请求超时({timeout_seconds}秒)"
except httpx.HTTPStatusError as e:
return f"错误:HTTP {e.response.status_code} - {e.response.reason_phrase}"
except httpx.RequestError as e:
return f"错误:网络请求失败 - {type(e).__name__}: {e}"
except Exception as e:
return f"错误:未知异常 - {type(e).__name__}: {e}"
代码解释:这是 MCP 异步工具的典型实现。函数定义为 async def,SDK 会自动识别并以异步方式调用。使用 httpx.AsyncClient 进行异步 HTTP 请求,这是 Python 生态中最流行的异步 HTTP 客户端库。关键设计点包括:follow_redirects=True 自动处理重定向;verify=True 启用 SSL 证书验证;返回内容限制在 50000 字符以内,防止超出 LLM 的上下文窗口。错误处理覆盖了超时、HTTP 状态错误、网络连接错误和未知异常四个层次,每一层都返回可读的错误描述,帮助 LLM 理解失败原因。
4.4 工具注册汇总
将所有工具注册到 Server:
# src/my_mcp_server/server.py(完整版)
from mcp.server.fastmcp import FastMCP
from my_mcp_server.tools.calculator import register_calculator
from my_mcp_server.tools.file_ops import register_file_ops
from my_mcp_server.tools.web_fetch import register_web_fetch
# 创建 Server 实例
mcp = FastMCP("my-mcp-server")
# 注册所有工具模块
register_calculator(mcp)
register_file_ops(mcp)
# 异步工具需要 await
import asyncio
asyncio.get_event_loop().run_until_complete(register_web_fetch(mcp))
# 也可以直接在 server.py 中注册简单工具
@mcp.tool()
def say_hello(name: str) -> str:
"""向指定人员打招呼。
Args:
name: 要打招呼的人员姓名
Returns:
一句问候语
"""
return f"Hello, {name}! 这是一条来自 MCP Server 的消息。"
def main():
"""Server 启动入口。"""
mcp.run(transport="stdio")
if __name__ == "__main__":
main()
代码解释:这是完整的 Server 入口文件。采用模块化注册策略——每个功能模块导出一个 register_xxx 函数,接收 MCP 实例作为参数,在函数内部完成工具注册。这种模式的好处是:新增工具模块时只需写一个新的 register_xxx 函数并在 Server 中调用,不影响其他模块。注意异步工具注册函数 register_web_fetch 需要通过事件循环来执行。最终的 main() 函数调用 mcp.run(transport="stdio") 启动 Server,也可以改为 transport="sse" 来支持网络传输模式。
五、调试与测试:如何验证 MCP Server 是否正常工作
5.1 使用 MCP Inspector
MCP 官方提供了一个可视化调试工具——MCP Inspector,它可以让你在不编写 Client 代码的情况下测试 Server:
# 使用 npx 直接运行(无需全局安装)
npx @modelcontextprotocol/inspector python -m my_mcp_server.server
# 如果使用 uv 管理依赖
npx @modelcontextprotocol/inspector uv run my-mcp-server
MCP Inspector 启动后会打开浏览器界面,你可以:
- 查看所有已注册工具的列表
- 查看每个工具的参数 Schema
- 手动输入参数并执行工具
- 查看返回结果和调试日志

图:使用 MCP Inspector 和 pytest 对 MCP Server 进行可视化调试与自动化测试的完整场景
5.2 编写单元测试
对于生产级 MCP Server,单元测试必不可少。以下是计算器工具的测试示例:
# tests/test_calculator.py
import pytest
from my_mcp_server.tools.calculator import CalculatorInput
from pydantic import ValidationError
class TestCalculatorInput:
"""计算器输入参数校验测试。"""
def test_valid_add(self):
"""测试有效的加法输入。"""
data = CalculatorInput(operation="add", a=1.0, b=2.0)
assert data.operation == "add"
assert data.a == 1.0
assert data.b == 2.0
def test_valid_divide(self):
"""测试有效的除法输入。"""
data = CalculatorInput(operation="divide", a=10.0, b=2.0)
assert data.operation == "divide"
def test_divide_by_zero_raises(self):
"""除法时除数为零应抛出异常。"""
with pytest.raises(ValidationError) as exc_info:
CalculatorInput(operation="divide", a=10.0, b=0.0)
assert "除数不能为零" in str(exc_info.value)
def test_invalid_operation(self):
"""无效的运算类型应抛出异常。"""
with pytest.raises(ValidationError):
CalculatorInput(operation="power", a=2.0, b=3.0)
def test_negative_numbers(self):
"""测试负数运算。"""
data = CalculatorInput(operation="subtract", a=-5.0, b=-3.0)
assert data.a == -5.0
assert data.b == -3.0
代码解释:这是使用 pytest 编写的参数校验单元测试。TestCalculatorInput 类集中测试 CalculatorInput 模型的校验逻辑。test_valid_add 和 test_valid_divide 验证正常输入能通过校验;test_divide_by_zero_raises 验证除零场景会抛出 ValidationError 并检查错误消息内容;test_invalid_operation 验证非支持的运算类型会被拒绝;test_negative_numbers 确保负数参数能正常处理。这些测试覆盖了正常路径、边界情况和错误场景,是保证工具质量的基础。
5.3 编写集成测试
单元测试只验证工具函数本身,集成测试则验证 Server 的完整请求-响应链路:
# tests/test_server.py
import pytest
import json
import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
@pytest.fixture
async def mcp_session():
"""创建与 MCP Server 的测试会话。"""
server_params = StdioServerParameters(
command="python",
args=["-m", "my_mcp_server.server"],
env=None,
)
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
yield session
@pytest.mark.asyncio
async def test_list_tools(mcp_session):
"""测试工具列表是否正确返回。"""
tools = await mcp_session.list_tools()
tool_names = [t.name for t in tools]
assert "say_hello" in tool_names
assert "calculate" in tool_names
assert "read_file" in tool_names
assert "list_directory" in tool_names
@pytest.mark.asyncio
async def test_call_calculate(mcp_session):
"""测试计算器工具调用。"""
result = await mcp_session.call_tool(
"calculate",
{"operation": "add", "a": 3, "b": 5},
)
assert len(result.content) > 0
text_content = result.content[0]
assert "3.0 + 5.0 = 8.0" in text_content.text
@pytest.mark.asyncio
async def test_call_say_hello(mcp_session):
"""测试问候工具调用。"""
result = await mcp_session.call_tool(
"say_hello",
{"name": "World"},
)
assert "Hello, World!" in result.content[0].text
@pytest.mark.asyncio
async def test_divide_by_zero_error(mcp_session):
"""测试除零错误处理。"""
result = await mcp_session.call_tool(
"calculate",
{"operation": "divide", "a": 10, "b": 0},
)
# 工具应返回错误信息而非崩溃
assert result.isError or "错误" in result.content[0].text
代码解释:集成测试通过 MCP Client SDK 直接与 Server 通信,验证完整链路。mcp_session fixture 使用 stdio_client 启动 Server 子进程并建立会话连接,session.initialize() 完成协议握手。test_list_tools 验证所有工具都被正确注册并可通过 list_tools() API 发现。test_call_calculate 和 test_call_say_hello 测试实际的工具调用,检查返回内容是否包含预期结果。test_divide_by_zero_error 验证错误处理——Server 应该返回错误信息而不是抛出异常导致崩溃,这确保了 Agent 能理解失败原因并采取补救措施。
5.4 调试技巧总结
上表总结了调试 MCP Server 时最常见的四类问题及其排查方向。其中"工具未被发现"是最常见的新手问题——原因通常是装饰器使用错误或模块未被正确导入。建议在开发时先用 MCP Inspector 确认工具列表完整,再进行功能测试。
六、发布与复用:MCP Server 的打包与分发
6.1 为什么需要打包?
如果你的 MCP Server 只给自己用,直接运行 Python 脚本就够了。但如果你想让团队成员、社区用户也能使用你的 Server,就需要将其打包成可安装的 Python 包。打包后的 Server 可以:
- 通过
pip install一键安装 - 在 Claude Desktop、Cursor 等客户端中配置使用
- 发布到 PyPI 供全球用户下载
- 版本管理和依赖追踪
6.2 完整的打包配置
我们之前已经创建了 pyproject.toml,现在补充完整的配置:
# pyproject.toml(完整版)
[project]
name = "my-mcp-server"
version = "0.1.0"
description = "A custom MCP Server providing calculator, file ops, and web fetch tools"
readme = "README.md"
license = { text = "MIT" }
requires-python = ">=3.10"
authors = [
{ name = "Your Name", email = "you@example.com" }
]
keywords = ["mcp", "ai-agent", "tools", "model-context-protocol"]
classifiers = [
"Development Status :: 3 - Alpha",
"Intended Audience :: Developers",
"License :: OSI Approved :: MIT License",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
"Topic :: Software Development :: Libraries :: Python Modules",
]
dependencies = [
"mcp>=1.2.0",
"pydantic>=2.0.0",
"httpx>=0.25.0",
]
[project.optional-dependencies]
dev = [
"pytest>=7.0.0",
"pytest-asyncio>=0.21.0",
"pytest-cov>=4.0.0",
"ruff>=0.1.0",
"mypy>=1.0.0",
]
[project.scripts]
my-mcp-server = "my_mcp_server.server:main"
[project.urls]
Homepage = "https://github.com/yourname/my-mcp-server"
Repository = "https://github.com/yourname/my-mcp-server"
Issues = "https://github.com/yourname/my-mcp-server/issues"
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[tool.ruff]
line-length = 100
target-version = "py310"
[tool.mypy]
python_version = "3.10"
strict = true
[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]
代码解释:完整的 pyproject.toml 包含了发布到 PyPI 所需的所有元数据。classifiers 帮助 PyPI 分类和搜索。[project.optional-dependencies] 定义开发依赖组——安装时使用 pip install -e ".[dev]" 即可同时安装开发和运行依赖。[project.scripts] 注册的 my-mcp-server 命令是关键——它让用户安装后可以直接在终端运行 my-mcp-server 启动 Server,而不需要知道 Python 模块路径。[tool.ruff] 和 [tool.mypy] 配置代码风格和类型检查工具,[tool.pytest.ini_options] 配置 pytest 的异步测试模式和测试路径。
6.3 构建与发布
# 安装构建工具
pip install build twine
# 构建包
python -m build
# 检查构建结果
ls dist/
# 输出: my_mcp_server-0.1.0-py3-none-any.whl my_mcp_server-0.1.0.tar.gz
# 发布到 PyPI(需要 PyPI 账号)
twine upload dist/*
# 发布到 TestPyPI(测试环境)
twine upload --repository testpypi dist/*
代码解释:构建流程使用 Python 官方推荐的 build 工具。python -m build 命令会根据 pyproject.toml 中的配置自动构建 wheel 包(.whl)和源码包(.tar.gz)。twine 是 PyPI 上传工具,twine upload 将构建产物上传到 PyPI 仓库。建议先上传到 TestPyPI 进行验证,确认安装和运行无误后再发布到正式 PyPI。发布后,任何人都可以通过 pip install my-mcp-server 安装你的 MCP Server。
6.4 在客户端中配置使用
安装后的 MCP Server 可以在各种支持 MCP 的客户端中使用。以 Claude Desktop 为例:
// ~/Library/Application Support/Claude/claude_desktop_config.json (macOS)
// 或 %APPDATA%\Claude\claude_desktop_config.json (Windows)
{
"mcpServers": {
"my-tools": {
"command": "my-mcp-server",
"env": {
"MCP_FILE_BASE": "/Users/yourname/projects"
}
}
}
}
代码解释:这是 Claude Desktop 的 MCP Server 配置文件。mcpServers 对象的每个 key 是 Server 的别名(这里是 my-tools),Claude 会用这个名字来引用 Server。command 指定启动命令——因为我们在 pyproject.toml 中注册了 my-mcp-server 入口点,这里直接写命令名即可,Claude 会在 PATH 中查找它。env 字段可以传递环境变量,这里设置了 MCP_FILE_BASE 来配置文件操作工具的基础目录。修改配置后重启 Claude Desktop,你就可以在对话中使用你的自定义工具了。
七、适用边界与风险提示
7.1 MCP Server 的适用场景
MCP Server 并非银弹,它有明确的适用场景:
适合用 MCP Server 的场景:
| 场景 | 示例 | 原因 |
|---|---|---|
| 本地工具集成 | 文件操作、代码执行、数据库查询 | stdio 传输无需网络配置 |
| 团队共享工具 | 内部 API 调用、数据查询工具 | 打包后团队可一键安装 |
| 多 Agent 复用 | 多个 Agent 都需要的通用工具 | 工具注册一次,多处使用 |
| 有状态工具会话 | 需要维持上下文的操作 | MCP 支持有状态连接 |
不适合用 MCP Server 的场景:
| 场景 | 原因 | 替代方案 |
|---|---|---|
| 简单的一次性查询 | 开发成本高于收益 | 直接在 prompt 中写逻辑 |
| 超高频调用 | stdio 传输有 IPC 开销 | 直接使用 HTTP API |
| 需要复杂认证流程 | MCP 认证机制尚在完善 | 等待 OAuth 支持成熟 |
| 跨语言工具 | Python SDK 仅支持 Python | 使用 TypeScript SDK |
7.2 安全风险与防护
开发 MCP Server 时必须重视以下安全风险:
风险一:路径穿越攻击
我们的文件操作工具已经包含了防护,但这里再强调一次:
# ❌ 危险写法:直接拼接路径
target = base_dir + "/" + user_input_path
# ✅ 安全写法:resolve 后检查前缀
target = (base_dir / user_input_path).resolve()
if not str(target).startswith(str(base_dir)):
return "拒绝访问"
风险二:命令注入
如果工具涉及执行系统命令,必须使用参数化调用:
# ❌ 危险写法:字符串拼接命令
import os
os.system(f"ls {user_input}")
# ✅ 安全写法:使用 subprocess 参数列表
import subprocess
result = subprocess.run(["ls", user_input], capture_output=True, text=True)
风险三:资源耗尽
工具应该对返回内容大小进行限制,防止 LLM 上下文溢出:
# 限制返回大小
MAX_RESPONSE_SIZE = 50000 # 字符数
content = result[:MAX_RESPONSE_SIZE]
if len(result) > MAX_RESPONSE_SIZE:
content += f"\n[截断,原始大小: {len(result)} 字符]"
7.3 性能考量
| 性能维度 | 建议 | 原因 |
|---|---|---|
| 同步 vs 异步 | I/O 密集型用异步 | 避免阻塞事件循环 |
| 返回大小 | 控制在 50K 字符内 | 防止 LLM 上下文溢出 |
| 超时设置 | 所有网络请求设置超时 | 防止无限等待 |
| 工具数量 | 单 Server 控制在 20 个内 | 避免工具列表过长影响 LLM 选择 |
| 日志级别 | 生产环境用 WARNING | 减少日志 I/O 开销 |
八、总结
本文核心回顾
本文从零开始实现了一个完整的 MCP Server,覆盖了从协议理解到打包发布的全流程。核心要点如下:
-
MCP Server 的三大职责——注册工具、处理请求、返回结果——构成了 Agent 与外部能力之间的标准化桥梁。通过
tools/list和tools/call两个核心 RPC 方法,Server 向 Client 声明能力并执行具体操作。 -
FastMCP 高层 API 大幅降低了开发门槛。
@mcp.tool()装饰器自动从函数签名和 docstring 中提取工具元数据,开发者只需关注业务逻辑本身。Pydantic 模型的引入让参数校验变得声明式且类型安全。 -
安全设计是 MCP Server 的生命线。路径穿越防护、命令注入防御、返回大小限制、超时控制——这四道防线缺一不可。文件操作工具的
SAFE_BASE_DIR设计和路径前缀检查是生产级安全实践的典范。 -
调试测试体系 包括三个层次:MCP Inspector 可视化验证、pytest 单元测试覆盖校验逻辑、集成测试验证完整请求-响应链路。三层测试确保 Server 在开发、迭代和发布各阶段的质量。
-
打包分发 让 MCP Server 从个人脚本变为可复用的生态组件。
pyproject.toml的完整配置、[project.scripts]入口点注册、PyPI 发布流程——这些步骤让你的 Server 能被全球开发者使用。
从协议到实践的思考
实现 MCP Server 的过程,本质上是将"工具调用"这一行为标准化。在 MCP 出现之前,每个 Agent 框架都有自己的工具定义方式——LangChain 的 Tool 类、OpenAI 的 Function Calling、Anthropic 的 Tool Use——碎片化的生态让工具难以复用。MCP 的价值在于提供了一个中立的标准,就像 HTTP 之于 Web 一样。
但 MCP 也不是终点。在下一篇《Agent工具设计原则》中,我们将跳出协议实现细节,从更高维度讨论:什么样的工具是好工具?工具粒度如何划分?工具描述怎么写才能让 LLM 准确理解?这些问题的答案,将决定你的 Agent 是否真正好用。
如果你按照本文的步骤实现了自己的 MCP Server,欢迎在评论区分享你的工具列表和遇到的问题。实践出真知,代码见功夫。

图:MCP Server 打包发布到 PyPI 并在客户端配置使用的完整分发流程
参考资料
- MCP 官方文档 - Model Context Protocol Specification
- MCP Python SDK - GitHub Repository
- MCP Inspector - 调试工具官方文档
- JSON-RPC 2.0 规范
- Pydantic V2 官方文档
- Hatchling 构建后端文档
- Python 打包用户指南 - pyproject.toml
- Claude Desktop MCP 配置指南
适用边界声明:本文内容基于 MCP Python SDK 1.2.x 版本,适用于 Python 3.10+ 环境。SDK API 可能随版本更新发生变化,请以官方最新文档为准。文中涉及的 Claude Desktop 配置方法可能因客户端版本不同而有差异。本文不涉及 MCP 的 SSE/WebSocket 传输模式和企业级部署方案,这些内容将在系列后续文章中讨论。
更多推荐


所有评论(0)