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

这一篇开始编写真实的 Python MCP Server。

示例会同时提供 Tool、Resource 和 Prompt，并分别使用三种方式验证：

• stdio：客户端启动 Server 子进程。
• Streamable HTTP：网络服务的主路径。
• 旧 HTTP+SSE：只用于兼容测试。

代码保持简单，不接入模型，也不引入 Agent。先确认 MCP Server 本身正确，再在下一篇交给 LangChain Agent。

1. 实现方式

官方 mcp 包提供 Python SDK、底层 Client/Server 类型和 mcp.server.fastmcp。独立的 fastmcp 包在相同协议之上提供更完整的 Server、Client、测试和部署体验。

项目同时使用官方 mcp SDK 和独立的 fastmcp 包。两者不是两个互不兼容的协议；无论用哪一个库实现，客户端和服务端交换的仍是 MCP 消息。

2. 创建独立环境

MCP 代码放在 llm-learning/mcp/，使用独立 Python 3.12 环境，避免影响前面已经验证的 LangChain、MLX 和向量数据库环境。

cd /path/to/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 定义 Tool、Resource 和 Prompt。三个测试脚本分别连接不同传输方式，然后直接列出并调用这些能力。示例会有少量重复代码，但每个文件都可以独立阅读，更适合刚接触 Python 和 MCP 的读者。

图中的三条路径只改变 Client 与 Server 的连接方式，不会改变 server.py 中已经注册的能力。stdio 由 Client 管理子进程；Streamable HTTP 和旧 SSE 则先启动独立网络服务，再由 Client 连接对应端点。

4. 定义 FastMCP Server

4.1 编写代码

"""定义一个简单的 FastMCP Server。

MCP 服务端对外暴露三类能力：
  - Tool：可被 LLM / Client 调用的函数（如计算预算）
  - Resource：可被读取的静态或动态资料（如城市介绍）
  - Prompt：预置的提示词模板（如行程规划指令）

本示例用 FastMCP 装饰器注册能力，再分别通过 stdio / HTTP / SSE 传输。
"""

from fastmcp import FastMCP


# 第 1 步：创建 FastMCP 实例。
# name 和 instructions 会出现在 MCP 协议握手信息中，供 Client 识别服务用途。
mcp = FastMCP(
    name="city-trip-mcp-server",
    instructions="提供城市资料、旅行预算计算和行程规划提示词。",
)


# 演示用内存数据；真实项目可替换为数据库或外部 API。
CITY_DATA = {
    "杭州": {
        "attractions": ["西湖", "灵隐寺"],
        "daily_budget": 500,
    },
    "北京": {
        "attractions": ["故宫", "天坛"],
        "daily_budget": 600,
    },
}


# 第 2 步：注册 Tool —— Client 通过 call_tool 调用。
@mcp.tool
def calculate_trip_budget(city: str, days: int) -> str:
    """计算一个人的旅行预算。"""

    city_info = CITY_DATA.get(city)
    if city_info is None:
        return f"暂时没有 {city} 的预算资料。"

    total_budget = city_info["daily_budget"] * days
    return f"{city}旅行 {days} 天，预计需要 {total_budget} 元。"


# 第 3 步：注册 Resource —— Client 通过 read_resource 读取。
# 固定 URI，返回服务说明文档。
@mcp.resource("guide://city-trip")
def service_guide() -> str:
    """返回服务使用说明。"""

    return "支持杭州和北京，可以查询城市资料、计算预算和生成行程提示词。"


# 带路径参数的 Resource 模板：city://{city}
# FastMCP 会把 URI 中的 {city} 映射为函数参数。
@mcp.resource("city://{city}")
def city_profile(city: str) -> str:
    """根据 URI 中的城市名称返回城市资料。"""

    city_info = CITY_DATA.get(city)
    if city_info is None:
        return f"暂时没有 {city} 的资料。"

    attractions = "、".join(city_info["attractions"])
    return f"{city}的推荐景点有：{attractions}。"


# 第 4 步：注册 Prompt —— Client 通过 get_prompt 获取消息模板。
# 返回的字符串会被包装成 MCP PromptMessage，供 Agent 直接使用。
@mcp.prompt
def plan_city_trip(city: str, days: int) -> str:
    """生成旅行规划提示词。"""

    return (
        f"请为我规划一次 {days} 天的{city}旅行。"
        f"先读取 city://{city} 资源了解景点，"
        "再调用 calculate_trip_budget 工具计算预算。"
    )


# 直接运行 server.py 时，默认以 stdio 传输启动（供 Cursor / Claude Desktop 等本地集成）。
if __name__ == "__main__":
    mcp.run(transport="stdio", show_banner=False)

FastMCP 实例表示一个 Server。装饰器会把普通 Python 函数注册成不同能力。

三类能力不能只按“都是 Python 函数”来理解。Tool 可以被调用并执行操作，Resource 通过 URI 读取内容，Prompt 则根据参数返回一组提示消息。Client 会先发现能力，再使用与能力类型对应的协议操作；Prompt 的返回值仍然不是模型回答。

4.2 Tool 的参数

