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 构造函数接受多个参数:

  • name: (可选) 为您的服务器设置一个可读的名称。默认为"FastMCP"。
  • instructions: (可选) 描述如何与该服务器交互的说明。这些说明帮助客户端理解服务器的用途和可用功能。
  • lifespan: (可选) 一个用于服务器启动和关闭逻辑的异步上下文管理器函数。
  • tags: (可选) 用于标记服务器本身的一组字符串。
  • **settings: 对应额外ServerSettings配置的关键字参数

组件

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"}

查看资源与模板获取详细文档。

提示词

提示词是可重复使用的消息模板,用于指导大语言模型。

@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}"

查看Prompts获取详细文档。

运行服务器

FastMCP服务器需要一种传输机制来与客户端通信。通常您可以通过在主服务器脚本的if __name__ == "__main__":代码块中调用FastMCP实例的mcp.run()方法来启动服务器。这种模式确保与各种MCP客户端的兼容性。

# 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 runs the server, defaulting to STDIO transport
    mcp.run()
    
    # To use a different transport, e.g., HTTP:
    # mcp.run(transport="streamable-http", host="127.0.0.1", port=9000)

FastMCP支持多种传输选项:

  • STDIO(默认,用于本地工具)
  • 可流式传输的HTTP(推荐用于Web服务)
  • SSE (传统网页传输协议,已弃用)

服务器也可以使用FastMCP CLI来运行。

有关每种传输方式的详细信息、如何配置它们(主机、端口、路径)以及何时使用哪种传输方式,请参阅运行您的FastMCP服务器指南。

服务器组合

New in version: 2.2.0

FastMCP支持通过import_server(静态复制)和mount(实时链接)将多个服务器组合在一起。这使您能够将大型应用程序组织成模块化组件或重用现有服务器。

查看Server Composition指南获取完整详情、最佳实践和示例。

# Example: Importing a subserver
from fastmcp import FastMCP
import asyncio

main = FastMCP(name="Main")
sub = FastMCP(name="Sub")

@sub.tool()
def hello(): 
    return "hi"

# Mount directly
main.mount("sub", sub)

代理服务器

New in version: 2.0.0

FastMCP可以作为任何MCP服务器(本地或远程)的代理,通过使用FastMCP.from_client,让你能够桥接传输协议或为现有服务器添加前端。例如,你可以通过标准输入输出在本地暴露一个远程SSE服务器,反之亦然。

详情及高级用法请参阅Proxying Servers指南。

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传输的主机地址(默认:"127.0.0.1")
  • port: SSE传输的端口号(默认值:8000)
  • log_level: 日志级别 (默认值: "INFO")
  • on_duplicate_tools: 如何处理重复工具注册
  • on_duplicate_resources: 如何处理重复的资源注册
  • on_duplicate_prompts: 如何处理重复的提示注册

所有这些都可以在创建FastMCP实例时直接作为参数进行配置。

自定义工具序列化

New in version: 2.2.7

默认情况下,当需要将工具返回值转换为文本时,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}

序列化函数接收任意数据对象并返回其字符串表示形式。该函数会应用于所有非字符串类型的工具返回值。对于已经返回字符串的工具,则会跳过序列化步骤。

这种定制在以下情况下很有用:

  • 以特定格式格式化数据(如YAML或自定义格式)
  • 控制特定的序列化选项(如缩进或排序)
  • 在将数据发送给客户端之前添加元数据或转换数据

如果序列化函数抛出异常,该工具将回退到默认的JSON序列化以避免服务器中断。

身份验证

New in version: 2.2.7

FastMCP支持OAuth 2.0认证,允许服务器保护其工具和资源。这通过在FastMCP初始化期间提供auth_server_providerauth设置来配置。

from fastmcp import FastMCP
from mcp.server.auth.settings import AuthSettings #, ... other auth imports
# from your_auth_implementation import MyOAuthServerProvider # Placeholder

# Create a server with authentication (conceptual example)
# mcp = FastMCP(
#     name="SecureApp",
#     auth_server_provider=MyOAuthServerProvider(),
#     auth=AuthSettings(
#         issuer_url="https://myapp.com",
#         # ... other OAuth settings ...
#         required_scopes=["myscope"],
#     ),
# )

由于当前MCP SDK认证提供程序接口的底层特性,详细实现无法通过简单示例说明。具体实现OAuthAuthorizationServerProvider的方法请参阅MCP SDK文档。FastMCP通过将提供程序和设置传递给底层MCP服务器来实现集成。

专门的认证指南将在FastMCP提供更高级别的抽象后更详细地介绍这一点。