fastMCP基础(一)
一、基础概念
1、模型上下文协议 (MCP)
是一种为 LLM 提供上下文和工具的新的标准化方法, 全称 Model Context Protocol 允许您构建服务器,以安全、标准化的方式向 LLM 应用程序公开数据和功能。它通常被描述为“用于 AI 的 USB-C 端口”,提供了一种将 LLM 连接到它们可以使用的资源的统一方式。将其视为 API 可能更容易,但专为 LLM 交互而设计。MCP 服务器可以:
- 通过 Resources 公开数据(可以把这些看作是 GET 端点;它们用于将信息加载到 LLM 的上下文中)
- 通过工具提供功能(有点像 POST 端点;它们用于执行代码或以其他方式产生副作用)
- 通过 Prompts(用于 LLM 交互的可重用模板)定义交互模式
- 等等
2、FastMCP
它使构建 MCP 服务器和客户端变得简单直观。使用干净的 Pythonic 代码创建工具、公开资源、定义提示等:
from fastmcp import FastMCPmcp = FastMCP("Demo 🚀")@mcp.tool
def add(a: int, b: int) -> int:"""Add two numbers"""return a + bif __name__ == "__main__":mcp.run()
3、FastMCP 1.0
FastMCP 是使用 Model Context Protocol 的标准框架。FastMCP 1.0 于 2024 年被纳入官方 MCP Python SDK。
4、FastMCP 2.0
FastMCP 2.0是一个积极维护的版本 ,为使用 MCP 生态系统提供了完整的工具包。FastMCP 2.0 具有一套全面的功能,远远超出了核心 MCP 规范,所有这些都是为了提供最简单的生产路径 。这些功能包括部署、身份验证、客户端、服务器代理和组合、从 REST API 生成服务器、动态工具重写、内置测试工具、集成等。
5、为什么选择 FastMCP?
MCP 协议功能强大,但实现它涉及许多样板 - 服务器设置、协议处理程序、内容类型、错误管理。FastMCP 处理所有复杂的协议细节和服务器管理,因此您可以专注于构建出色的工具。它被设计为高级和 Pythonic;在大多数情况下,你只需要装饰一个函数。
FastMCP 2.0 已经发展成为一个远远超出基本协议实现的综合平台。虽然 1.0 提供了服务器构建功能(现在是官方 MCP SDK 的一部分),但 2.0 提供了一个完整的生态系统,包括客户端库、身份验证系统、部署工具、与主要 AI 平台的集成、测试框架和生产就绪的基础设施模式。
6、FastMCP 的目标
- 🚀 快速 :高级接口意味着更少的代码和更快的开发
- 🍀 简单 :使用最少的样板构建 MCP 服务器
- 🐍 Pythonic:对 Python 开发人员来说感觉很自然
- 🔍 完整 :适用于从开发到生产的所有 MCP 使用案例的综合平台
7、安装
pip install fastmcp
检查安装是否成功
fastmcp versionFastMCP version: 2.10.4
MCP version: 1.10.1
Python version: 3.12.9
Platform: macOS-14.7.1-arm64-arm-64bit
FastMCP root path: /opt/anaconda3/envs/agno_py12/lib/python3.12/site-packages
二、快速入门
1、创建 FastMCP 服务器
FastMCP 服务器是工具、资源和其他 MCP 组件的集合。要创建服务器,请先实例化 FastMCP 类。创建一个名为 my_server.py 的新文件并添加以下代码:
from fastmcp import FastMCPmcp = FastMCP("My MCP Server")
就是这样!您已经创建了一个 FastMCP 服务器
2、添加工具
要添加返回简单问候的工具,请编写一个函数并使用 @mcp.tool 装饰它以将其注册到服务器:
from fastmcp import FastMCPmcp = FastMCP("My MCP Server")@mcp.tool
def greet(name: str) -> str:return f"Hello, {name}!"
3、测试服务器
要测试服务器,请创建一个 FastMCP 客户端并将其指向服务器对象。
import asyncio
from fastmcp import FastMCP, Clientmcp = 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:)。您可以在同一上下文中进行多个客户端调用。
4、运行服务器
为了使用 Python 运行服务器,我们需要在服务器文件的 main 块中添加一个 run 语句。
from fastmcp import FastMCPmcp = FastMCP("My MCP Server")@mcp.tool
def greet(name: str) -> str:return f"Hello, {name}!"if __name__ == "__main__":mcp.run()
这允许我们使用默认的 stdio 传输方式,使用 python my_server.py 运行服务器,这是向客户端公开 MCP 服务器的标准方式。
5、与 Python 服务器交互
现在服务器可以使用 python my_server.py 执行,我们可以像任何其他 MCP 服务器一样与它交互。
在新文件中,创建一个客户端并将其指向服务器文件:
import asyncio
from fastmcp import Clientclient = 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"))
运行结果:
6、使用 FastMCP CLI
要让 FastMCP 为我们运行服务器,我们可以使用 fastmcp run 命令。这将启动服务器并保持运行,直到它停止。默认情况下,它将使用 stdio transport,这是一种简单的基于文本的协议,用于与服务器交互。
fastmcp run my_server.py:mcp
- 请注意,FastMCP 不需要服务器文件中的 main 块,如果存在,它将忽略它。相反,它会查找 CLI 命令中提供的服务器对象(此处为 mcp)。如果未提供服务器对象,fastmcp run 将自动在文件中搜索名为 “mcp”、“app” 或 “server” 的服务器。
三、FastMCP 服务器
FastMCP 应用程序的核心部分是 FastMCP 服务器类。此类充当应用程序工具、资源和提示的主容器,并管理与 MCP 客户端的通信。
1、创建 Server
实例化服务器非常简单。您通常会为服务器提供一个名称,这有助于在客户端应用程序或日志中识别它。
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.""",
)
2、FastMCP 构造函数参数
- name : 服务器可读名称
- instructions :有关如何与此服务器交互的说明。这些说明可帮助客户端了解服务器的用途和可用功能
- lifespan: 用于服务器启动和关闭逻辑的异步上下文管理器函数
- tags: 用于标记服务器本身的一组字符串
- tools : 要添加到服务器的工具(或要转换为工具的函数)的列表。在某些情况下,以编程方式提供工具可能比使用 @mcp.tool 装饰器更方便
- settings: 与其他 ServerSettings 配置对应的关键字参数
3、组件
3.1、Tool
工具是客户端可以调用以执行作或访问外部系统的函数。
@mcp.tool
def multiply(a: float, b: float) -> float:"""Multiplies two numbers together."""return a * b
3.2、Resources
资源公开客户端可以读取的数据源。
@mcp.resource("data://config")
def get_config() -> dict:"""Provides the application configuration."""return {"theme": "dark", "version": "1.0"}
3.3、Resource Templates
资源模板是允许客户端请求特定数据的参数化资源。
@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 functionreturn {"id": user_id, "name": f"User {user_id}", "status": "active"}
3.4、prompt
提示词是用于指导 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}"
4、Tag-Based Filtering
FastMCP 支持基于标签的过滤,以根据可配置的包含/排除标签集选择性地公开组件。这对于为不同的环境或用户创建不同的服务器视图非常有用。
使用 tags 参数定义组件时,可以对其进行标记:
@mcp.tool(tags={"public", "utility"})
def public_tool() -> str:return "This tool is public"@mcp.tool(tags={"internal", "admin"})
def admin_tool() -> str:return "This tool is for admins only"
筛选逻辑的工作原理如下:
- Include tags:如果指定,则仅公开具有至少一个匹配标记的组件
- Exclude tags: 筛选出具有任何匹配标记的组件
- Precedence: 排除标记始终优先于包含标记
您可以在创建服务器时配置基于标签的筛选:
# Only expose components tagged with "public"
mcp = FastMCP(include_tags={"public"})# Hide components tagged as "internal" or "deprecated"
mcp = FastMCP(exclude_tags={"internal", "deprecated"})# Combine both: show admin tools but hide deprecated ones
mcp = FastMCP(include_tags={"admin"}, exclude_tags={"deprecated"})
此筛选适用于所有组件类型(工具、资源、资源模板和提示),并影响列表和访问。
5、运行 Server
FastMCP 服务器需要一种传输机制来与 Client 端通信。你通常通过在 FastMCP 实例上调用 mcp.run() 方法来启动你的服务器,通常在主服务器脚本的 if name == “main”: 块中。此模式可确保与各种 MCP 客户端兼容。
FastMCP 支持多种传输选项:
- STDIO (默认,用于本地工具)
- Streamable HTTP (建议用于 Web 服务)
- SSE(旧版 Web 传输,已弃用)
该服务器也可以使用 FastMCP CLI 运行。
有关每种传输、如何配置它们(主机、端口、路径)以及何时使用哪种传输的详细信息,请参阅运行 FastMCP 服务器指南。
6、编写服务器
FastMCP 支持使用 import_server(静态复制)和 mount(实时链接)将多个服务器组合在一起。这允许您将大型应用程序组织成模块化组件或重用现有服务器。
请参阅 服务器组合 有关完整详细信息、最佳实践和示例。
# Example: Importing a subserver
from fastmcp import FastMCP
import asynciomain = FastMCP(name="Main")
sub = FastMCP(name="Sub")@sub.tool
def hello(): return "hi"# Mount directly
main.mount(sub, prefix="sub")
7、代理服务器
FastMCP 可以充当任何使用 FastMCP.as_proxy 的 MCP 服务器(本地或远程)的代理,让您桥接传输或向现有服务器添加前端。例如,您可以通过 stdio 在本地公开远程 SSE 服务器,反之亦然。
在使用断开连接的客户端时,代理会为每个请求创建新的会话,从而自动安全地处理每个请求。
from fastmcp import FastMCP, Clientbackend = Client("http://example.com/mcp/sse")
proxy = FastMCP.as_proxy(backend, name="ProxyServer")
# Now use the proxy like any FastMCP server
8、服务器配置
8.1、服务器的特有配置
在创建 FastMCP 实例和控制服务器行为时,会传递特定于服务器的设置:
from fastmcp import FastMCP# Configure server-specific settings
mcp = FastMCP(name="ConfiguredServer",dependencies=["requests", "pandas>=2.0.0"], # Optional server dependenciesinclude_tags={"public", "api"}, # Only expose these tagged componentsexclude_tags={"internal", "deprecated"}, # Hide these tagged componentson_duplicate_tools="error", # Handle duplicate registrationson_duplicate_resources="warn",on_duplicate_prompts="replace",
)
说明:
- dependencies :list[str] | None
- 包含软件包规范的可选服务器依赖项列表
- include_tags : set[str] | None
- 仅公开具有至少一个匹配标记的组件
- exclude_tags:set[str] | None
- 隐藏具有任何匹配标记的组件
- on_duplicate_tools: Literal[“error”, “warn”, “replace”]
- 如何处理重复的工具注册
- on_duplicate_resources: Literal[“error”, “warn”, “replace”]
- 如何处理重复的资源注册
- on_duplicate_prompts:Literal[“error”, “warn”, “replace”]
- 如何处理重复的提示注册
9、全局设置
全局设置会影响所有 FastMCP 服务器,可以通过环境变量(前缀为 FASTMCP_)或在 .env 文件中进行配置:
import fastmcp# Access global settings
print(fastmcp.settings.log_level) # Default: "INFO"
print(fastmcp.settings.mask_error_details) # Default: False
print(fastmcp.settings.resource_prefix_format) # Default: "path"
常见的全局设置包括:
- log_level:日志记录级别(“DEBUG”、“INFO”、“WARNING”、“ERROR”、“CRITICAL”),设置为 FASTMCP_LOG_LEVEL
- mask_error_details:是否对客户端隐藏详细的错误信息,使用 FASTMCP_MASK_ERROR_DETAILS 设置
- resource_prefix_format:如何设置资源前缀(“path”或“protocol”)的格式,使用 FASTMCP_RESOURCE_PREFIX_FORMAT
10、自定义工具序列化
默认情况下,FastMCP 会在需要将工具返回值转换为文本时将其序列化为 JSON。您可以通过在创建服务器时提供 tool_serializer 函数来自定义此行为:
import yaml
from fastmcp import FastMCP# Define a custom serializer that formats dictionaries as YAML
def yaml_serializer(data):return yaml.dump(data, sort_keys=False)# Create a server with the custom serializer
mcp = FastMCP(name="MyServer", tool_serializer=yaml_serializer)@mcp.tool
def get_config():"""Returns configuration in YAML format."""return {"api_key": "abc123", "debug": True, "rate_limit": 100}
serializer 函数采用任何数据对象并返回字符串表示形式。这适用于工具中的所有非字符串返回值 。已返回字符串的工具会绕过序列化程序。
当您需要执行以下作时,此自定义非常有用:
- 以特定方式设置数据格式(如 YAML 或自定义格式)
- 控制特定的序列化选项 (如缩进或排序)
- 在将数据发送到客户端之前添加元数据或转换数据