FastMCP 会根据函数名、Docstring 和参数类型自动生成 Tool Schema：

@mcp.tool
def calculate_trip_budget(city: str, days: int) -> str:
    """计算一个人的旅行预算。"""

这里不需要手工编写 JSON。city: str 表示城市是字符串，days: int 表示天数是整数。Client 执行 tools/list 后会获得类似下面的结构：

{
  "name": "calculate_trip_budget",
  "description": "计算一个人的旅行预算。",
  "inputSchema": {
    "type": "object",
    "properties": {
      "city": {"type": "string"},
      "days": {"type": "integer"}
    },
    "required": ["city", "days"]
  }
}

为了让示例容易理解，Tool 直接返回字符串。后面接入 Agent 时，模型仍然可以根据工具名称、说明和参数 Schema 决定是否调用它。

4.3 Resource URI

示例同时定义了固定 Resource 和 Resource Template。固定 Resource 只有一个确定 URI：

@mcp.resource("guide://city-trip")

Client 读取 guide://city-trip 时，Server 才执行函数并返回服务说明。Resource Template 则把参数放进 URI：

@mcp.resource("city://{city}")
def city_profile(city: str) -> str:
    return f"正在查询 {city} 的资料。"

执行 resources/list 可以发现固定 Resource，模板通过 resources/templates/list 发现。Client 读取 city://杭州 时，FastMCP 会把“杭州”作为 city 参数传给 city_profile()。

4.4 Prompt 参数

Prompt 和普通函数一样接收参数，本例直接返回一段字符串：

@mcp.prompt
def plan_city_trip(city: str, days: int) -> str:
    return (
        f"请为我规划一次 {days} 天的{city}旅行。"
        f"先读取 city://{city} 资源了解景点。"
    )

Prompt 中虽然写了 Resource URI 和 Tool 名称，但获取 Prompt 不会自动读取 Resource 或调用 Tool。Client 拿到的只是一段提示消息，是否继续交给模型执行，由 Host 或 Agent 决定。

5. 通过 stdio 测试 MCP

01_test_stdio.py 的完整代码如下：

"""通过 stdio 测试 FastMCP Server。

stdio 是最常见的本地 MCP 传输方式：
  - Client 把 server.py 当作子进程启动
  - 双方通过标准输入 / 标准输出交换 JSON-RPC 消息
  - 不需要单独起 HTTP 服务，适合 IDE 插件和命令行工具
"""

import asyncio
from pathlib import Path

from fastmcp import Client

# 指向同目录下的 server.py；Client 会自动以 stdio 模式拉起该进程。
SERVER_FILE = Path(__file__).with_name("server.py")

async def main() -> None:
    # 第 1 步：连接 Server 并进入异步上下文。
    # with 块结束时 Client 会自动关闭子进程。
    async with Client(SERVER_FILE) as client:
        # 第 2 步：列出 Server 注册的三类 MCP 能力。
        tools = await client.list_tools()
        resources = await client.list_resources()
        templates = await client.list_resource_templates()
        prompts = await client.list_prompts()

        print("Tools:", [tool.name for tool in tools])
        print("Resources:", [str(resource.uri) for resource in resources])
        print("Resource templates:", [template.uriTemplate for template in templates])
        print("Prompts:", [prompt.name for prompt in prompts])

        # 第 3 步：调用 Tool —— 等价于 MCP 的 tools/call。
        tool_result = await client.call_tool(
            "calculate_trip_budget",
            {"city": "杭州", "days": 3},
        )
        print("Tool result:", tool_result.content[0].text)

        # 第 4 步：读取固定 Resource。
        resource_result = await client.read_resource("guide://city-trip")
        print("Resource:", resource_result[0].text)

        # 第 5 步：读取带参数的 Resource 模板（city://杭州）。
        city_result = await client.read_resource("city://杭州")
        print("Resource template:", city_result[0].text)

        # 第 6 步：获取 Prompt 模板，返回可直接喂给 LLM 的消息。
        prompt_result = await client.get_prompt(
            "plan_city_trip",
            {"city": "杭州", "days": 3},
        )
        print("Prompt:", prompt_result.messages[0].content.text)

if __name__ == "__main__":
    asyncio.run(main())

运行：

python mcp/p02_fastmcp_server/01_test_stdio.py

真实输出中的关键部分如下：

Tools: ['calculate_trip_budget']
Resources: ['guide://city-trip']
Resource templates: ['city://{city}']
Prompts: ['plan_city_trip']
Tool result: 杭州旅行 3 天，预计需要 1500 元。
Resource: 支持杭州和北京，可以查询城市资料、计算预算和生成行程提示词。
Resource template: 杭州的推荐景点有：西湖、灵隐寺。
Prompt: 请为我规划一次 3 天的杭州旅行。先读取 city://杭州 资源了解景点，
再调用 calculate_trip_budget 工具计算预算。

这里不需要提前启动 Server。Client(SERVER_FILE) 会启动独立 Python 子进程，离开 async with 后再关闭连接和子进程。前四个 list 方法负责查看 Server 提供了什么，后面的 call_tool、read_resource 和 get_prompt 负责实际使用这些能力。

