MCP 系列 2:使用 FastMCP 开发 Python MCP 服务


上一篇介绍了 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 只定义服务能力。三个测试脚本分别验证不同传输方式,避免把服务启动和客户端调用混在一起。

同一份 FastMCP Server 通过三种传输方式提供能力

图中的三条路径只改变 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 函数注册成不同能力。

FastMCP 三类能力的注册、发现与调用关系

三类能力不能只按“都是 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

文章作者: hnbian
版权声明: 本博客所有文章除特別声明外,均采用 CC BY 4.0 许可协议。转载请注明来源 hnbian !
评论
  目录