上一篇介绍了 MCP 的架构和通信机制。这一篇开始编写真实的 Python MCP Server。
示例会同时提供 Tool、Resource 和 Prompt,并分别使用三种方式验证:
- stdio:客户端启动 Server 子进程。
- Streamable HTTP:网络服务的主路径。
- 旧 HTTP+SSE:只用于兼容测试。
代码保持简单,不接入模型,也不引入 Agent。先确认 MCP Server 本身正确,再在下一篇交给 LangChain Agent。
1. FastMCP 与 MCP Python SDK
官方 mcp 包提供 Python SDK、底层 Client/Server 类型和 mcp.server.fastmcp。独立的 fastmcp 包在相同协议之上提供更完整的 Server、Client、测试和部署体验。
本文使用:
mcp==1.27.0
fastmcp==3.2.4
两者不是两个互不兼容的协议。无论用哪一个库实现,客户端和服务端交换的仍是 MCP 消息。
2. 创建独立环境
MCP 代码放在 llm_learning/mcp/,使用独立 Python 3.12 环境,避免影响前面已经验证的 LangChain、MLX 和向量数据库环境。
cd source/_posts/llm_learning
python3.12 -m venv --prompt llm_learning_mcp .venv_mcp
source .venv_mcp/bin/activate
python -m pip install -U pip
python -m pip install \
-r mcp/p02_fastmcp_server/requirements-lock.txt
python -m pip check
实测依赖检查结果:
No broken requirements found.
3. 项目结构
mcp/p02_fastmcp_server/
├── server.py
├── 01_test_stdio.py
├── 02_test_streamable_http.py
├── 03_test_legacy_sse.py
├── requirements.in
├── requirements-lock.txt
└── README.md
server.py 只定义服务能力。三个测试脚本分别验证不同传输方式,避免把服务启动和客户端调用混在一起。

图中的三条路径只改变 Client 与 Server 的连接方式,不会改变 server.py 中已经注册的能力。stdio 由 Client 管理子进程;Streamable HTTP 和旧 SSE 则先启动独立网络服务,再由 Client 连接对应端点。
4. 定义 FastMCP Server
完整的 server.py 如下:
"""定义一个同时提供 Tool、Resource 和 Prompt 的 FastMCP 学习服务。"""
from fastmcp import FastMCP
# Server 名称和说明会在 MCP 初始化阶段提供给客户端。
mcp = FastMCP(
name="learning-mcp-server",
instructions="提供基础计算、城市资料和 MCP 学习提示词。",
)
@mcp.tool
def add(a: float, b: float) -> float:
"""计算两个数字的和。"""
return a + b
@mcp.tool
def get_city_info(city: str) -> str:
"""查询固定城市资料,不访问外部网络。"""
city_data = {
"杭州": "杭州是浙江省省会,以西湖闻名。",
"北京": "北京是中华人民共和国首都。",
}
return city_data.get(city, f"暂时没有 {city} 的资料。")
@mcp.resource("guide://mcp")
def mcp_guide() -> str:
"""返回一段固定的 MCP 学习资料。"""
return "MCP Server 可以通过 Tools、Resources 和 Prompts 向客户端提供能力。"
@mcp.prompt
def explain_topic(topic: str) -> str:
"""生成一个用于解释技术概念的提示词。"""
return f"请面向初学者,用三句话解释 {topic}。"
if __name__ == "__main__":
# 直接运行该文件时使用 stdio;HTTP 服务通过 fastmcp run 命令启动。
mcp.run(transport="stdio", show_banner=False)
FastMCP 实例表示一个 Server。装饰器会把普通 Python 函数注册成不同能力。