• stdio 中不能随意 print

stdio Server 的 stdout 是协议通道。如果服务端执行下面的代码：

print("Server started")

这行普通文本会和 JSON-RPC 消息混在一起，客户端可能无法解析。Server 日志应该写入 stderr，或者使用默认配置正确的日志组件。

测试脚本可以正常向自己的 stdout 打印结果，因为它是 Client 进程，不是正在输出协议消息的 Server stdout。

6. 测试 Streamable HTTP

6.1 启动 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，避免示例服务意外暴露到局域网。

6.2 代码

02_test_streamable_http.py：

"""通过 Streamable HTTP 测试 FastMCP Server。"""

import asyncio

from fastmcp import Client

MCP_URL = "http://127.0.0.1:18100/mcp"

async def main() -> None:
    async with Client(MCP_URL, timeout=15) 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(
            "calculate_trip_budget",
            {"city": "杭州", "days": 3},
        )
        print("Tool result:", tool_result.content[0].text)

        resource_result = await client.read_resource("city://杭州")
        print("Resource:", resource_result[0].text)

        prompt_result = await client.get_prompt(
            "plan_city_trip",
            {"city": "杭州", "days": 3},
        )
        print("Prompt:", prompt_result.messages[0].content.text)


if __name__ == "__main__":
    asyncio.run(main())

在第二个终端运行：

python mcp/p02_fastmcp_server/02_test_streamable_http.py

真实输出：

Tools: ['calculate_trip_budget']
Resources: ['guide://city-trip']
Prompts: ['plan_city_trip']

Tool result: 杭州旅行 3 天，预计需要 1500 元。
Resource: 杭州的推荐景点有：西湖、灵隐寺。
Prompt: 请为我规划一次 3 天的杭州旅行。先读取 city://杭州 资源了解景点，
再调用 calculate_trip_budget 工具计算预算。

同一份 Server 代码没有修改，只是 Client 从启动本地子进程改成连接独立 HTTP 端点。Tool、Resource 和 Prompt 的调用方法与 stdio 完全相同。

7. 通过 SSE 实现 MCP

旧 SSE 不作为新项目主路径，但为了理解已有项目，仍然进行一次真实验证。

7.1 启动 Server

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

7.2 代码

测试代码 03_test_legacy_sse.py：

"""通过旧 HTTP+SSE 测试 FastMCP Server。"""

import asyncio

from fastmcp import Client

SSE_URL = "http://127.0.0.1:18101/sse"

async def main() -> None:
    async with Client(SSE_URL, timeout=15) 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(
            "calculate_trip_budget",
            {"city": "杭州", "days": 3},
        )
        print("Tool result:", tool_result.content[0].text)

        resource_result = await client.read_resource("city://杭州")
        print("Resource:", resource_result[0].text)

        prompt_result = await client.get_prompt(
            "plan_city_trip",
            {"city": "杭州", "days": 3},
        )
        print("Prompt:", prompt_result.messages[0].content.text)


if __name__ == "__main__":
    asyncio.run(main())

运行后得到：

Tools: ['calculate_trip_budget']
Resources: ['guide://city-trip']
Prompts: ['plan_city_trip']

Tool result: 杭州旅行 3 天，预计需要 1500 元。
Resource: 杭州的推荐景点有：西湖、灵隐寺。
Prompt: 请为我规划一次 3 天的杭州旅行。先读取 city://杭州 资源了解景点，
再调用 calculate_trip_budget 工具计算预算。

这证明旧 SSE 不只能够建立连接，也能完整承载 Tool、Resource 和 Prompt 的 MCP 消息。不过，它仍然只应作为兼容链路，不代表新服务应该优先部署旧 SSE。

8. 三种传输方式如何选择

8.1 本机应用优先 stdio

如果 Server 只服务于本机桌面应用或命令行 Client，stdio 不需要端口，也不需要长期运行服务，配置最直接。

8.2 网络服务优先 Streamable HTTP

如果多个客户端需要通过网络访问，使用 Streamable HTTP。它支持独立服务、多客户端、Session 和流式消息，也更容易接入认证和网关。

8.3 SSE 只用于兼容

只有必须连接旧客户端或旧 Server 时才保留旧 SSE。新项目不要因为名称中有“流式”就误认为它比 Streamable HTTP 更新。

9. 总结

同一个 FastMCP Server 可以通过不同传输方式提供相同能力。Server 负责定义能力，Transport 负责传输协议消息，Client 负责初始化、发现和调用。

项目	stdio	Streamable HTTP	HTTP+SSE
Server 形态	Client 启动的子进程	独立网络服务	独立兼容服务
是否需要端口	否	是	是
主要端点	stdin/stdout	/mcp	/sse 和消息 POST 端点
多客户端	通常一会话一进程	支持	支持但属于旧协议
推荐场景	本机工具	新网络服务	兼容旧系统
验证结果	Tool、Resource Template、Resource、Prompt	Tool、Resource Template、Resource、Prompt	Tool、Resource Template、Resource、Prompt