文章目录
欢迎使用FastMCP！
https://gofastmcp.com/getting-started/welcome
快速、Python风格的构建MCP服务器和客户端的方式。
模型上下文协议（MCP）是一种新的、标准化的方式，为您的LLMs提供上下文和工具，而FastMCP使得构建MCP服务器和客户端变得简单直观。使用干净、Python风格的代码创建工具、公开资源、定义提示等：
from fastmcp import FastMCP
mcp = FastMCP("Demo 🚀")
@mcp.tool()
def add(a: int, b: int) -> int:
"""Add two numbers"""
return a + b
if __name__ == "__main__":
mcp.run()
什么是MCP？
模型上下文协议允许您构建服务器，以安全、标准化的方式向LLM应用程序公开数据和功能。它常被描述为“AI的USB-C端口”，提供了一种统一的方式将LLM连接到它们可以使用的资源。您可能更容易将其视为一个API，但它专门为LLM交互而设计。MCP服务器可以：
- 通过
Resources公开数据（将这些视为类似GET端点；它们用于将信息加载到LLM的上下文中） - 通过
Tools提供功能（类似于POST端点；它们用于执行代码或产生副作用） - 通过
Prompts定义交互模式（LLM交互的可重复使用的模板） - 以及更多！
目前有一个低级别的Python SDK可用于直接实现该协议，但FastMCP旨在通过提供一个高级、Pythonic的接口来简化这一过程。
FastMCP 1.0非常成功，现在它已被包含在官方的MCP Python SDK中！
为什么选择FastMCP？
MCP协议功能强大，但实现它需要大量的模板代码——服务器设置、协议处理程序、内容类型、错误管理。FastMCP处理所有复杂的协议细节和服务器管理，让您可以专注于构建优秀的工具。它旨在提供高级和Pythonic的体验；在大多数情况下，只需装饰一个函数即可。
FastMCP的目标是：
🚀 快速：高级接口意味着代码更少，开发更快
🍀 简单：用最少的模板代码构建MCP服务器
🐍 Pythonic：对Python开发者来说感觉自然
🔍 完整：FastMCP旨在提供核心MCP规范的全实现
FastMCP v1专注于抽象化暴露MCP服务器功能的最常见模板代码，现在已包含在官方MCP Python SDK中。FastMCP v2在此基础上扩展，引入了新的功能，主要专注于简化服务器交互，包括灵活的客户端、代理和组合以及部署。
安装
https://gofastmcp.com/getting-started/installation
安装 FastMCP
我们建议使用 uv 来安装和管理 FastMCP。
如果您计划在项目中使用 FastMCP，您可以使用以下命令将其添加为依赖项：
uv add fastmcp
或者，您可以直接使用 pip 或 uv pip 安装它：
uv
uv pip install fastmcp
pip
pip install fastmcp
验证安装
为了验证FastMCP是否正确安装，您可以运行以下命令：
fastmcp version
您应该看到以下输出：
$ fastmcp version
FastMCP version: 0.4.2.dev41+ga077727.d20250410
MCP version: 1.6.0
Python version: 3.12.2
Platform: macOS-15.3.1-arm64-arm-64bit
FastMCP root path: ~/Developer/fastmcp
用于开发的安装
如果您计划为FastMCP做出贡献，您应该首先克隆仓库并使用uv来安装所有依赖项。
git clone https://github.com/jlowin/fastmcp.git
cd fastmcp
uv sync
这将安装所有依赖项，包括开发依赖项，并创建一个虚拟环境。
要运行测试，请使用pytest：
pytest
快速开始
https://gofastmcp.com/getting-started/quickstart
欢迎！本指南将帮助您快速设置FastMCP并运行您的第一个MCP服务器。
如果您尚未安装FastMCP，请按照安装说明进行操作。
创建一个FastMCP服务器
FastMCP服务器是一组工具、资源和其他MCP组件的集合。要创建服务器，首先实例化FastMCP类。
创建一个名为my_server.py的新文件，并添加以下代码：
my_server.py
from fastmcp import FastMCP
mcp = FastMCP("My MCP Server")
那就可以了！你已经创建了一个FastMCP服务器，尽管它非常无聊。让我们添加一个工具让它更有趣。
添加工具
要添加一个返回简单问候的工具，编写一个函数并用 @mcp.tool 装饰它以将其注册到服务器：
my_server.py
from fastmcp import FastMCP
mcp = FastMCP("My MCP Server")
@mcp.tool()
def greet(name: str) -> str:
return f"Hello, {name}!"
测试服务器
要测试服务器，创建一个FastMCP客户端并将其指向服务器对象。
my_server.py
from fastmcp import FastMCP, Client
mcp = FastMCP("My MCP Server")
@mcp.tool()
def greet(name: str) -> str:
return f"Hello, {name}!"
client = Client(mcp)
async def call_tool(name: str):
async with client:
result = await client.call_tool("greet", {"name": name})
print(result)
asyncio.run(call_tool("Ford"))
这里有一些需要注意的事项：
- 客户端是异步的，因此我们需要使用
asyncio.run来运行客户端。
我们必须在使用客户端之前进入客户端上下文（async with client:）。您可以在同一个上下文中进行多个客户端调用。
运行服务器
为了使用Python运行服务器，我们需要在服务器文件中的__main__块中添加一个run语句。
my_server.py
from fastmcp import FastMCP, Client
mcp = FastMCP("My MCP Server")
@mcp.tool()
def greet(name: str) -> str:
return f"Hello, {name}!"
if __name__ == "__main__":
mcp.run()
这允许我们使用 python my_server.py 运行服务器，使用默认的 stdio 传输方式，这是将 MCP 服务器暴露给客户端的标准方式。
为什么我们需要 if __name__ == "__main__": 块？
在FastMCP生态系统内，这一行可能是不必要的。然而，包含它确保您的FastMCP服务器以一致的方式为所有用户和客户端运行，因此作为最佳实践是推荐的。
与 Python 服务器交互
现在服务器可以通过 python my_server.py 执行，我们可以像与其他 MCP 服务器一样与之交互。
在新的文件中创建一个客户端并将其指向服务器文件：
my_client.py
from fastmcp import Client
client = Client("my_server.py")
async def call_tool(name: str):
async with client:
result = await client.call_tool("greet", {"name": name})
print(result)
asyncio.run(call_tool("Ford"))
使用FastMCP CLI
要使FastMCP为我们运行服务器，我们可以使用fastmcp run命令。这将启动服务器并保持其运行，直到停止。默认情况下，它将使用stdio传输，这是一个简单的基于文本的协议，用于与服务器交互。
fastmcp run my_server.py:mcp
请注意，FastMCP 不需要服务器文件中的 __main__ 块，如果存在，它也会忽略它。相反，它会在 CLI 命令中查找提供的服务器对象（在此处，为 mcp）。如果没有提供服务器对象，fastmcp run 将会自动在文件中搜索名为“mcp”、“app”或“server”的服务器。
我们将客户端指向服务器文件，该文件被识别为Python MCP服务器，并默认通过python my_server.py执行。这会执行服务器文件的__main__块。还有其他运行服务器的方法，这些方法在服务器配置指南中有描述。
FastMCP 服务器
https://gofastmcp.com/servers/fastmcp
了解 FastMCP 服务器类的核心内容以及如何运行它。
FastMCP 应用程序的核心是 FastMCP 服务器类。这个类充当应用程序工具、资源和提示的主要容器，并管理与 MCP 客户端的通信。
创建服务器
创建服务器非常简单。您通常需要提供一个服务器名称，这有助于在客户端应用程序或日志中识别它。
from fastmcp import FastMCP
# Create a basic server instance
mcp = FastMCP(name="MyAssistantServer")
# You can also add instructions for how to interact with the server
mcp_with_instructions = FastMCP(
name="HelpfulAssistant", instructions="This server provides data analysis tools. Call get_average() to analyze numerical data."
)
组件
FastMCP服务器向客户端暴露了多种类型的组件：
工具
工具是客户端可以调用来执行操作或访问外部系统的函数。
@mcp.tool()
def multiply(a: float, b: float) -> float:
"""Multiplies two numbers together."""
return a * b
资源
资源暴露了客户端可以读取的数据源。
@mcp.resource("data://config")
def get_config() -> dict:
"""Provides the application configuration."""
return {"theme": "dark", "version": "1.0"}
资源模板
资源模板是参数化资源，允许客户端请求特定数据。
@mcp.resource("users://{user_id}/profile")
def get_user_profile(user_id: int) -> dict:
"""Retrieves a user's profile by ID."""
# The {user_id} in the URI is extracted and passed to this function
return {"id": user_id, "name": f"User {user_id}", "status": "active"}
提示
提示是用于引导LLM的可重复使用的消息模板。
@mcp.prompt()
def analyze_data(data_points: list[float]) -> str:
"""Creates a prompt asking for analysis of numerical data."""
formatted_data = ", ".join(str(point) for point in data_points)
return f"Please analyze these data points: {formatted_data}"
运行服务器
FastMCP服务器需要一个传输机制来与客户端通信。在MCP协议中，服务器通常作为独立的进程运行，客户端连接到这些进程。
__main__ 块模式
将您的服务器设置为可执行的标准方式是在一个 if __name__ == "__main__": 块内包含一个 run() 调用：
# my_server.py
from fastmcp import FastMCP
mcp = FastMCP(name="MyServer")
@mcp.tool()
def greet(name: str) -> str:
"""Greet a user by name."""
return f"Hello, {name}!"
if __name__ == "__main__":
# This code only runs when the file is executed directly
# Basic run with default settings (stdio transport)
mcp.run()
# Or with specific transport and parameters
# mcp.run(transport="sse", host="127.0.0.1", port=9000)
传输选项
FastMCP支持两种传输机制：
STDIO Transport (Default)
标准输入/输出（STDIO）传输是默认且最广泛兼容的选项：
# Run with stdio (default)
mcp.run() # or explicitly: mcp.run(transport="stdio")
使用STDIO：
- 客户端为每个会话启动一个新的服务器进程。
- 通过标准输入/输出流进行通信。
- 当客户端断开连接时，服务器进程终止。
- 这对于与Claude Desktop等工具的集成非常理想，每个对话都有自己的服务器实例。
SSE Transport (Server-Sent Events)
对于长时间运行的、为多个客户端提供服务的服务器，FastMCP支持SSE：
# Run with SSE on default host/port (0.0.0.0:8000)
mcp.run(transport="sse")
使用SSE（Server-Sent Events）：
- 服务器作为一个持久的网络服务器运行。
- 多个客户端可以同时连接。
- 服务器保持运行，直到被明确终止。
- 这对于远程访问服务非常理想。
你可以在运行服务器时直接配置传输参数：
# Configure with specific parameters
mcp.run(
transport="sse",
host="127.0.0.1", # Override default host
port=8888, # Override default port
log_level="debug" # Set logging level
)
# You can also run asynchronously with the same parameters
import asyncio
asyncio.run(
mcp.run_sse_async(
host="127.0.0.1",
port=8888,
log_level="debug"
)
)
传递给 run() 或 run_sse_async() 的传输参数会覆盖在创建 FastMCP 实例时定义的任何设置。SSE 传输最常用的参数有：
host: 要绑定的主机（默认为：“0.0.0.0”）port: 要绑定的端口（默认为：8000）log_level: 日志级别（默认为：“INFO”）
高级传输配置
在底层，FastMCP 的 run() 方法接受任意关键字参数 (**transport_kwargs)，这些参数被传递到特定传输的运行方法中：
# For SSE transport, kwargs are passed to run_sse_async()
mcp.run(transport="sse", **transport_kwargs)
# For stdio transport, kwargs are passed to run_stdio_async()
mcp.run(transport="stdio", **transport_kwargs)
这意味着任何未来的特定于传输的选项将通过相同的接口自动可用，而无需更改你的代码。
使用FastMCP CLI
FastMCP CLI提供了一种方便的方式来运行服务器：
# Run a server (defaults to stdio transport)
fastmcp run my_server.py:mcp
# Explicitly specify a transport
fastmcp run my_server.py:mcp --transport sse
# Configure SSE transport with host and port
fastmcp run my_server.py:mcp --transport sse --host 127.0.0.1 --port 8888
# With log level
fastmcp run my_server.py:mcp --transport sse --log-level DEBUG
CLI可以动态地在你文件中找到并运行FastMCP服务器对象，但包含 if __name__ == "__main__": 块确保了与所有客户端的兼容性。
服务器组合
自版本 2.2.0 新增
FastMCP 支持使用 import_server（静态复制）和 mount（实时链接）将多个服务器组合在一起。这允许您将大型应用程序组织成模块化组件或重用现有服务器。
请参阅服务器组合指南以获取完整细节、最佳实践和示例。
# Example: Importing a subserver
from fastmcp import FastMCP
import asyncio
main = FastMCP(name="Main")
sub = FastMCP(name="Sub")
@sub.tool()
def hello():
return "hi"
main.mount("sub", sub)
代理服务器
自版本 2.0.0 新增
FastMCP 可以通过使用 FastMCP.from_client 作为任何 MCP 服务器（本地或远程）的代理，让您能够连接传输或为现有服务器添加前端。例如，您可以通过 stdio 在本地暴露远程 SSE 服务器，反之亦然。
有关详细信息和高阶用法，请参阅 代理服务器 指南。
from fastmcp import FastMCP, Client
backend = Client("http://example.com/mcp/sse")
proxy = FastMCP.from_client(backend, name="ProxyServer")
# Now use the proxy like any FastMCP server
服务器配置
服务器的行为，例如传输设置（用于SSE的主机、端口）以及如何处理重复组件，可以通过ServerSettings进行配置。这些设置可以在FastMCP初始化期间传递，通过以FASTMCP_SERVER_为前缀的环境变量设置，或从.env文件中加载。
from fastmcp import FastMCP
# Configure during initialization
mcp = FastMCP(
name="ConfiguredServer", port=8080, # Directly maps to ServerSettings
on_duplicate_tools="error" # Set duplicate handling
)
# Settings are accessible via mcp.settings
print(mcp.settings.port) # Output: 8080
print(mcp.settings.on_duplicate_tools) # Output: "error"
关键配置选项
host: SSE传输的主机地址（默认：“0.0.0.0”）port: SSE传输的端口号（默认：8000）log_level: 日志记录级别（默认：“INFO”）on_duplicate_tools: 如何处理重复的工具注册on_duplicate_resources: 如何处理重复的资源注册on_duplicate_prompts: 如何处理重复的提示注册
所有这些都可以在创建FastMCP实例时直接作为参数进行配置。
认证
自版本 2.2.7 新增
FastMCP 从 MCP 协议继承了 OAuth 2.0 认证支持，允许服务器在认证之后保护其工具和资源。
OAuth 2.0 支持
mcp.server.auth 模块实现了 OAuth 2.0 服务器接口，服务器可以通过提供 OAuthServerProvider 协议的实现来使用该接口。
from fastmcp import FastMCP
from mcp.server.auth.settings import (
RevocationOptions,
ClientRegistrationOptions,
AuthSettings,
)
# Create a server with authentication
mcp = FastMCP(
name="SecureApp",
auth_provider=MyOAuthServerProvider(),
auth=AuthSettings(
issuer_url="https://myapp.com",
revocation_options=RevocationOptions(
enabled=True,
),
client_registration_options=ClientRegistrationOptions(
enabled=True,
valid_scopes=["myscope", "myotherscope"],
default_scopes=["myscope"],
),
required_scopes=["myscope"],
),
)
Server - 工具
https://gofastmcp.com/servers/tools
将功能作为可执行能力暴露给您的MCP客户端。
工具是核心构建块，允许您的LLM与外部系统交互、执行代码以及访问其训练数据之外的数据。在FastMCP中，工具是通过MCP协议暴露给LLM的Python函数。
工具是什么？
在FastMCP中，工具可以将常规的Python函数转换为LLMs在对话中可以调用的功能。当LLM决定使用一个工具时：
1、它发送一个基于工具模式的参数请求。
2、FastMCP验证这些参数是否符合您函数的签名。
3、您的函数使用验证后的输入执行。
4、结果返回给LLM，LLM可以使用它来生成响应。
这允许LLMs执行查询数据库、调用API、进行计算或访问文件等任务，扩展它们的能力，使其超越训练数据中的内容。
工具
@tool 装饰器
创建一个工具就像用 @mcp.tool() 装饰一个 Python 函数一样简单：
from fastmcp import FastMCP
mcp = FastMCP(name="CalculatorServer")
@mcp.tool()
def add(a: int, b: int) -> int:
"""Adds two integer numbers together."""
return a + b
参数
注解
参数的类型注解对于工具的正确功能至关重要。它们：
1、向LLM（大型语言模型）告知每个参数预期的数据类型
2、使FastMCP能够验证来自客户端的输入数据
3、为MCP协议生成准确的JSON模式
请使用标准的Python类型注解为参数进行注解：
@mcp.tool()
def analyze_text(
text: str, max_tokens: int = 100, language: str | None = None
) -> dict:
"""Analyze the provided text."""
# Implementation...
参数元数据
您可以使用 Pydantic 的 Field 类和 Annotated 来提供关于参数的额外元数据。这种方法更受青睐，因为它更现代，并且将类型提示与验证规则分开：
from typing import Annotated
from pydantic import Field
@mcp.tool()
def process_image(
image_url: Annotated[str, Field(description="URL of the image to process")], resize: Annotated[bool, Field(description="Whether to resize the image")] = False, width: Annotated[int, Field(description="Target width in pixels", ge=1, le=2000)] = 800, format: Annotated[
Literal["jpeg", "png", "webp"],
Field(description="Output image format")
] = "jpeg"
) -> dict:
"""Process an image with optional resizing."""
# Implementation...
@mcp.tool()
def search_database(
query: str = Field(description="Search query string"), limit: int = Field(10, description="Maximum number of results", ge=1, le=100)
) -> list:
"""Search the database with the provided query."""
# Implementation...
支持的类型
FastMCP 支持广泛的类型注解，包括所有 Pydantic 类型：
| 类型注解 | 示例 | 描述 |
|---|---|---|
| 基本类型 | int, float, str, bool | 简单标量值 - 请参阅 内置类型 |
| 二进制数据 | bytes | 二进制内容 - 请参阅 二进制数据 |
| 日期和时间 | datetime, date, timedelta | 日期和时间对象 - 请参阅 日期和时间类型 |
| 集合类型 | list[str], dict[str, int], set[int] | 项的集合 - 请参阅 集合类型 |
| 可选类型 | `float | None, Optional[float]` |
| 联合类型 | `str | int, Union[str, int]` |
| 限制类型 | Literal["A", "B"], Enum | 具有特定允许值的参数 - 请参阅 限制类型 |
| 路径 | Path | 文件系统路径 - 请参阅 路径 |
| UUIDs | UUID | 通用唯一标识符 - 请参阅 UUIDs |
| Pydantic 模型 | UserData | 复杂结构化数据 - 请参阅 Pydantic 模型 |
有关此处未列出的其他类型注解，请参阅下方的 参数类型 部分，以获取更详细的信息和示例。
可选参数
FastMCP遵循Python的标准函数参数约定。没有默认值的参数是必需的，而有默认值的参数是可选的。
@mcp.tool()
def search_products(
query: str, # Required - no default value
max_results: int = 10, # Optional - has default value
sort_by: str = "relevance", # Optional - has default value
category: str | None = None # Optional - can be None
) -> list[dict]:
"""Search the product catalog."""
# Implementation...
元数据
虽然FastMCP可以从您的函数中推断出名称和描述，但您可以使用@mcp.tool装饰器的参数来覆盖这些信息并添加标签：
@mcp.tool(
name="find_products", # Custom tool name for the LLM
description="Search the product catalog with optional category filtering.", # Custom description
tags={"catalog", "search"} # Optional tags for organization/filtering
)
def search_products_implementation(query: str, category: str | None = None) -> list[dict]:
"""Internal function description (ignored if description is provided above)."""
# Implementation...
print(f"Searching for '{query}' in category '{category}'")
return [{"id": 2, "name": "Another Product"}]
异步工具
FastMCP无缝支持标准（def）和异步（async def）函数作为工具。
# Synchronous tool (suitable for CPU-bound or quick tasks)
@mcp.tool()
def calculate_distance(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
"""Calculate the distance between two coordinates."""
# Implementation...
return 42.5
# Asynchronous tool (ideal for I/O-bound operations)
@mcp.tool()
async def fetch_weather(city: str) -> dict:
"""Retrieve current weather conditions for a city."""
# Use 'async def' for operations involving network calls, file I/O, etc.
# This prevents blocking the server while waiting for external operations.
async with aiohttp.ClientSession() as session:
async with session.get(f"https://api.example.com/weather/{city}") as response:
# Check response status before returning
response.raise_for_status()
return await response.json()
返回值
FastMCP会自动将您的函数返回的值转换为客户端适当的MCP内容格式：
str: 作为TextContent发送。dict,list, PydanticBaseModel: 序列化为JSON字符串，并作为TextContent发送。bytes: 使用Base64编码，并作为BlobResourceContents（通常在EmbeddedResource中）发送。fastmcp.Image: 一个用于轻松返回图像数据的辅助类。作为ImageContent发送。None: 导致空响应（不向客户端发送任何内容）。
FastMCP会尝试将其他类型序列化为字符串，如果可能的话。
目前，FastMCP只响应您的工具的返回值，而不是返回注释。
from fastmcp import FastMCP, Image
import io
try:
from PIL import Image as PILImage
except ImportError:
raise ImportError("Please install the `pillow` library to run this example.")
mcp = FastMCP("Image Demo")
@mcp.tool()
def generate_image(width: int, height: int, color: str) -> Image:
"""Generates a solid color image."""
# Create image using Pillow
img = PILImage.new("RGB", (width, height), color=color)
# Save to a bytes buffer
buffer = io.BytesIO()
img.save(buffer, format="PNG")
img_bytes = buffer.getvalue()
# Return using FastMCP's Image helper
return Image(data=img_bytes, format="png")
@mcp.tool()
def do_nothing() -> None:
"""This tool performs an action but returns no data."""
print("Performing a side effect...")
return None
错误处理
如果您的工具遇到错误，只需抛出一个标准的Python异常（ValueError、TypeError、FileNotFoundError、自定义异常等）。
@mcp.tool()
def divide(a: float, b: float) -> float:
"""Divide a by b."""
if b == 0:
# Raise a standard exception
raise ValueError("Division by zero is not allowed.")
if not isinstance(a, (int, float)) or not isinstance(b, (int, float)):
raise TypeError("Both arguments must be numbers.")
return a / b
标注
自版本 2.2.7 新增
FastMCP 允许您通过标注为您的工具添加专用元数据。这些标注用于向客户端应用程序传达工具的行为，而无需在 LLM 提示中消耗令牌上下文。
标注在客户端应用程序中具有多个用途：
- 添加用户友好的标题以供显示
- 指示工具是否修改数据或系统
- 描述工具的安全性配置文件（破坏性 vs. 非破坏性）
- 信号工具是否与外部系统交互
您可以使用 @mcp.tool() 装饰器中的 annotations 参数为工具添加标注：
@mcp.tool(
annotations={
"title": "Calculate Sum",
"readOnlyHint": True,
"openWorldHint": False
}
)
def calculate_sum(a: float, b: float) -> float:
"""Add two numbers together."""
return a + b
MCP 上下文
工具可以通过 Context 对象访问 MCP 功能，如日志记录、读取资源或报告进度。要使用它，请将类型提示 Context 添加到您的工具函数的参数中。
from fastmcp import FastMCP, Context
mcp = FastMCP(name="ContextDemo")
@mcp.tool()
async def process_data(data_uri: str, ctx: Context) -> dict:
"""Process data from a resource with progress reporting."""
await ctx.info(f"Processing data from {data_uri}")
# Read a resource
resource = await ctx.read_resource(data_uri)
data = resource[0].content if resource else ""
# Report progress
await ctx.report_progress(progress=50, total=100)
# Example request to the client's LLM for help
summary = await ctx.sample(f"Summarize this in 10 words: {data[:200]}")
await ctx.report_progress(progress=100, total=100)
return {
"length": len(data), "summary": summary.text
}
参数类型
FastMCP支持广泛的参数类型，以便您在设计工具时具有灵活性。
FastMCP通常支持Pydantic作为字段支持的所有类型，包括所有Pydantic自定义类型。这意味着您可以在工具参数中使用Pydantic可以验证和解析的任何类型。
当可能时，FastMCP支持类型转换。这意味着如果客户端发送的数据与预期类型不匹配，FastMCP将尝试将其转换为适当的类型。例如，如果客户端为一个标注为int的参数发送一个字符串，FastMCP将尝试将其转换为整数。如果转换不可行，FastMCP将返回一个验证错误。
内置类型
最常见的参数类型是 Python 的内置标量类型：
@mcp.tool()
def process_values(
name: str, # Text data
count: int, # Integer numbers
amount: float, # Floating point numbers
enabled: bool # Boolean values (True/False)
):
"""Process various value types."""
# Implementation...
日期和时间类型
FastMCP支持datetime模块中的各种日期和时间类型：
from datetime import datetime, date, timedelta
@mcp.tool()
def process_date_time(
event_date: date, # ISO format date string or date object
event_time: datetime, # ISO format datetime string or datetime object
duration: timedelta = timedelta(hours=1) # Integer seconds or timedelta
) -> str:
"""Process date and time information."""
# Types are automatically converted from strings
assert isinstance(event_date, date)
assert isinstance(event_time, datetime)
assert isinstance(duration, timedelta)
return f"Event on {event_date} at {event_time} for {duration}"
集合类型
FastMCP支持所有标准的Python集合类型：
@mcp.tool()
def analyze_data(
values: list[float], # List of numbers
properties: dict[str, str], # Dictionary with string keys and values
unique_ids: set[int], # Set of unique integers
coordinates: tuple[float, float], # Tuple with fixed structure
mixed_data: dict[str, list[int]] # Nested collections
):
"""Analyze collections of data."""
# Implementation...
联合和可选类型
对于可以接受多种类型或可能被省略的参数：
@mcp.tool()
def flexible_search(
query: str | int, # Can be either string or integer
filters: dict[str, str] | None = None, # Optional dictionary
sort_field: str | None = None # Optional string
):
"""Search with flexible parameter types."""
# Implementation...
限制类型
当一个参数必须是预定义值集中的一个时，你可以使用文字类型或枚举：
文字面量
文字面量将参数限制在特定的值集中：
from typing import Literal
@mcp.tool()
def sort_data(
data: list[float], order: Literal["ascending", "descending"] = "ascending", algorithm: Literal["quicksort", "mergesort", "heapsort"] = "quicksort"
):
"""Sort data using specific options."""
# Implementation...
枚举
为了使用更结构化的受限值集合，请使用 Python 的 Enum 类：
from enum import Enum
class Color(Enum):
RED = "red"
GREEN = "green"
BLUE = "blue"
@mcp.tool()
def process_image(
image_path: str,
color_filter: Color = Color.RED
):
"""Process an image with a color filter."""
# Implementation...
# color_filter will be a Color enum member
二进制数据
在工具参数中处理二进制数据有两种方法：
字节
@mcp.tool()
def process_binary(data: bytes):
"""Process binary data directly.
The client can send a binary string, which will be
converted directly to bytes.
"""
# Implementation using binary data
data_length = len(data)
# ...
基于Base64编码的字符串
from typing import Annotated
from pydantic import Field
@mcp.tool()
def process_image_data(
image_data: Annotated[str, Field(description="Base64-encoded image data")]
):
"""Process an image from base64-encoded string.
The client is expected to provide base64-encoded data as a string.
You'll need to decode it manually.
"""
# Manual base64 decoding
import base64
binary_data = base64.b64decode(image_data)
# Process binary_data...
路径
pathlib 模块中的 Path 类型可用于文件系统路径：
from pathlib import Path
@mcp.tool()
def process_file(path: Path) -> str:
"""Process a file at the given path."""
assert isinstance(path, Path) # Path is properly converted
return f"Processing file at {path}"
UUIDs
uuid模块中的UUID类型可用于唯一标识符：
import uuid
@mcp.tool()
def process_item(
item_id: uuid.UUID # String UUID or UUID object
) -> str:
"""Process an item with the given UUID."""
assert isinstance(item_id, uuid.UUID) # Properly converted to UUID
return f"Processing item {item_id}"
Pydantic 模型
对于具有嵌套字段和验证的复杂、结构化数据，请使用 Pydantic 模型：
from pydantic import BaseModel, Field
from typing import Optional
class User(BaseModel):
username: str
email: str = Field(description="User's email address")
age: int | None = None
is_active: bool = True
@mcp.tool()
def create_user(user: User):
"""Create a new user in the system."""
# The input is automatically validated against the User model
# Even if provided as a JSON string or dict
# Implementation...
Pydantic 字段
FastMCP 通过 Pydantic 的 Field 类支持强大的参数验证。这对于确保输入值满足特定要求，而不仅仅是它们的类型，非常有用。
请注意，字段可以在 Pydantic 模型之外使用，以提供元数据和验证约束。首选方法是使用 Annotated 与 Field：
from typing import Annotated
from pydantic import Field
@mcp.tool()
def analyze_metrics(
# Numbers with range constraints
count: Annotated[int, Field(ge=0, le=100)], # 0 <= count <= 100
ratio: Annotated[float, Field(gt=0, lt=1.0)], # 0 < ratio < 1.0
# String with pattern and length constraints
user_id: Annotated[str, Field(
pattern=r"^[A-Z]{2}\d{4}$", # Must match regex pattern
description="User ID in format XX0000"
)],
# String with length constraints
comment: Annotated[str, Field(min_length=3, max_length=500)] = "",
# Numeric constraints
factor: Annotated[int, Field(multiple_of=5)] = 10, # Must be multiple of 5
):
"""Analyze metrics with validated parameters."""
# Implementation...
@mcp.tool()
def validate_data(
# Value constraints
age: int = Field(ge=0, lt=120), # 0 <= age < 120
# String constraints
email: str = Field(pattern=r"^[\w.-]+@[\w.-]+.\w+$"), # Email pattern
# Collection constraints
tags: list[str] = Field(min_length=1, max_length=10) # 1-10 tags
):
"""Process data with field validations."""
# Implementation...
服务器行为
重复工具
自版本 2.1.0 新增
如果您尝试使用相同的名称注册多个工具，您可以控制 FastMCP 服务器的行为。这可以通过在创建 FastMCP 实例时使用 on_duplicate_tools 参数进行配置。
from fastmcp import FastMCP
mcp = FastMCP(
name="StrictServer", # Configure behavior for duplicate tool names
on_duplicate_tools="error"
)
@mcp.tool()
def my_tool(): return "Version 1"
# This will now raise a ValueError because 'my_tool' already exists
# and on_duplicate_tools is set to "error".
# @mcp.tool()
# def my_tool(): return "Version 2"
复制行为选项包括：
“warn”（默认）：记录警告，新工具将替换旧工具。“error”：引发ValueError，防止重复注册。“replace”：用新工具以静默方式替换现有工具。“ignore”：保留原始工具并忽略新的注册尝试。
Server - 资源 & 模板
https://gofastmcp.com/servers/resources
将数据源和动态内容生成器暴露给您的MCP客户端。
资源代表MCP客户端可以读取的数据或文件，资源模板通过允许客户端根据URI中传递的参数请求动态生成的资源来扩展这一概念。
FastMCP通过主要使用@mcp.resource装饰器简化了静态和动态资源的定义。
资源是什么？
资源为LLM或客户端应用程序提供对数据的只读访问。当客户端请求资源URI时：
1、FastMCP找到相应的资源定义。
2、如果它是动态的（由函数定义），则执行该函数。
3、将内容（文本、JSON、二进制数据）返回给客户端。
这允许LLM访问文件、数据库内容、配置或与对话相关的动态生成信息。
资源
@resource 装饰器
定义资源最常见的方式是通过装饰一个 Python 函数。该装饰器需要资源唯一的 URI。
import json
from fastmcp import FastMCP
mcp = FastMCP(name="DataServer")
# Basic dynamic resource returning a string
@mcp.resource("resource://greeting")
def get_greeting() -> str:
"""Provides a simple greeting message."""
return "Hello from FastMCP Resources!"
# Resource returning JSON data (dict is auto-serialized)
@mcp.resource("data://config")
def get_config() -> dict:
"""Provides application configuration as JSON."""
return {
"theme": "dark", "version": "1.2.0", "features": ["tools", "resources"], }
返回值
FastMCP会自动将您的函数返回值转换为适当的MCP资源内容：
str: 作为TextResourceContents发送（默认情况下，mime_type="text/plain"）。dict,list,pydantic.BaseModel: 自动序列化为JSON字符串，并作为TextResourceContents发送（默认情况下，mime_type="application/json"）。bytes: 使用Base64编码，并作为BlobResourceContents发送。您应指定适当的mime_type（例如，"image/png"，"application/octet-stream"）。None: 导致返回一个空资源内容列表。
资源元数据
您可以使用装饰器中的参数来自定义资源的属性：
from fastmcp import FastMCP
mcp = FastMCP(name="DataServer")
# Example specifying metadata
@mcp.resource(
uri="data://app-status", # Explicit URI (required)
name="ApplicationStatus", # Custom name
description="Provides the current status of the application.", # Custom description
mime_type="application/json", # Explicit MIME type
tags={"monitoring", "status"} # Categorization tags
)
def get_application_status() -> dict:
"""Internal function description (ignored if description is provided above)."""
return {"status": "ok", "uptime": 12345, "version": mcp.settings.version} # Example usage
访问 MCP 上下文
自版本 2.2.5 新增
资源和资源模板可以通过 Context 对象访问额外的 MCP 信息和功能。要访问它，请向您的资源函数添加一个类型注解为 Context 的参数：
from fastmcp import FastMCP, Context
mcp = FastMCP(name="DataServer")
@mcp.resource("resource://system-status")
async def get_system_status(ctx: Context) -> dict:
"""Provides system status information."""
return {
"status": "operational", "request_id": ctx.request_id
}
@mcp.resource("resource://{name}/details")
async def get_details(name: str, ctx: Context) -> dict:
"""Get details for a specific name."""
return {
"name": name, "accessed_at": ctx.request_id
}
异步资源
使用 async def 为执行 I/O 操作（例如，从数据库或网络读取）的资源函数，以避免阻塞服务器。
import aiofiles
from fastmcp import FastMCP
mcp = FastMCP(name="DataServer")
@mcp.resource("file:///app/data/important_log.txt", mime_type="text/plain")
async def read_important_log() -> str:
"""Reads content from a specific log file asynchronously."""
try:
async with aiofiles.open("/app/data/important_log.txt", mode="r") as f:
content = await f.read()
return content
except FileNotFoundError:
return "Log file not found."
资源类
虽然@mcp.resource非常适合动态内容，但您也可以使用mcp.add_resource()和具体的Resource子类直接注册预定义的资源（如静态文件或简单文本）。
from pathlib import Path
from fastmcp import FastMCP
from fastmcp.resources import FileResource, TextResource, DirectoryResource
mcp = FastMCP(name="DataServer")
# 1、Exposing a static file directly
readme_path = Path("./README.md").resolve()
if readme_path.exists():
# Use a file:// URI scheme
readme_resource = FileResource(
uri=f"file://{readme_path.as_posix()}", path=readme_path, # Path to the actual file
name="README File", description="The project's README.", mime_type="text/markdown", tags={"documentation"}
)
mcp.add_resource(readme_resource)
# 2、Exposing simple, predefined text
notice_resource = TextResource(
uri="resource://notice", name="Important Notice", text="System maintenance scheduled for Sunday.", tags={"notification"}
)
mcp.add_resource(notice_resource)
# 3、Using a custom key different from the URI
special_resource = TextResource(
uri="resource://common-notice", name="Special Notice", text="This is a special notice with a custom storage key.", )
mcp.add_resource(special_resource, key="resource://custom-key")
# 4、Exposing a directory listing
data_dir_path = Path("./app_data").resolve()
if data_dir_path.is_dir():
data_listing_resource = DirectoryResource(
uri="resource://data-files", path=data_dir_path, # Path to the directory
name="Data Directory Listing", description="Lists files available in the data directory.", recursive=False # Set to True to list subdirectories
)
mcp.add_resource(data_listing_resource) # Returns JSON list of files
自定义资源键
自 2.2.0 版本开始新增
当使用 mcp.add_resource() 直接添加资源时，您可以可选地提供一个自定义存储键：
# Creating a resource with standard URI as the key
resource = TextResource(uri="resource://data")
mcp.add_resource(resource) # Will be stored and accessed using "resource://data"
# Creating a resource with a custom key
special_resource = TextResource(uri="resource://special-data")
mcp.add_resource(special_resource, key="internal://data-v2") # Will be stored and accessed using "internal://data-v2"
资源模板
资源模板允许客户端请求内容依赖于URI中嵌入的参数的资源。使用相同的@mcp.resource装饰器定义模板，但在URI字符串中包含{parameter_name}占位符，并将相应的参数添加到你的函数签名中。
资源模板与常规资源共享大多数配置选项（名称、描述、mime_type、标签），但增加了定义映射到函数参数的URI参数的能力。
资源模板为每个唯一的参数集生成一个新的资源，这意味着资源可以按需动态创建。例如，如果注册了资源模板"user://profile/{name}"，MCP客户端可以请求"user://profile/ford"或"user://profile/marvin"来检索这两个用户配置文件作为资源，而无需单独注册每个资源。
以下是一个完整的示例，展示了如何定义两个资源模板：
from fastmcp import FastMCP
mcp = FastMCP(name="DataServer")
# Template URI includes {city} placeholder
@mcp.resource("weather://{city}/current")
def get_weather(city: str) -> dict:
"""Provides weather information for a specific city."""
# In a real implementation, this would call a weather API
# Here we're using simplified logic for example purposes
return {
"city": city.capitalize(), "temperature": 22, "condition": "Sunny", "unit": "celsius"
}
# Template with multiple parameters
@mcp.resource("repos://{owner}/{repo}/info")
def get_repo_info(owner: str, repo: str) -> dict:
"""Retrieves information about a GitHub repository."""
# In a real implementation, this would call the GitHub API
return {
"owner": owner, "name": repo, "full_name": f"{owner}/{repo}", "stars": 120, "forks": 48
}
通配符参数
自版本 2.2.4 新增
请注意：FastMCP 对通配符参数的支持是对模型上下文协议标准的扩展，而该标准本身遵循 RFC 6570。由于所有模板处理都在 FastMCP 服务器上完成，因此这不应与其他 MCP 实现造成任何兼容性问题。
资源模板支持通配符参数，可以匹配多个路径段。虽然标准参数（{param}）只能匹配单个路径段，并且不跨越“/”边界，但通配符参数（{param*}）可以捕获包括斜杠在内的多个段。通配符捕获所有后续的路径段，直到 URI 模板中定义的部分（无论是字面量还是另一个参数）。这允许你在单个 URI 模板中拥有多个通配符参数。
from fastmcp import FastMCP
mcp = FastMCP(name="DataServer")
# Standard parameter only matches one segment
@mcp.resource("files://{filename}")
def get_file(filename: str) -> str:
"""Retrieves a file by name."""
# Will only match files://<single-segment>
return f"File content for: {filename}"
# Wildcard parameter can match multiple segments
@mcp.resource("path://{filepath*}")
def get_path_content(filepath: str) -> str:
"""Retrieves content at a specific path."""
# Can match path://docs/server/resources.mdx
return f"Content at path: {filepath}"
# Mixing standard and wildcard parameters
@mcp.resource("repo://{owner}/{path*}/template.py")
def get_template_file(owner: str, path: str) -> dict:
"""Retrieves a file from a specific repository and path, but
only if the resource ends with `template.py`"""
# Can match repo://jlowin/fastmcp/src/resources/template.py
return {
"owner": owner, "path": path + "/template.py", "content": f"File at {path}/template.py in {owner}'s repository"
}
默认值
自版本 2.2.0 新增
在创建资源模板时，FastMCP 对 URI 模板参数与函数参数之间的关系强制执行两条规则：
1、必需的函数参数： 所有没有默认值（必需参数）的函数参数必须出现在 URI 模板中。
2、URI 参数： 所有 URI 模板参数都必须作为函数参数存在。
然而，具有默认值的函数参数不需要包含在 URI 模板中。当客户端请求资源时，FastMCP 将：
- 从模板中包含的参数的 URI 中提取参数值
- 对于不在 URI 模板中的任何函数参数使用默认值
这允许灵活的 API 设计。例如，一个具有可选参数的简单搜索模板：
from fastmcp import FastMCP
mcp = FastMCP(name="DataServer")
@mcp.resource("search://{query}")
def search_resources(query: str, max_results: int = 10, include_archived: bool = False) -> dict:
"""Search for resources matching the query string."""
# Only 'query' is required in the URI, the other parameters use their defaults
results = perform_search(query, limit=max_results, archived=include_archived)
return {
"query": query, "max_results": max_results, "include_archived": include_archived, "results": results
}
from fastmcp import FastMCP
mcp = FastMCP(name="DataServer")
# Define a user lookup function that can be accessed by different identifiers
@mcp.resource("users://email/{email}")
@mcp.resource("users://name/{name}")
def lookup_user(name: str | None = None, email: str | None = None) -> dict:
"""Look up a user by either name or email."""
if email:
return find_user_by_email(email) # pseudocode
elif name:
return find_user_by_name(name) # pseudocode
else:
return {"error": "No lookup parameters provided"}
服务器行为
重复资源
自版本 2.1.0 新增
您可以在初始化 FastMCP 服务器时配置服务器如何处理尝试使用相同 URI 注册多个资源或模板的情况。使用 on_duplicate_resources 设置。
from fastmcp import FastMCP
mcp = FastMCP(
name="ResourceServer", on_duplicate_resources="error" # Raise error on duplicates
)
@mcp.resource("data://config")
def get_config_v1(): return {"version": 1}
# This registration attempt will raise a ValueError because
# "data://config" is already registered and the behavior is "error".
# @mcp.resource("data://config")
# def get_config_v2(): return {"version": 2}
Server - 提示
https://gofastmcp.com/servers/prompts
为MCP客户端创建可重用、参数化的提示模板。
提示是可重用的消息模板，有助于LLM生成结构化、有目的的响应。FastMCP简化了这些模板的定义，主要使用@mcp.prompt装饰器。
提示是什么？
提示为LLM提供参数化的消息模板。当客户端请求一个提示时：
1、FastMCP找到相应的提示定义。
2、如果它有参数，它们将与您的函数签名进行验证。
3、您的函数使用验证后的输入执行。
4、生成的消息返回给LLM以指导其响应。
这允许您定义一致、可重用的模板，LLM可以在不同的客户端和上下文中使用。
提示
@prompt 装饰器
定义提示的最常见方式是通过装饰一个Python函数。装饰器使用函数名作为提示的标识符。
from fastmcp import FastMCP
from fastmcp.prompts.prompt import UserMessage, AssistantMessage, Message
mcp = FastMCP(name="PromptServer")
# Basic prompt returning a string (converted to UserMessage)
@mcp.prompt()
def ask_about_topic(topic: str) -> str:
"""Generates a user message asking for an explanation of a topic."""
return f"Can you please explain the concept of '{topic}'?"
# Prompt returning a specific message type
@mcp.prompt()
def generate_code_request(language: str, task_description: str) -> UserMessage:
"""Generates a user message requesting code generation."""
content = f"Write a {language} function that performs the following task: {task_description}"
return UserMessage(content=content)
返回值
FastMCP智能处理来自您的提示函数的不同返回类型：
str: 自动转换为单个UserMessage。Message（例如，UserMessage，AssistantMessage）：直接使用提供的内容。dict: 如果具有正确的结构，则解析为Message对象。list[Message]: 作为一系列消息（一个对话）使用。
@mcp.prompt()
def roleplay_scenario(character: str, situation: str) -> list[Message]:
"""Sets up a roleplaying scenario with initial messages."""
return [
UserMessage(f"Let's roleplay. You are {character}. The situation is: {situation}"), AssistantMessage("Okay, I understand. I am ready. What happens next?")
]
@mcp.prompt()
def ask_for_feedback() -> dict:
"""Generates a user message asking for feedback."""
return {"role": "user", "content": "What did you think of my previous response?"}
类型注解
类型注解对于提示非常重要。它们：
1、通知FastMCP每个参数的预期类型。
2、允许验证来自客户端的参数。
3、用于生成MCP协议中提示的架构。
from pydantic import Field
from typing import Literal, Optional
@mcp.prompt()
def generate_content_request(
topic: str = Field(description="The main subject to cover"), format: Literal["blog", "email", "social"] = "blog", tone: str = "professional", word_count: Optional[int] = None
) -> str:
"""Create a request for generating content in a specific format."""
prompt = f"Please write a {format} post about {topic} in a {tone} tone."
if word_count:
prompt += f" It should be approximately {word_count} words long."
return prompt
必需参数与可选参数
在函数签名中的参数被认为是必需的，除非它们有默认值。
@mcp.prompt()
def data_analysis_prompt(
data_uri: str, # Required - no default value
analysis_type: str = "summary", # Optional - has default value
include_charts: bool = False # Optional - has default value
) -> str:
"""Creates a request to analyze data with specific parameters."""
prompt = f"Please perform a '{analysis_type}' analysis on the data found at {data_uri}."
if include_charts:
prompt += " Include relevant charts and visualizations."
return prompt
提示元数据
虽然FastMCP可以从您的函数中推断出名称和描述，但您可以使用@mcp.prompt装饰器的参数来覆盖这些内容并添加标签：
@mcp.prompt(
name="analyze_data_request", # Custom prompt name
description="Creates a request to analyze data with specific parameters", # Custom description
tags={"analysis", "data"} # Optional categorization tags
)
def data_analysis_prompt(
data_uri: str = Field(description="The URI of the resource containing the data."), analysis_type: str = Field(default="summary", description="Type of analysis.")
) -> str:
"""This docstring is ignored when description is provided."""
return f"Please perform a '{analysis_type}' analysis on the data found at {data_uri}."
异步提示
FastMCP无缝支持标准（def）和异步（async def）函数作为提示。
# Synchronous prompt
@mcp.prompt()
def simple_question(question: str) -> str:
"""Generates a simple question to ask the LLM."""
return f"Question: {question}"
# Asynchronous prompt
@mcp.prompt()
async def data_based_prompt(data_id: str) -> str:
"""Generates a prompt based on data that needs to be fetched."""
# In a real scenario, you might fetch data from a database or API
async with aiohttp.ClientSession() as session:
async with session.get(f"https://api.example.com/data/{data_id}") as response:
data = await response.json()
return f"Analyze this data: {data['content']}"
访问 MCP 上下文
新增于版本：2.2.5
提示可以通过 Context 对象访问额外的 MCP 信息和功能。要访问它，请向您的提示函数添加一个参数，并使用 Context 类型注解：
from fastmcp import FastMCP, Context
mcp = FastMCP(name="PromptServer")
@mcp.prompt()
async def generate_report_request(report_type: str, ctx: Context) -> str:
"""Generates a request for a report."""
return f"Please create a {report_type} report. Request ID: {ctx.request_id}"
服务器行为
重复提示
自版本 2.1.0 新增
您可以通过配置 FastMCP 服务器如何处理尝试使用相同名称注册多个提示的操作。在 FastMCP 初始化期间使用 on_duplicate_prompts 设置。
from fastmcp import FastMCP
mcp = FastMCP(
name="PromptServer", on_duplicate_prompts="error" # Raise an error if a prompt name is duplicated
)
@mcp.prompt()
def greeting(): return "Hello, how can I help you today?"
# This registration attempt will raise a ValueError because
# "greeting" is already registered and the behavior is "error".
# @mcp.prompt()
# def greeting(): return "Hi there! What can I do for you?"
Server - MCP 上下文
https://gofastmcp.com/servers/context
在您的 MCP 对象中访问日志、进度和资源等功能。
当定义 FastMCP 工具、资源、资源模板或 提示 时，您的函数可能需要与底层 MCP 会话交互或访问服务器功能。FastMCP 提供了 Context 对象来实现此目的。
什么是上下文？
Context 对象提供了一个干净的接口，用于在您的函数中访问MCP功能，包括：
- 日志记录：向客户端发送调试、信息、警告和错误消息
- 进度报告：更新客户端关于长时间运行操作进度的信息
- 资源访问：从与服务器注册的资源中读取数据
- LLM采样：请求客户端的LLM根据提供的消息生成文本
- 请求信息：访问当前请求的元数据
- 服务器访问：当需要时，访问底层的FastMCP服务器实例
访问上下文
要在您的任何函数中使用上下文对象，只需在函数签名中添加一个参数，并将其类型提示为Context。当您的函数被调用时，FastMCP将自动注入上下文实例。
from fastmcp import FastMCP, Context
mcp = FastMCP(name="ContextDemo")
@mcp.tool()
async def process_file(file_uri: str, ctx: Context) -> str:
"""Processes a file, using context for logging and resource access."""
request_id = ctx.request_id
await ctx.info(f"[{request_id}] Starting processing for {file_uri}")
try:
# Use context to read a resource
contents_list = await ctx.read_resource(file_uri)
if not contents_list:
await ctx.warning(f"Resource {file_uri} is empty.")
return "Resource empty"
data = contents_list[0].content # Assuming TextResourceContents
await ctx.debug(f"Read {len(data)} bytes from {file_uri}")
# Report progress
await ctx.report_progress(progress=50, total=100)
# Simulate work
processed_data = data.upper() # Example processing
await ctx.report_progress(progress=100, total=100)
await ctx.info(f"Processing complete for {file_uri}")
return f"Processed data length: {len(processed_data)}"
except Exception as e:
# Use context to log errors
await ctx.error(f"Error processing {file_uri}: {str(e)}")
raise # Re-raise to send error back to client
Key Points:
- 参数名（例如
ctx，context）并不重要，重要的是类型提示Context。 - 上下文参数可以放置在函数签名的任何位置。
- 上下文是可选的-不需要它的函数可以省略参数。
- 上下文仅在请求期间可用；尝试在请求外部使用上下文方法将引发错误。
- 上下文方法是异步的，所以您的函数通常也需要是异步的。
上下文能力
日志记录
将日志消息发送回MCP客户端。这在调试和提供对请求期间函数执行的可视化方面非常有用。
@mcp.tool()
async def analyze_data(data: list[float], ctx: Context) -> dict:
"""Analyze numerical data with logging."""
await ctx.debug("Starting analysis of numerical data")
await ctx.info(f"Analyzing {len(data)} data points")
try:
result = sum(data) / len(data)
await ctx.info(f"Analysis complete, average: {result}")
return {"average": result, "count": len(data)}
except ZeroDivisionError:
await ctx.warning("Empty data list provided")
return {"error": "Empty data list"}
except Exception as e:
await ctx.error(f"Analysis failed: {str(e)}")
raise
Available Logging Methods:
ctx.debug(message: str): 用于调试的低级详细信息ctx.info(message: str): 执行的一般信息ctx.warning(message: str): 未阻止执行的潜在问题ctx.error(message: str): 执行期间发生的错误ctx.log(level: Literal["debug", "info", "warning", "error"], message: str, logger_name: str | None = None): 支持自定义记录器名称的通用日志方法
进度报告
对于长时间运行的操作，通知客户端进度情况。这允许客户端显示进度指示器，并提供更好的用户体验。
@mcp.tool()
async def process_items(items: list[str], ctx: Context) -> dict:
"""Process a list of items with progress updates."""
total = len(items)
results = []
for i, item in enumerate(items):
# Report progress as percentage
await ctx.report_progress(progress=i, total=total)
# Process the item (simulated with a sleep)
await asyncio.sleep(0.1)
results.append(item.upper())
# Report 100% completion
await ctx.report_progress(progress=total, total=total)
return {"processed": len(results), "results": results}
Method signature:
ctx.report_progress(progress: float, total: float | None = None)progress: 当前进度值 (e.g., 24)total: 可选总值（例如，100）。如果提供，客户可以将其解释为百分比。
进度报告要求客户端在初始请求中已发送 progressToken。如果客户端不支持进度报告，这些调用将无效。
资源访问
从与您的FastMCP服务器注册的资源中读取数据。这允许函数访问文件、配置或动态生成的内容。
@mcp.tool()
async def summarize_document(document_uri: str, ctx: Context) -> str:
"""Summarize a document by its resource URI."""
# Read the document content
content_list = await ctx.read_resource(document_uri)
if not content_list:
return "Document is empty"
document_text = content_list[0].content
# Example: Generate a simple summary (length-based)
words = document_text.split()
total_words = len(words)
await ctx.info(f"Document has {total_words} words")
# Return a simple summary
if total_words > 100:
summary = " ".join(words[:100]) + "..."
return f"Summary ({total_words} words total): {summary}"
else:
return f"Full document ({total_words} words): {document_text}"
Method signature:
ctx.read_resource(uri: str | AnyUrl) -> list[ReadResourceContents]uri: 要读取的资源 URI- 返回资源内容部分的列表（通常仅包含一项）
返回的内容通常通过 content_list[0].content 访问，并且可以是文本或二进制数据，具体取决于资源。
LLM Sampling
自版本 2.0.0 新增
请求客户端的LLM根据提供的消息生成文本。这在您的函数需要利用LLM的能力来处理数据或生成响应时非常有用。
@mcp.tool()
async def analyze_sentiment(text: str, ctx: Context) -> dict:
"""Analyze the sentiment of a text using the client's LLM."""
# Create a sampling prompt asking for sentiment analysis
prompt = f"Analyze the sentiment of the following text as positive, negative, or neutral. Just output a single word - 'positive', 'negative', or 'neutral'. Text to analyze: {text}"
# Send the sampling request to the client's LLM
response = await ctx.sample(prompt)
# Process the LLM's response
sentiment = response.text.strip().lower()
# Map to standard sentiment values
if "positive" in sentiment:
sentiment = "positive"
elif "negative" in sentiment:
sentiment = "negative"
else:
sentiment = "neutral"
return {"text": text, "sentiment": sentiment}
方法签名：
ctx.sample(messages: str | list[str | SamplingMessage], system_prompt: str | None = None, temperature: float | None = None, max_tokens: int | None = None) -> TextContent | ImageContentmessages：要发送到LLM的字符串或字符串/消息对象列表system_prompt：用于指导LLM行为的可选系统提示temperature：可选采样温度（控制随机性）max_tokens：可选的要生成的最大令牌数（默认为512）- 将LLM的响应返回为 TextContent 或 ImageContent
当提供一个简单字符串时，它被视为用户消息。对于更复杂的场景，可以提供具有不同角色的消息列表。
@mcp.tool()
async def generate_example(concept: str, ctx: Context) -> str:
"""Generate a Python code example for a given concept."""
# Using a system prompt and a user message
response = await ctx.sample(
messages=f"Write a simple Python code example demonstrating '{concept}'.", system_prompt="You are an expert Python programmer. Provide concise, working code examples without explanations.", temperature=0.7, max_tokens=300
)
code_example = response.text
return f"```python\n{code_example}\n```"
参见 Client Sampling Client Sampling 获取有关客户端如何处理这些请求的更多详细信息。
请求信息
访问当前请求和客户端的元数据。
@mcp.tool()
async def request_info(ctx: Context) -> dict:
"""Return information about the current request."""
return {
"request_id": ctx.request_id,
"client_id": ctx.client_id or "Unknown client"
}
Available Properties:
ctx.request_id -> str: Get the unique ID for the current MCP requestctx.client_id -> str | None: Get the ID of the client making the request, if provided during initialization
高级访问
对于高级用例，您可以访问底层MCP会话、FastMCP服务器和HTTP请求。
访问FastMCP和会话
@mcp.tool()
async def advanced_tool(ctx: Context) -> str:
"""Demonstrate advanced context access."""
# Access the FastMCP server instance
server_name = ctx.fastmcp.name
# Low-level session access (rarely needed)
session = ctx.session
request_context = ctx.request_context
return f"Server: {server_name}"
访问 HTTP 请求
自版本 2.2.7 新增
对于网络应用程序，您可以访问底层的 HTTP 请求：
@mcp.tool()
async def handle_web_request(ctx: Context) -> dict:
"""Access HTTP request information from the Starlette request."""
request = ctx.get_http_request()
# Access HTTP headers, query parameters, etc.
user_agent = request.headers.get("user-agent", "Unknown")
client_ip = request.client.host if request.client else "Unknown"
return {
"user_agent": user_agent, "client_ip": client_ip, "path": request.url.path, }
高级属性参考
ctx.fastmcp -> FastMCP: 访问上下文所属的服务器实例ctx.session: 访问原始的mcp.server.session.ServerSession对象ctx.request_context: 访问原始的mcp.shared.context.RequestContext对象ctx.get_http_request() -> Request: 访问活动的 Starlette 请求对象（当与 Web 服务器一起运行时）
直接使用 session 或 request_context 需要理解低级 MCP Python SDK，可能比直接在 Context 对象上使用的方法稳定性更低。
在不同组件中使用上下文
所有FastMCP组件（工具、资源、模板和提示）都可以使用上下文对象，遵循相同的模式 - 只需添加一个带有Context类型注解的参数。
资源和模板中的上下文
资源和资源模板可以访问上下文以自定义其行为：
@mcp.resource("resource://user-data")
async def get_user_data(ctx: Context) -> dict:
"""Fetch personalized user data based on the request context."""
user_id = ctx.client_id or "anonymous"
await ctx.info(f"Fetching data for user {user_id}")
# Example of using context for dynamic resource generation
return {
"user_id": user_id, "last_access": datetime.now().isoformat(), "request_id": ctx.request_id
}
@mcp.resource("resource://users/{user_id}/profile")
async def get_user_profile(user_id: str, ctx: Context) -> dict:
"""Fetch user profile from database with context-aware logging."""
await ctx.info(f"Fetching profile for user {user_id}")
# Example of using context in a template resource
# In a real implementation, you might query a database
return {
"id": user_id,
"name": f"User {user_id}",
"request_id": ctx.request_id
}
提示中的上下文
提示可以使用上下文来生成更动态的模板：
@mcp.prompt()
async def data_analysis_request(dataset: str, ctx: Context) -> str:
"""Generate a request to analyze data with contextual information."""
await ctx.info(f"Generating data analysis prompt for {dataset}")
# Could use context to read configuration or personalize the prompt
return f"""Please analyze the following dataset: {dataset}
Request initiated at: {datetime.now().isoformat()}
Request ID: {ctx.request_id}
"""
现在，所有FastMCP对象都支持使用相同的一致模式进行上下文注入，从而可以轻松地将会话感知功能添加到MCP服务器的各个方面。
客户端概述
https://gofastmcp.com/clients/client
了解如何使用FastMCP客户端与MCP服务器交互。
新功能自版本：2.0.0开始
fastmcp.Client提供了一个高级的、异步的接口，用于与任何模型上下文协议（MCP）服务器进行交互，无论它是使用FastMCP还是其他实现构建的。它通过处理协议细节和连接管理来简化通信。
快速MCP客户端
FastMCP客户端架构将协议逻辑（Client）与连接机制（Transport）分离。
Client：处理发送MCP请求（如tools/call、resources/read），接收响应和管理回调。Transport：负责建立和维护与服务器的连接（例如，通过WebSockets、SSE、Stdio或在内存中）。
传输
客户端必须使用一个 transport 进行初始化。您可以选择提供一个已经实例化的传输对象，或者提供一个传输源，让 FastMCP 尝试推断正确的传输方式来使用。
以下推断规则用于根据输入类型确定适当的 ClientTransport：
1、ClientTransport 实例：如果您提供一个已经实例化的传输对象，它将被直接使用。
2、FastMCP 实例：创建一个 FastMCPTransport 以实现高效的内存通信（理想用于测试）。
3、指向现有文件的 Path 或 str：
- 如果以
.py结尾：创建一个PythonStdioTransport来使用python运行脚本。 - 如果以
.js结尾：创建一个NodeStdioTransport来使用node运行脚本。
4、指向 URL 的 AnyUrl 或 str：
- 如果以
http://或https://开头：创建一个SSETransport。 - 如果以
ws://或wss://开头：创建一个WSTransport。
5、其他：如果无法推断类型，将引发一个 ValueError。
import asyncio
from fastmcp import Client, FastMCP
# Example transports (more details in Transports page)
server_instance = FastMCP(name="TestServer") # In-memory server
sse_url = "http://localhost:8000/sse" # SSE server URL
ws_url = "ws://localhost:9000" # WebSocket server URL
server_script = "my_mcp_server.py" # Path to a Python server file
# Client automatically infers the transport type
client_in_memory = Client(server_instance)
client_sse = Client(sse_url)
client_ws = Client(ws_url)
client_stdio = Client(server_script)
print(client_in_memory.transport)
print(client_sse.transport)
print(client_ws.transport)
print(client_stdio.transport)
# Expected Output (types may vary slightly based on environment):
# <FastMCP(server='TestServer')>
# <SSE(url='http://localhost:8000/sse')>
# <WebSocket(url='ws://localhost:9000')>
# <PythonStdioTransport(command='python', args=['/path/to/your/my_mcp_server.py'])>
客户端使用
连接生命周期
客户端以异步方式运行，必须在 async with 块中使用。此上下文管理器负责建立连接、初始化MCP会话以及在退出时清理资源。
import asyncio
from fastmcp import Client
client = Client("my_mcp_server.py") # Assumes my_mcp_server.py exists
async def main():
# Connection is established here
async with client:
print(f"Client connected: {client.is_connected()}")
# Make MCP calls within the context
tools = await client.list_tools()
print(f"Available tools: {tools}")
if any(tool.name == "greet" for tool in tools):
result = await client.call_tool("greet", {"name": "World"})
print(f"Greet result: {result}")
# Connection is closed automatically here
print(f"Client connected: {client.is_connected()}")
if __name__ == "__main__":
asyncio.run(main())
您可以使用已建立的会话在同一个 async with 块中对服务器进行多次调用。
客户端方法
Client 提供与标准 MCP 请求相对应的方法：
工具操作
list_tools(): 获取服务器上可用的工具列表。
tools = await client.list_tools()
# tools -> list[mcp.types.Tool]
call_tool(name: str, arguments: dict[str, Any] | None = None): 在 server 上执行一个工具
result = await client.call_tool("add", {"a": 5, "b": 3})
# result -> list[mcp.types.TextContent | mcp.types.ImageContent | ...]
print(result[0].text) # Assuming TextContent, e.g., '8'
- 参数作为字典传递。如果需要，FastMCP服务器会自动处理复杂类型的JSON字符串解析。
- 返回内容对象列表（通常为
TextContent或ImageContent）。
资源操作
list_resources(): 获取静态资源列表。
resources = await client.list_resources()
# resources -> list[mcp.types.Resource]
list_resource_templates(): 获取一列资源模板
templates = await client.list_resource_templates()
# templates -> list[mcp.types.ResourceTemplate]
read_resource(uri: str | AnyUrl): 读取资源或已解析模板的内容。
# Read a static resource
readme_content = await client.read_resource("file:///path/to/README.md")
# readme_content -> list[mcp.types.TextResourceContents | mcp.types.BlobResourceContents]
print(readme_content[0].text) # Assuming text
# Read a resource generated from a template
weather_content = await client.read_resource("data://weather/london")
print(weather_content[0].text) # Assuming text JSON
Prompt 操作
list_prompts(): 获取可用的提示模板列表。get_prompt(name: str, arguments: dict[str, Any] | None = None): 获取渲染后的提示信息列表。
原始 MCP 协议对象
FastMCP 客户端试图提供一个“友好”的接口来访问 MCP 协议，但有时您可能需要访问原始 MCP 协议对象。每个主要客户端方法，当返回数据时，都有一个对应的 *_mcp 方法，该方法直接返回原始 MCP 协议对象。
# Standard method - returns just the list of tools
tools = await client.list_tools()
# tools -> list[mcp.types.Tool]
# Raw MCP method - returns the full protocol object
result = await client.list_tools_mcp()
# result -> mcp.types.ListToolsResult
tools = result.tools
可用 MCP 原生方法
list_tools_mcp(): Returnsmcp.types.ListToolsResultcall_tool_mcp(name, arguments): Returnsmcp.types.CallToolResultlist_resources_mcp(): Returnsmcp.types.ListResourcesResultlist_resource_templates_mcp(): Returnsmcp.types.ListResourceTemplatesResultread_resource_mcp(uri): Returnsmcp.types.ReadResourceResultlist_prompts_mcp(): Returnsmcp.types.ListPromptsResultget_prompt_mcp(name, arguments): Returnsmcp.types.GetPromptResultcomplete_mcp(ref, argument): Returnsmcp.types.CompleteResult
这些方法对于调试或需要访问简化方法未公开的元数据或字段时特别有用。
高级功能
MCP 允许服务器与客户端交互，以提供额外的功能。Client 构造函数接受额外的配置来处理这些服务器请求。
LLM Sampling
MCP 服务器可以从客户端请求 LLM 补充内容。客户端可以提供一个 sampling_handler 来处理这些请求。采样处理程序从服务器接收一条消息列表和其他参数，并应返回一个字符串补充内容。
以下示例使用 marvin 库来生成补充内容：
import marvin
from fastmcp import Client
from fastmcp.client.sampling import (
SamplingMessage, SamplingParams, RequestContext, )
async def sampling_handler(
messages: list[SamplingMessage], params: SamplingParams, context: RequestContext
) -> str:
return await marvin.say_async(
message=[m.content.text for m in messages], instructions=params.systemPrompt, )
client = Client(
..., sampling_handler=sampling_handler, )
日志记录
MCP服务器可以向客户端发送日志。客户端可以设置一个日志回调来接收这些日志。
from fastmcp import Client
from fastmcp.client.logging import LogHandler, LogMessage
async def my_log_handler(params: LogMessage):
print(f"[Server Log - {params.level.upper()}] {params.logger or 'default'}: {params.data}")
client_with_logging = Client(
..., log_handler=my_log_handler, )
根
根是客户端通知服务器它们可以访问的资源或它们访问的某些边界的一种方式。服务器可以使用这些信息来调整行为或提供更准确的响应。
服务器可以从客户端请求根，客户端可以在它们的根发生变化时通知服务器。
在创建客户端时设置根，用户可以提供根的列表（可以是字符串列表）或一个返回根列表的异步函数。
静态根
from fastmcp import Client
client = Client(
...,
roots=["/path/to/root1", "/path/to/root2"], )
动态根回调
from fastmcp import Client
from fastmcp.client.roots import RequestContext
async def roots_callback(context: RequestContext) -> list[str]:
print(f"Server requested roots (Request ID: {context.request_id})")
return ["/path/to/root1", "/path/to/root2"]
client = Client(
...,
roots=roots_callback, )
工具方法
ping(): 向服务器发送ping请求以验证连接性。
async def check_connection():
async with client:
await client.ping()
print("Server is reachable")
错误处理
当call_tool请求在服务器上导致错误（例如，工具函数抛出异常）时，client.call_tool()方法将引发一个fastmcp.client.ClientError。
async def safe_call_tool():
async with client:
try:
# Assume 'divide' tool exists and might raise ZeroDivisionError
result = await client.call_tool("divide", {"a": 10, "b": 0})
print(f"Result: {result}")
except ClientError as e:
print(f"Tool call failed: {e}")
except ConnectionError as e:
print(f"Connection failed: {e}")
except Exception as e:
print(f"An unexpected error occurred: {e}")
# Example Output if division by zero occurs:
# Tool call failed: Division by zero is not allowed.
其他错误，例如连接失败，将引发标准的 Python 异常（例如，ConnectionError，TimeoutError）。
客户端传输通常有其自身的错误处理机制，因此您不能总是捕获像由 call_tool 引发的错误那样的错误，这些错误发生在 async with 块之外。
相反，您可以使用 call_tool_mcp() 来获取原始的 mcp.types.CallToolResult 对象，并通过检查其 isError 属性来自行处理错误。
客户端 - 传输
https://gofastmcp.com/clients/transports
了解FastMCP客户端连接到服务器的不同方式。
新功能自版本：2.0.0
FastMCP 客户端依赖于ClientTransport对象来处理连接到MCP服务器以及与MCP服务器通信的细节。FastMCP为常见的连接方法提供了几个内置的传输实现。
虽然客户端通常会自动推断正确的传输方式（参见客户端概述），但您也可以显式实例化传输以获得更多控制。
标准输入输出传输
这些传输管理一个作为子进程运行的多通道协议（MCP）服务器，通过标准输入（stdin）和标准输出（stdout）与它进行通信。这是Claude桌面等客户端使用的标准机制。
Python Stdio
- 类:
fastmcp.client.transports.PythonStdioTransport - 推断自:
.py文件路径。 - 用例: 在子进程中运行基于 Python 的 MCP 服务器脚本（如使用 FastMCP 或基本
mcp库的脚本）。
这是在开发期间或与期望启动服务器脚本的工具集成时与本地 FastMCP 服务器交互的最常见方式。
from fastmcp import Client
from fastmcp.client.transports import PythonStdioTransport
server_script = "my_mcp_server.py" # Assumes this file exists and runs mcp.run()
# Option 1: Inferred transport
client_inferred = Client(server_script)
# Option 2: Explicit transport (e.g., to use a specific python executable or add args)
transport_explicit = PythonStdioTransport(
script_path=server_script, python_cmd="/usr/bin/python3.11", # Specify python version
# args=["--some-server-arg"], # Pass args to the script
# env={"MY_VAR": "value"}, # Set environment variables
# cwd="/path/to/run/in" # Set working directory
)
client_explicit = Client(transport_explicit)
async def use_stdio_client(client):
async with client:
tools = await client.list_tools()
print(f"Connected via Python Stdio, found tools: {tools}")
# asyncio.run(use_stdio_client(client_inferred))
# asyncio.run(use_stdio_client(client_explicit))
服务器脚本（示例中的 my_mcp_server.py）必须包含启动 MCP 服务器并监听标准输入输出的逻辑，通常通过 mcp.run() 或 fastmcp.server.run() 实现。客户端仅启动该脚本；它不会注入服务器逻辑。
Node.js Stdio
- 类:
fastmcp.client.transports.NodeStdioTransport - 推断自:
.js文件路径。 - 用例: 在子进程中运行基于 Node.js 的 MCP 服务器脚本。
类似于 Python 传输，但用于 JavaScript 服务器。
from fastmcp import Client
from fastmcp.client.transports import NodeStdioTransport
node_server_script = "my_mcp_server.js" # Assumes this JS file starts an MCP server on stdio
# Option 1: Inferred transport
client_inferred = Client(node_server_script)
# Option 2: Explicit transport
transport_explicit = NodeStdioTransport(
script_path=node_server_script, node_cmd="node" # Or specify path to Node executable
)
client_explicit = Client(transport_explicit)
# Usage is the same as other clients
# async with client_explicit:
# tools = await client_explicit.list_tools()
UVX Stdio (Experimental)
- 类:
fastmcp.client.transports.UvxStdioTransport - 推断来源: 无法自动推断。必须显式实例化。
- 使用场景: 使用
uvx（uv工具链的一部分）将MCP服务器打包为Python工具运行。这允许在不显式安装到当前环境的情况下运行工具。
这对于执行作为命令行工具或软件包分发的MCP服务器非常有用。
from fastmcp.client.transports import UvxStdioTransport
# Example: Run a hypothetical 'cloud-analyzer-mcp' tool via uvx
# Assume this tool, when run, starts an MCP server on stdio
transport = UvxStdioTransport(
tool_name="cloud-analyzer-mcp", # from_package="cloud-analyzer-cli", # Optionally specify package if tool name differs
# with_packages=["boto3", "requests"], # Add dependencies if needed
# tool_args=["--config", "prod.yaml"] # Pass args to the tool itself
)
client = Client(transport)
# async with client:
# analysis = await client.call_tool("analyze_bucket", {"name": "my-data"})
NPX Stdio (Experimental)
- 类名:
fastmcp.client.transports.NpxStdioTransport - 推断来源: 无法自动推断。必须显式实例化。
- 使用场景: 使用
npx运行打包为 NPM 包的 MCP 服务器。
与 UvxStdioTransport 类似，但适用于 Node.js 生态系统。
from fastmcp.client.transports import NpxStdioTransport
# Example: Run a hypothetical 'npm-mcp-server-package' via npx
transport = NpxStdioTransport(
package="npm-mcp-server-package", # args=["--port", "stdio"] # Args passed to the package script
)
client = Client(transport)
# async with client:
# response = await client.call_tool("get_npm_data", {})
网络传输
这些传输连接到通过网络运行的服务器，通常是可以通过URL访问的长期运行的服务。
SSE (Server-Sent Events)
- 类:
fastmcp.client.transports.SSETransport - 推断来源:
http://或https://URL - 用例: 连接到通过 HTTP/S 暴露的持久 MCP 服务器，通常使用 FastMCP 的
mcp.run(transport="sse")模式。
SSE 是一种简单、单向的协议，其中服务器通过标准 HTTP 连接向客户端推送消息。
from fastmcp import Client
from fastmcp.client.transports import SSETransport
sse_url = "http://localhost:8000/sse"
# Option 1: Inferred transport
client_inferred = Client(sse_url)
# Option 2: Explicit transport (e.g., to add custom headers)
headers = {"Authorization": "Bearer mytoken"}
transport_explicit = SSETransport(url=sse_url, headers=headers)
client_explicit = Client(transport_explicit)
async def use_sse_client(client):
async with client:
tools = await client.list_tools()
print(f"Connected via SSE, found tools: {tools}")
# asyncio.run(use_sse_client(client_inferred))
# asyncio.run(use_sse_client(client_explicit))
WebSocket
- 类名:
fastmcp.client.transports.WSTransport - 推断自:
ws://或wss://URL - 用例: 使用WebSocket协议连接到MCP服务器进行双向通信。
WebSocket提供客户端和服务器之间持久、全双工的连接。
from fastmcp import Client
from fastmcp.client.transports import WSTransport
ws_url = "ws://localhost:9000"
# Option 1: Inferred transport
client_inferred = Client(ws_url)
# Option 2: Explicit transport
transport_explicit = WSTransport(url=ws_url)
client_explicit = Client(transport_explicit)
async def use_ws_client(client):
async with client:
tools = await client.list_tools()
print(f"Connected via WebSocket, found tools: {tools}")
# asyncio.run(use_ws_client(client_inferred))
# asyncio.run(use_ws_client(client_explicit))
内存传输
FastMCP 传输
- 类:
fastmcp.client.transports.FastMCPTransport - 推断自:
fastmcp.server.FastMCP的一个实例。 - 用例: 直接连接到在同一 Python 进程 中运行的
FastMCP服务器实例。
这对于以下情况非常有用：
- 测试: 为你的 FastMCP 服务器编写单元或集成测试，无需子进程或网络连接。
- 嵌入: 将 MCP 服务器作为更大应用程序中的组件使用。
from fastmcp import FastMCP, Client
from fastmcp.client.transports import FastMCPTransport
# 1、Create your FastMCP server instance
server = FastMCP(name="InMemoryServer")
@server.tool()
def ping(): return "pong"
# 2、Create a client pointing directly to the server instance
# Option A: Inferred
client_inferred = Client(server)
# Option B: Explicit
transport_explicit = FastMCPTransport(mcp=server)
client_explicit = Client(transport_explicit)
# 3、Use the client (no subprocess or network involved)
async def test_in_memory():
async with client_inferred: # Or client_explicit
result = await client_inferred.call_tool("ping")
print(f"In-memory call result: {result[0].text}") # Output: pong
# asyncio.run(test_in_memory())
通信通过高效的内存队列进行，因此速度非常快。
选择传输方式
-
本地开发/测试: 使用
PythonStdioTransport（从.py文件推断）或FastMCPTransport（用于同一进程测试）。 -
连接到远程/持久服务器: 使用
SSETransport（用于http/s）或WSTransport（用于ws/s）。 -
运行打包工具: 如果您需要运行 MCP 服务器而不进行本地安装，请使用
UvxStdioTransport（Python/uv）或NpxStdioTransport（Node/npm）。 -
与 Claude Desktop（或类似工具）集成: 这些工具通常期望运行 Python 脚本，因此您的服务器应可通过
python your_server.py运行，这使得PythonStdioTransport成为客户端上相关的机制。
Patterns - 代理服务器
https://gofastmcp.com/patterns/proxy
使用FastMCP作为其他MCP服务器的中介或更改传输。
新功能自版本：2.0.0
FastMCP提供强大的代理功能，允许一个FastMCP服务器实例作为另一个MCP服务器（可能是远程的、运行在不同的传输上，甚至是另一个FastMCP实例）的前端。这是通过使用FastMCP.from_client()类方法实现的。
代理是什么？
代理意味着设置一个不直接实现自身工具或资源的FastMCP服务器。相反，当它收到一个请求（如tools/call或resources/read）时，它会将该请求转发到一个后端 MCP服务器，接收响应，然后将该响应回传给原始客户端。
用例
- 传输桥接：通过不同的传输方式（例如，本地Stdio用于Claude桌面）暴露运行在一种传输方式上的服务器（例如，远程SSE服务器）。
- 添加功能：在现有服务器前插入一层，以添加缓存、日志记录、身份验证或修改请求/响应（尽管直接修改需要子类化
FastMCPProxy）。 - 安全边界：将代理用作受控的内部服务器网关。
- 简化客户端配置：即使后端服务器位置或传输方式发生变化，也提供一个单一、稳定的端点（代理）。
创建代理
创建代理最简单的方法是使用 FastMCP.from_client() 类方法。这将创建一个标准的 FastMCP 服务器，该服务器将请求转发到另一个 MCP 服务器。
from fastmcp import FastMCP, Client
# Create a client configured to talk to the backend server
# This could be any MCP server - remote, local, or using any transport
backend_client = Client("backend_server.py") # Could be "http://remote.server/sse", etc.
# Create the proxy server with from_client()
proxy_server = FastMCP.from_client(
backend_client, name="MyProxyServer" # Optional settings for the proxy
)
# That's it! You now have a proxy FastMCP server that can be used
# with any transport (SSE, stdio, etc.) just like any other FastMCP server
from_client 的工作原理：
1、 它使用提供的客户端连接到后端服务器。
2、 它发现后端服务器上所有可用的工具、资源、资源模板和提示。
3、 它创建相应的“代理”组件，这些组件将请求转发到后端。
4、 它返回一个标准的 FastMCP 服务器实例，该实例可以用作其他任何实例。
目前，代理主要关注于暴露主要的 MCP 对象（工具、资源、模板和提示）。一些高级 MCP 功能，如通知和采样，在当前版本的代理中尚未得到完全支持。对这些附加功能的支持可能会在未来版本中添加。
桥接传输
一个常见的用例是桥接传输。例如，通过Stdio使远程SSE服务器在本地可用：
from fastmcp import FastMCP, Client
# Client targeting a remote SSE server
client = Client("http://example.com/mcp/sse")
# Create a proxy server - it's just a regular FastMCP server
proxy = FastMCP.from_client(client, name="SSE to Stdio Proxy")
# The proxy can now be used with any transport
# No special handling needed - it works like any FastMCP server
内存代理
您还可以代理一个内存中的 FastMCP 实例，这对于调整您不完全控制的服务器的配置或行为非常有用。
from fastmcp import FastMCP
# Original server
original_server = FastMCP(name="Original")
@original_server.tool()
def tool_a() -> str:
return "A"
# Create a proxy of the original server
proxy = FastMCP.from_client(
original_server, name="Proxy Server"
)
# proxy is now a regular FastMCP server that forwards
# requests to original_server
FastMCPProxy 类
内部，FastMCP.from_client() 使用 FastMCPProxy 类。通常您不需要直接与这个类交互，但如果需要，它是可用的。
直接使用这个类可能在高级场景中是必要的，例如通过继承 FastMCPProxy 来在转发请求前后添加自定义逻辑。
传输组合blueskygithubx由 Mintlify 提供支持
Patterns - Server 组合
https://gofastmcp.com/patterns/composition
通过挂载和导入，将多个FastMCP服务器组合成一个更大的应用程序。
新功能自版本：2.2.0
随着您的MCP应用程序的增长，您可能希望将工具、资源和提示组织成逻辑模块或重用现有的服务器组件。FastMCP通过两种方法支持组合：
import_server：用于一次性复制带有前缀的组件（静态组合）。mount：用于创建一个实时链接，其中主服务器将请求委派给子服务器（动态组合）。
为什么使用 Compose 服务器？
- 模块化：将大型应用程序拆分为更小、更专注的服务器（例如，
WeatherServer、DatabaseServer、CalendarServer）。 - 可重用性：创建通用的实用服务器（例如，
TextProcessingServer）并在需要的地方挂载它们。 - 团队合作：不同的团队可以分别工作在单独的 FastMCP 服务器上，之后再将它们合并。
- 组织：将相关的功能逻辑地分组在一起。
导入与挂载
选择导入或挂载取决于您的用例和需求。一般来说，导入最适合简单情况，因为它将导入的服务器组件复制到主服务器中，将其视为原生集成。挂载最适合更复杂的情况，其中您需要在运行时将请求委派给子服务器。
| 特性 | 导入 | 挂载 |
|---|---|---|
| 方法 | FastMCP.import_server() | FastMCP.mount() |
| 组合类型 | 单次复制（静态） | 实时链接（动态） |
| 更新 | 子服务器更改不会反映 | 子服务器更改立即反映 |
| 生命周期 | 不管理 | 自动管理 |
| 同步性 | 异步（必须等待） | 同步 |
| 最适合 | 打包最终组件 | 模块化运行时组合 |
代理服务器
FastMCP支持MCP代理，允许您在本地FastMCP实例中镜像本地或远程服务器。代理与导入和挂载完全兼容。
Importing (Static Composition)
import_server() 方法将所有组件（工具、资源、模板、提示）从一个 FastMCP 实例（子服务器）复制到另一个实例（主服务器）。为了避免命名冲突，会添加一个前缀。
from fastmcp import FastMCP
import asyncio
# --- Define Subservers ---
# Weather Service
weather_mcp = FastMCP(name="WeatherService")
@weather_mcp.tool()
def get_forecast(city: str) -> dict:
"""Get weather forecast."""
return {"city": city, "forecast": "Sunny"}
@weather_mcp.resource("data://cities/supported")
def list_supported_cities() -> list[str]:
"""List cities with weather support."""
return ["London", "Paris", "Tokyo"]
# Calculator Service
calc_mcp = FastMCP(name="CalculatorService")
@calc_mcp.tool()
def add(a: int, b: int) -> int:
"""Add two numbers."""
return a + b
@calc_mcp.prompt()
def explain_addition() -> str:
"""Explain the concept of addition."""
return "Addition is the process of combining two or more numbers."
# --- Define Main Server ---
main_mcp = FastMCP(name="MainApp")
# --- Import Subservers ---
async def setup():
# Import weather service with prefix "weather"
await main_mcp.import_server("weather", weather_mcp)
# Import calculator service with prefix "calc"
await main_mcp.import_server("calc", calc_mcp)
# --- Now, main_mcp contains *copied* components ---
# Tools:
# - "weather_get_forecast"
# - "calc_add"
# Resources:
# - "weather+data://cities/supported" (prefixed URI)
# Prompts:
# - "calc_explain_addition"
if __name__ == "__main__":
# In a real app, you might run this async or setup imports differently
asyncio.run(setup())
# Run the main server, which now includes components from both subservers
main_mcp.run()
导入工作原理
当你调用 await main_mcp.import_server(prefix, subserver) 时：
1、工具：subserver 中的所有工具都被添加到 main_mcp 中。它们的名称会自动使用 prefix 和默认分隔符（_）进行前缀。
subserver.tool(name="my_tool")变为main_mcp.tool(name="{prefix}_my_tool")。
2、资源：subserver 中的所有资源都被添加。它们的 URI 会使用 prefix 和默认分隔符（+）进行前缀。
subserver.resource(uri="data://info")变为main_mcp.resource(uri="{prefix}+data://info")。
3、资源模板：subserver 中的所有模板都被添加。它们的 URI 模板会类似于资源进行前缀。
subserver.resource(uri="data://{id}")变为main_mcp.resource(uri="{prefix}+data://{id}")。
4、提示：subserver 中的所有提示都被添加，名称的前缀与工具类似。
subserver.prompt(name="my_prompt")变为main_mcp.prompt(name="{prefix}_my_prompt")。
请注意，import_server 在方法被调用时，会从 subserver 到 main_mcp 实例进行 一次性复制 组件。在调用 import_server 之后对 subserver 所做的更改 不会 反映在 main_mcp 中。此外，当使用 import_server 时，主服务器 不会 执行 subserver 的 lifespan 上下文。
自定义分隔符
您可能更喜欢为前缀名称和URI使用不同的分隔符。您可以在调用import_server()时自定义这些分隔符：
await main_mcp.import_server(
prefix="api", app=some_subserver, tool_separator="/", # Tool name becomes: "api/sub_tool_name"
resource_separator=":", # Resource URI becomes: "api:data://sub_resource"
prompt_separator="." # Prompt name becomes: "api.sub_prompt_name"
)
在选择分隔符时请小心。某些MCP客户端（如Claude Desktop）可能对工具名称中允许的字符有限制（例如，可能不支持 /）。默认值（_ 用于名称，+ 用于URIs）通常是比较安全的。
Mounting (Live Linking)
mount() 方法在 main_mcp 服务器和 subserver 之间创建一个 实时链接。而不是复制组件，匹配 prefix 的组件请求在运行时，被 委派 给 subserver。
import asyncio
from fastmcp import FastMCP, Client
# --- Define Subserver ---
dynamic_mcp = FastMCP(name="DynamicService")
@dynamic_mcp.tool()
def initial_tool(): return "Initial Tool Exists"
# --- Define Main Server ---
main_mcp = FastMCP(name="MainAppLive")
# --- Mount Subserver (Sync operation) ---
main_mcp.mount("dynamic", dynamic_mcp)
print("Mounted dynamic_mcp.")
# --- Add a tool AFTER mounting ---
@dynamic_mcp.tool()
def added_later(): return "Tool Added Dynamically!"
print("Added 'added_later' tool to dynamic_mcp.")
# --- Test Access ---
async def test_dynamic_mount():
# Need to use await for get_tools now
tools_before = await main_mcp.get_tools()
print("Tools available via main_mcp:", list(tools_before.keys()))
# Expected: ['dynamic_initial_tool', 'dynamic_added_later']
async with Client(main_mcp) as client:
# Call the dynamically added tool via the main server
result = await client.call_tool("dynamic_added_later")
print("Result of calling dynamic_added_later:", result[0].text)
# Expected: Tool Added Dynamically!
if __name__ == "__main__":
# Need async context to test
asyncio.run(test_dynamic_mount())
# To run the server itself:
# main_mcp.run()
如何挂载工作原理
当你调用 main_mcp.mount(prefix, server) 时：
1、实时链接：在 main_mcp 和 subserver 之间建立了一个实时连接。
2、动态更新：对 subserver（例如，添加新工具）所做的更改，在通过 main_mcp 访问组件时将立即反映出来。
3、生命周期管理：subserver 的 生命周期 上下文由 main_mcp 自动管理和执行。
4、委托：在运行时，对匹配前缀的组件的请求被委托给子服务器。
命名工具、资源、模板和提示时，与 import_server 一样适用相同的命名规则。
自定义分隔符
与 import_server 类似，您可以自定义前缀名称和 URI 的分隔符：
main_mcp.mount(
prefix="api",
app=some_subserver,
tool_separator="/", # Tool name becomes: "api/sub_tool_name"
resource_separator=":", # Resource URI becomes: "api:data://sub_resource"
prompt_separator="." # Prompt name becomes: "api.sub_prompt_name"
)
示例：模块化应用程序
以下是一个模块化应用程序如何使用 import_server 的示例：
main.py
from fastmcp import FastMCP
import asyncio
# Import the servers (see other files)
from modules.text_server import text_mcp
from modules.data_server import data_mcp
app = FastMCP(name="MainApplication")
# Setup function for async imports
async def setup():
# Import the utility servers
await app.import_server("text", text_mcp)
await app.import_server("data", data_mcp)
@app.tool()
def process_and_analyze(record_id: int) -> str:
"""Fetches a record and analyzes its string representation."""
# In a real application, you'd use proper methods to interact between
# imported tools rather than accessing internal managers
# Get record data
record = {"id": record_id, "value": random.random()}
# Count words in the record string representation
word_count = len(str(record).split())
return (
f"Record {record_id} has {word_count} words in its string "
f"representation."
)
if __name__ == "__main__":
# Run async setup before starting the server
asyncio.run(setup())
# Run the server
app.run()
modules/text_server.py
from fastmcp import FastMCP
text_mcp = FastMCP(name="TextUtilities")
@text_mcp.tool()
def count_words(text: str) -> int:
"""Counts words in a text."""
return len(text.split())
@text_mcp.resource("resource://stopwords")
def get_stopwords() -> list[str]:
"""Return a list of common stopwords."""
return ["the", "a", "is", "in"]
modules/data_server.py
from fastmcp import FastMCP
import random
from typing import dict
data_mcp = FastMCP(name="DataAPI")
@data_mcp.tool()
def fetch_record(record_id: int) -> dict:
"""Fetches a dummy data record."""
return {"id": record_id, "value": random.random()}
@data_mcp.resource("data://schema/{table}")
def get_table_schema(table: str) -> dict:
"""Provides a dummy schema for a table."""
return {"table": table, "columns": ["id", "value"]}
现在，运行 main.py 将启动一个服务器，该服务器公开以下内容：
text_count_wordsdata_fetch_recordprocess_and_analyzetext+resource://stopwordsdata+data://schema/{table}(template)
这种模式促进了在您的FastMCP项目中的代码组织和重用。
Patterns - 装饰方法
https://gofastmcp.com/patterns/decorating-methods
正确使用实例方法、类方法和静态方法与FastMCP装饰器一起使用。
FastMCP的装饰器系统旨在与函数一起工作，但如果您尝试装饰实例或类方法，可能会遇到意外的行为。本指南解释了使用所有FastMCP装饰器（@tool()、@resource()和@prompt()）的正确方法。
为什么方法难以使用？
当你将像 @tool()、@resource() 或 @prompt() 这样的 FastMCP 装饰器应用到方法上时，装饰器会在装饰时捕获该函数。例如，对于实例方法和类方法，这会带来挑战，因为：
1、对于实例方法：装饰器在存在任何实例之前就获取了未绑定方法
2、对于类方法：装饰器在将其绑定到类之前就获取了该函数
这意味着直接装饰这些方法不会按预期工作。实际上，LLM 会看到像 self 或 cls 这样的参数，它无法为其提供值。
推荐模式
实例方法
不要这样做（这不起作用）：
from fastmcp import FastMCP
mcp = FastMCP()
class MyClass:
@mcp.tool() # This won't work correctly
def add(self, x, y):
return x + y
@mcp.resource("resource://{param}") # This won't work correctly
def get_resource(self, param: str):
return f"Resource data for {param}"
当以这种方式应用装饰器时，它会捕获未绑定方法。当LLM后来尝试使用这个组件时，它会看到self作为一个必需的参数，但不知道应该为它提供什么，这可能导致错误或意外行为。
应该这样做：
from fastmcp import FastMCP
mcp = FastMCP()
class MyClass:
def add(self, x, y):
return x + y
def get_resource(self, param: str):
return f"Resource data for {param}"
# Create an instance first, then add the bound methods
obj = MyClass()
mcp.add_tool(obj.add)
mcp.add_resource_fn(obj.get_resource, uri="resource://{param}") # For resources or templates
# Note: FastMCP provides add_resource() for adding Resource objects directly and
# add_resource_fn() for adding functions that generate resources or templates
# Now you can call it without 'self' showing up as a parameter
await mcp.call_tool('add', {'x': 1, 'y': 2}) # Returns 3
这种方法之所以有效，是因为：
1、你首先创建了一个类的实例（obj）。
2、当你通过实例访问方法（obj.add）时，Python会创建一个绑定方法，其中self已经被设置为该实例。
3、当你注册这个绑定方法时，系统看到一个可调用对象，它只期望适当的参数，而不是self。
类方法
与实例方法类似，直接装饰类方法不会正确工作：
不要这样做：
from fastmcp import FastMCP
mcp = FastMCP()
class MyClass:
@classmethod
@mcp.tool() # This won't work correctly
def from_string(cls, s):
return cls(s)
这里的问题是，FastMCP装饰器在@classmethod装饰器之前应用（Python从下到上应用装饰器）。因此，它在函数被转换成类方法之前捕获了该函数，导致了不正确的行为。
应该这样做：
from fastmcp import FastMCP
mcp = FastMCP()
class MyClass:
@classmethod
def from_string(cls, s):
return cls(s)
# Add the class method after the class is defined
mcp.add_tool(MyClass.from_string)
这种方法有效是因为：
1、在类定义期间，@classmethod 装饰器被正确应用。
2、当你访问 MyClass.from_string 时，Python提供了一个特殊的方法对象，它会自动将类绑定到 cls 参数。
3、当注册时，只有适当的参数暴露给LLM，隐藏了 cls 参数的实现细节。
静态方法
与实例方法和类方法不同，静态方法与FastMCP装饰器配合良好：
from fastmcp import FastMCP
mcp = FastMCP()
class MyClass:
@staticmethod
@mcp.tool() # This works!
def utility(x, y):
return x + y
@staticmethod
@mcp.resource("resource://data") # This works too!
def get_data():
return "Static resource data"
这种方法有效是因为：
1、@staticmethod 装饰器首先被应用（最后执行），将方法转换为一个常规函数。
2、当FastMCP装饰器被应用时，它捕获的是一个有效的常规函数。
3、静态方法没有任何绑定要求 - 它不接收 self 或 cls 参数。
或者，你可以使用与其他方法相同的模式：
from fastmcp import FastMCP
mcp = FastMCP()
class MyClass:
@staticmethod
def utility(x, y):
return x + y
# This also works
mcp.add_tool(MyClass.utility)
这种方法同样有效，因为静态方法实质上只是类命名空间中的一个函数。
附加模式
在类初始化时创建组件
在创建对象时，您可以自动注册实例方法：
from fastmcp import FastMCP
mcp = FastMCP()
class ComponentProvider:
def __init__(self, mcp_instance):
# Register methods
mcp_instance.add_tool(self.tool_method)
mcp_instance.add_resource_fn(self.resource_method, uri="resource://data")
def tool_method(self, x):
return x * 2
def resource_method(self):
return "Resource data"
# The methods are automatically registered when creating the instance
provider = ComponentProvider(mcp)
这种模式在以下情况下非常有用：
- 你想在类本身内部封装注册逻辑。
- 你有多个应该一起注册的相关组件。
- 你想在创建实例时确保方法总是被正确注册。
类在初始化期间自动注册其方法，确保在注册之前它们正确地绑定到实例。
摘要
虽然FastMCP的装饰器模式与常规函数和静态方法无缝配合工作，但对于实例方法和类方法，您应该在创建实例或类之后添加它们。这确保了在注册之前方法被正确绑定。
以下模式适用于所有FastMCP装饰器和注册方法：
@tool()和add_tool()@resource()和add_resource_fn()@prompt()和add_prompt()
理解这些模式可以使您有效地将组件组织到类中，同时保持正确的方法绑定，让您在牺牲FastMCP装饰器系统简单性的同时，享受到面向对象设计的优势。
组合OpenAPIblueskygithubx由Mintlify提供支持
Patterns - OpenAPI 集成
https://gofastmcp.com/patterns/openapi
从 OpenAPI 规范生成 MCP 服务器
新增于版本：2.0.0
FastMCP 可以自动从 OpenAPI 规范生成 MCP 服务器。用户只需提供 OpenAPI 规范（3.0 或 3.1）和 API 客户端。
import httpx
from fastmcp import FastMCP
# Create a client for your API
api_client = httpx.AsyncClient(base_url="https://api.example.com")
# Load your OpenAPI spec
spec = {...}
# Create an MCP server from your OpenAPI spec
mcp = FastMCP.from_openapi(openapi_spec=spec, client=api_client)
if __name__ == "__main__":
mcp.run()
配置选项
超时
您可以设置所有API请求的超时时间：
# Set a 5 second timeout for all requests
mcp = FastMCP.from_openapi(
openapi_spec=spec,
client=api_client,
timeout=5.0
)
这个超时设置适用于由工具、资源和资源模板发起的所有请求。
路由映射
默认情况下，OpenAPI 路由根据以下规则映射到 MCP 组件：
| OpenAPI 路由 | 示例 | MCP 组件 | 备注 |
|---|---|---|---|
无路径参数的 GET | GET /stats | 资源 | 简单的资源，用于获取数据 |
带路径参数的 GET | GET /users/{id} | 资源模板 | 路径参数变为模板参数 |
POST、PUT、PATCH、DELETE 等 | POST /users | 工具 | 修改数据的操作 |
内部，FastMCP 使用一个按优先级排序的 RouteMap 对象集合来确定组件类型。路由映射指示特定的 HTTP 方法（或方法集）和路径模式应被视为特定的组件类型。这是默认的路由映射集合：
# Simplified version of the actual mapping rules
DEFAULT_ROUTE_MAPPINGS = [
# GET with path parameters -> ResourceTemplate
RouteMap(methods=["GET"], pattern=r".*\{.*\}.*",
route_type=RouteType.RESOURCE_TEMPLATE),
# GET without path parameters -> Resource
RouteMap(methods=["GET"], pattern=r".*",
route_type=RouteType.RESOURCE),
# All other methods -> Tool
RouteMap(methods=["POST", "PUT", "PATCH", "DELETE", "OPTIONS", "HEAD"], pattern=r".*", route_type=RouteType.TOOL), ]
自定义路由图
用户可以添加自定义路由图以覆盖默认的映射行为。用户提供的路由图始终首先应用，在默认路由图之前。
from fastmcp.server.openapi import RouteMap, RouteType
# Custom mapping rules
custom_maps = [
# Force all analytics endpoints to be Tools
RouteMap(methods=["GET"],
pattern=r"^/analytics/.*",
route_type=RouteType.TOOL)]
# Apply custom mappings
mcp = await FastMCP.from_openapi(
openapi_spec=spec, client=api_client, route_maps=custom_maps
)
工作原理
1、FastMCP解析您的OpenAPI规范以提取路由和模式
2、它应用映射规则来分类每个路由
3、当MCP客户端调用工具或访问资源时：
- FastMCP根据OpenAPI定义构建HTTP请求
- 它通过提供的httpx客户端发送请求
- 它将HTTP响应转换为适当的MCP格式
请求参数处理
FastMCP仔细处理OpenAPI请求中的不同类型参数：
查询参数
默认情况下，FastMCP只会包含具有非空值的查询参数。具有None值或空字符串（""）的参数将自动从请求中过滤掉。这确保了API服务器不会接收到可能引起问题的不必要的空参数。
例如，如果您使用以下参数调用一个工具：
await client.call_tool("search_products", {
"category": "electronics", # Will be included
"min_price": 100, # Will be included
"max_price": None, # Will be excluded
"brand": "", # Will be excluded
})
生成的HTTP请求将只包括 category=electronics&min_price=100。
路径参数
对于路径参数，通常由REST API所要求，FastMCP会过滤掉None值，并检查是否提供了所有必需的路径参数。如果缺少必需的路径参数或其值为None，则会引发错误。
# This will work
await client.call_tool("get_product", {"product_id": 123})
# This will raise ValueError: "Missing required path parameters: {'product_id'}"
await client.call_tool("get_product", {"product_id": None})
完整示例
import asyncio
import httpx
from fastmcp import FastMCP
# Sample OpenAPI spec for a Pet Store API
petstore_spec = {
"openapi": "3.0.0",
"info": {
"title": "Pet Store API",
"version": "1.0.0",
"description": "A sample API for managing pets",
},
"paths": {
"/pets": {
"get": {
"operationId": "listPets",
"summary": "List all pets",
"responses": {"200": {"description": "A list of pets"}},
},
"post": {
"operationId": "createPet",
"summary": "Create a new pet",
"responses": {"201": {"description": "Pet created successfully"}},
},
},
"/pets/{petId}": {
"get": {
"operationId": "getPet",
"summary": "Get a pet by ID",
"parameters": [
{
"name": "petId",
"in": "path",
"required": True,
"schema": {"type": "string"},
}
],
"responses": {
"200": {"description": "Pet details"},
"404": {"description": "Pet not found"},
},
}
},
},
}
async def check_mcp(mcp: FastMCP):
# List what components were created
tools = await mcp.get_tools()
resources = await mcp.get_resources()
templates = await mcp.get_resource_templates()
print(
f"{len(tools)} Tool(s): {', '.join([t.name for t in tools.values()])}"
) # Should include createPet
print(
f"{len(resources)} Resource(s): {', '.join([r.name for r in resources.values()])}"
) # Should include listPets
print(
f"{len(templates)} Resource Template(s): {', '.join([t.name for t in templates.values()])}"
) # Should include getPet
return mcp
if __name__ == "__main__":
# Client for the Pet Store API
client = httpx.AsyncClient(base_url="https://petstore.example.com/api")
# Create the MCP server
mcp = FastMCP.from_openapi(
openapi_spec=petstore_spec, client=client, name="PetStore"
)
asyncio.run(check_mcp(mcp))
# Start the MCP server
mcp.run()
Patterns - FastAPI 集成
https://gofastmcp.com/patterns/fastapi
从 FastAPI 应用生成 MCP 服务器
新功能版本：2.0.0
FastMCP 可以自动将 FastAPI 应用程序转换为 MCP 服务器。
FastMCP 并不包含 FastAPI 作为依赖项；您必须单独安装它才能运行这些示例。
from fastapi import FastAPI
from fastmcp import FastMCP
# A FastAPI app
app = FastAPI()
@app.get("/items")
def list_items():
return [{"id": 1, "name": "Item 1"}, {"id": 2, "name": "Item 2"}]
@app.get("/items/{item_id}")
def get_item(item_id: int):
return {"id": item_id, "name": f"Item {item_id}"}
@app.post("/items")
def create_item(name: str):
return {"id": 3, "name": name}
# Create an MCP server from your FastAPI app
mcp = FastMCP.from_fastapi(app=app)
if __name__ == "__main__":
mcp.run() # Start the MCP server
配置选项
超时
您可以为所有API请求设置超时：
# Set a 5 second timeout for all requests
mcp = FastMCP.from_fastapi(app=app, timeout=5.0)
路由映射
默认情况下，FastMCP会根据以下规则将FastAPI路由映射到MCP组件：
| FastAPI 路由类型 | FastAPI 示例 | MCP 组件 | 备注 |
|---|---|---|---|
| 无路径参数的GET | @app.get("/stats") | 资源 | 简单的资源，用于获取数据 |
| 带路径参数的GET | @app.get("/users/{id}") | 资源模板 | 路径参数变为模板参数 |
| POST、PUT、DELETE等 | @app.post("/users") | 工具 | 修改数据的操作 |
有关路由映射或自定义映射规则的更多详细信息，请参阅OpenAPI集成文档；FastMCP对FastAPI和OpenAPI集成使用相同的映射规则。
完整示例
以下是一个带有数据模型的更详细示例：
import asyncio
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from fastmcp import FastMCP, Client
# Define your Pydantic model
class Item(BaseModel):
name: str
price: float
# Create your FastAPI app
app = FastAPI()
items = {} # In-memory database
@app.get("/items")
def list_items():
"""List all items"""
return list(items.values())
@app.get("/items/{item_id}")
def get_item(item_id: int):
"""Get item by ID"""
if item_id not in items:
raise HTTPException(404, "Item not found")
return items[item_id]
@app.post("/items")
def create_item(item: Item):
"""Create a new item"""
item_id = len(items) + 1
items[item_id] = {"id": item_id, **item.model_dump()}
return items[item_id]
# Test your MCP server with a client
async def check_mcp(mcp: FastMCP):
# List the components that were created
tools = await mcp.get_tools()
resources = await mcp.get_resources()
templates = await mcp.get_resource_templates()
print(
f"{len(tools)} Tool(s): {', '.join([t.name for t in tools.values()])}"
)
print(
f"{len(resources)} Resource(s): {', '.join([r.name for r in resources.values()])}"
)
print(
f"{len(templates)} Resource Template(s): {', '.join([t.name for t in templates.values()])}"
)
return mcp
if __name__ == "__main__":
# Create MCP server from FastAPI app
mcp = FastMCP.from_fastapi(app=app)
asyncio.run(check_mcp(mcp))
# In a real scenario, you would run the server:
mcp.run()
优势
- 利用现有的FastAPI应用 - 无需重写您的API逻辑
- 模式复用 - FastAPI的Pydantic模型和验证可继承
- 全面功能支持 - 与FastAPI的认证、依赖项等兼容
- ASGI传输 - 直接通信，无需额外的HTTP开销
Patterns - 贡献模块
https://gofastmcp.com/patterns/contrib
社区贡献的扩展FastMCP的模块
新版本：2.2.1
FastMCP包含一个contrib包，其中包含社区贡献的模块。这些模块扩展了FastMCP的功能，但并非由核心团队官方维护。
贡献模块提供了额外的功能、集成或模式，以补充核心FastMCP库。它们为社区提供了一个分享有用扩展的方式，同时保持核心库的专注和可维护性。
可用的模块可以在贡献目录中查看。
使用方法
要使用一个 contrib 模块，从 fastmcp.contrib 包中导入它：
from fastmcp.contrib import my_module
重要注意事项
- 稳定性：
contrib中的模块可能具有与核心库不同的测试要求或稳定性保证。 - 兼容性：核心FastMCP的更改可能会破坏
contrib中的模块，而主变更日志中没有明确的警告。 - 依赖项：贡献模块可能具有核心库不需要的额外依赖项。这些依赖项通常在模块的README或单独的要求文件中记录。
贡献
我们欢迎对 contrib 包的贡献！如果您有一个以有用方式扩展 FastMCP 的模块，请考虑贡献它：
1、在 src/fastmcp/contrib/ 中为您的模块创建一个新目录
2、在 tests/contrib/ 中为您的模块添加适当的测试
3、在 README.md 文件中包含全面的文档，包括用法和示例，以及任何额外的依赖项或安装说明
4、提交一个拉取请求
理想的 contrib 模块：
- 解决特定的用例或集成需求
- 遵循 FastMCP 编码标准
- 包含详细的文档和示例
- 具有全面的测试
- 指定任何额外的依赖项
Patterns - 测试MCP服务器
https://gofastmcp.com/patterns/testing
了解如何有效地测试您的FastMCP服务器
彻底测试您的MCP服务器对于确保它们在部署时正确工作至关重要。FastMCP通过多种测试模式使这一过程变得简单。
内存测试
测试MCP服务器的最高效方式是将您的FastMCP服务器实例直接传递给一个客户端。这允许进行内存测试，而无需启动单独的服务器进程，这特别有用，因为以编程方式管理MCP服务器可能会很具挑战性。
以下是一个使用Client测试服务器的pytest示例：
import pytest
from fastmcp import FastMCP, Client
@pytest.fixture
def mcp_server():
server = FastMCP("TestServer")
@server.tool()
def greet(name: str) -> str:
return f"Hello, {name}!"
return server
async def test_tool_functionality(mcp_server):
# Pass the server directly to the Client constructor
async with Client(mcp_server) as client:
result = await client.call_tool("greet", {"name": "World"})
assert "Hello, World!" in str(result[0])
这种模式在客户端和服务器之间创建了一个直接的连接，使你能够高效地测试服务器的功能。
2025-05-03(六)
扫一扫
627
被折叠的 条评论
为什么被折叠?
到【灌水乐园】发言