三类能力不能只按“都是 Python 函数”来理解。Tool 可以被调用并执行操作,Resource 通过 URI 读取内容,Prompt 则根据参数返回一组提示消息。Client 会先发现能力,再使用与能力类型对应的协议操作;Prompt 的返回值仍然不是模型回答。
5. Tool 的参数 Schema 从哪里来
add() 的两个参数使用 float 类型注解,FastMCP 会据此生成 Tool 的 JSON Schema。函数名成为工具名,Docstring 成为工具说明。
客户端执行 tools/list 后,可以获得类似下面的结构:
{
"name": "add",
"description": "计算两个数字的和。",
"inputSchema": {
"type": "object",
"properties": {
"a": {"type": "number"},
"b": {"type": "number"}
},
"required": ["a", "b"]
}
}
Schema 不是只给人阅读。后面 Agent 会把它交给 Qwen3,模型根据名称、说明和参数结构生成工具请求。
6. Resource URI 与 Prompt 参数
Resource 使用 URI 标识:
@mcp.resource("guide://mcp")
客户端读取 guide://mcp 时,Server 才执行函数并返回内容。示例使用固定资源,实际项目也可以使用带参数的 URI Template。
Prompt 则允许客户端传入参数:
@mcp.prompt
def explain_topic(topic: str) -> str:
return f"请面向初学者,用三句话解释 {topic}。"
客户端拿到的是 Prompt 消息,不是模型回答。是否将它交给模型,由 Host 或 Agent 决定。
7. 使用 stdio 进行测试
01_test_stdio.py 的完整代码如下:
"""通过 stdio 启动 FastMCP 子进程,并测试 Tool、Resource 和 Prompt。"""
import asyncio
from pathlib import Path
from fastmcp import Client
SERVER_FILE = Path(__file__).with_name("server.py")
async def main() -> None:
"""连接 stdio Server,并依次读取三种核心能力。"""
# 把 Python 文件路径交给 Client 后,客户端会启动独立 Server 子进程。
async with Client(SERVER_FILE) as client:
tools = await client.list_tools()
resources = await client.list_resources()
prompts = await client.list_prompts()
print("Tools:", [tool.name for tool in tools])
print("Resources:", [str(resource.uri) for resource in resources])
print("Prompts:", [prompt.name for prompt in prompts])
tool_result = await client.call_tool("add", {"a": 18, "b": 24})
print("add(18, 24):", tool_result.content[0].text)
resource_result = await client.read_resource("guide://mcp")
print("Resource:", resource_result[0].text)
prompt_result = await client.get_prompt(
"explain_topic",
{"topic": "MCP"},
)
print("Prompt:", prompt_result.messages[0].content.text)
if __name__ == "__main__":
asyncio.run(main())
运行:
python mcp/p02_fastmcp_server/01_test_stdio.py
真实输出:
Tools: ['add', 'get_city_info']
Resources: ['guide://mcp']
Prompts: ['explain_topic']
add(18, 24): 42.0
Resource: MCP Server 可以通过 Tools、Resources 和 Prompts 向客户端提供能力。
Prompt: 请面向初学者,用三句话解释 MCP。
这里不需要提前启动 Server。Client(SERVER_FILE) 会启动独立 Python 子进程,离开 async with 后再关闭连接和子进程。
8. stdio 中为什么不能随意 print
stdio Server 的 stdout 是协议通道。如果服务端执行下面的代码:
print("Server started")
这行普通文本会和 JSON-RPC 消息混在一起,客户端可能无法解析。Server 日志应该写入 stderr,或者使用默认配置正确的日志组件。
测试脚本可以正常向自己的 stdout 打印结果,因为它是 Client 进程,不是正在输出协议消息的 Server stdout。
9. 启动 Streamable HTTP Server
在第一个终端运行:
source .venv_mcp/bin/activate
fastmcp run mcp/p02_fastmcp_server/server.py \
--transport http \
--host 127.0.0.1 \
--port 18100
FastMCP 3 中的 –transport http 表示 Streamable HTTP,默认端点为:
http://127.0.0.1:18100/mcp
只绑定 127.0.0.1,避免学习服务意外暴露到局域网。
10. 测试 Streamable HTTP
02_test_streamable_http.py:
"""连接 FastMCP Streamable HTTP 服务并调用三种核心能力。"""
import asyncio
from fastmcp import Client
MCP_URL = "http://127.0.0.1:18100/mcp"
async def main() -> None:
"""验证 Streamable HTTP MCP 端点。"""
async with Client(MCP_URL, timeout=15) as client:
tools = await client.list_tools()
print("HTTP Tools:", [tool.name for tool in tools])
tool_result = await client.call_tool("get_city_info", {"city": "杭州"})
print("Tool result:", tool_result.content[0].text)
resource_result = await client.read_resource("guide://mcp")
print("Resource:", resource_result[0].text)
prompt_result = await client.get_prompt(
"explain_topic",
{"topic": "Streamable HTTP"},
)
print("Prompt:", prompt_result.messages[0].content.text)
if __name__ == "__main__":
asyncio.run(main())
在第二个终端运行:
python mcp/p02_fastmcp_server/02_test_streamable_http.py
真实输出:
HTTP Tools: ['add', 'get_city_info']
Tool result: 杭州是浙江省省会,以西湖闻名。
Resource: MCP Server 可以通过 Tools、Resources 和 Prompts 向客户端提供能力。
Prompt: 请面向初学者,用三句话解释 Streamable HTTP。
同一份 Server 代码没有修改,只是传输方式从子进程标准流变成了独立 HTTP 服务。
11. 真实验证旧 SSE 兼容方式
旧 SSE 不作为新项目主路径,但为了理解已有项目,仍然进行一次真实验证。
启动兼容服务:
fastmcp run mcp/p02_fastmcp_server/server.py \
--transport sse \
--host 127.0.0.1 \
--port 18101
旧 SSE 端点为:
http://127.0.0.1:18101/sse
测试代码 03_test_legacy_sse.py:
"""连接旧 HTTP+SSE MCP 服务,仅用于验证旧客户端兼容路径。"""
import asyncio
from fastmcp import Client
LEGACY_SSE_URL = "http://127.0.0.1:18101/sse"
async def main() -> None:
"""连接旧 SSE 端点并调用一个简单工具。"""
async with Client(LEGACY_SSE_URL, timeout=15) as client:
tools = await client.list_tools()
print("Legacy SSE Tools:", [tool.name for tool in tools])
result = await client.call_tool("add", {"a": 8, "b": 9})
print("add(8, 9):", result.content[0].text)
if __name__ == "__main__":
asyncio.run(main())
运行后得到:
Legacy SSE Tools: ['add', 'get_city_info']
add(8, 9): 17.0
这只能证明兼容链路可用,不代表新服务应该优先部署旧 SSE。
12. 三种传输方式如何选择
12.1 本机应用优先 stdio
如果 Server 只服务于本机桌面应用或命令行 Client,stdio 不需要端口,也不需要长期运行服务,配置最直接。
12.2 网络服务优先 Streamable HTTP
如果多个客户端需要通过网络访问,使用 Streamable HTTP。它支持独立服务、多客户端、Session 和流式消息,也更容易接入认证和网关。
12.3 旧 SSE 只用于兼容
只有必须连接旧客户端或旧 Server 时才保留旧 SSE。新项目不要因为名称中有“流式”就误认为它比 Streamable HTTP 更新。
13. 常见问题
13.1 访问 /mcp 得到 406 或其他错误
MCP 端点不是普通网页。在本文固定的 FastMCP 版本中,浏览器直接发送普通 GET 请求时,因为没有声明接受 text/event-stream,会得到 406 Not Acceptable。其他实现还可能根据是否支持独立 SSE 流、请求头和 Session 状态返回不同结果。
应使用 MCP Client 完成初始化和能力调用,不能把浏览器是否显示页面作为服务是否成功的判断标准。
13.2 客户端连接错端点
Streamable HTTP 使用 /mcp,旧 SSE 使用 /sse。端口相同也不能混用路径和传输类型。
13.3 stdio 一启动就解析失败
先检查 Server 是否向 stdout 打印了 Banner、调试文本或普通日志。stdio 的 stdout 只能包含 MCP 消息。
13.4 Tool 参数校验失败
检查调用参数名是否与函数签名一致。add() 需要 a 和 b,缺少字段或传入无法转换的类型都会失败。
13.5 端口被占用
lsof -nP -iTCP:18100 -sTCP:LISTEN
结束旧进程或更换端口,并同步修改客户端 URL。
14. 总结
同一个 FastMCP Server 可以通过不同传输方式提供相同能力。Server 负责定义能力,Transport 负责传输协议消息,Client 负责初始化、发现和调用。
| 项目 | stdio | Streamable HTTP | 旧 HTTP+SSE |
|---|---|---|---|
| Server 形态 | Client 启动的子进程 | 独立网络服务 | 独立兼容服务 |
| 是否需要端口 | 否 | 是 | 是 |
| 主要端点 | stdin/stdout | /mcp | /sse 和消息 POST 端点 |
| 多客户端 | 通常一会话一进程 | 支持 | 支持但属于旧协议 |
| 推荐场景 | 本机工具 | 新网络服务 | 兼容旧系统 |
| 本文实测 | Tool、Resource、Prompt | Tool、Resource、Prompt | Tool |