FastMCP v2：构建MCP服务器和客户端的Python利器

news2025/5/15 17:57:56

FastMCP v2：构建MCP服务器和客户端的Python利器

引言

在人工智能与大语言模型（LLMs）的应用场景中，如何高效地构建服务器和客户端以实现数据交互与功能调用是关键问题。Model Context Protocol (MCP) 为此提供了一种标准，而FastMCP v2则是基于Python实现的用于构建MCP服务器和客户端的强大工具。它在原有版本基础上进行了显著扩展，引入了丰富的高级特性，让开发者能够更轻松地创建和管理MCP应用程序。

什么是FastMCP v2

FastMCP v2是FastMCP 1.0的继任者，是一个专注于简化MCP服务器和客户端开发的Python库。它通过简洁的Python接口和丰富的功能特性，帮助开发者快速实现工具、资源、提示等组件的注册与管理，同时支持服务器代理、组合以及与OpenAPI/FastAPI集成等高级功能。

主要特点

简洁的Python接口：使用装饰器和简单的函数调用，让开发者可以用最少的代码完成复杂的MCP服务器和客户端开发。
功能丰富：支持工具、资源、模板、提示、上下文等核心概念，满足各种应用场景需求。
高级特性：包括服务器代理、服务器组合、OpenAPI/FastAPI集成等，方便与现有系统集成。

安装

可以通过以下命令使用pip安装FastMCP：

pip install fastmcp

核心功能及用途

1. 工具（Tools）

工具是FastMCP中用于执行特定任务的可调用函数，类似于HTTP的 POST/PUT 请求。在实际应用中，工具可以用于数据处理、模型调用、文件操作等各种任务。

工具注册

可以通过 @mcp.tool() 装饰器将一个函数注册为工具。例如：

from fastmcp import FastMCP

mcp = FastMCP("Demo 🚀")

@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two numbers"""
    return a + b

在这个例子中，add 函数被注册为一个工具，用于将两个整数相加。工具的文档字符串会被用作工具的描述，方便其他开发者理解工具的用途。

工具调用

在客户端可以通过 Client 类调用注册的工具。示例如下：

from fastmcp.client.client import Client
from mcp.types import TextContent

async def test_call_tool():
    async with Client(mcp) as client:
        result = await client.call_tool("add", {"a": 1, "b": 2})
        assert len(result) == 1
        assert isinstance(result[0], TextContent)
        assert result[0].text == "3"

2. 资源与模板（Resources & Templates）

资源类似于HTTP的 GET 请求，用于加载信息到上下文。模板则是用于动态生成资源的一种机制。资源可以是文件内容、数据库查询结果等。

资源注册

可以通过 @mcp.resource() 装饰器将一个函数注册为资源。例如：

from fastmcp import FastMCP
from pathlib import Path

mcp = FastMCP("Demo 🚀")
test_dir = Path(__file__).parent

@mcp.resource("file://test_dir/config.json")
def read_config_json() -> str:
    """Read the config.json file"""
    try:
        return (test_dir / "config.json").read_text()
    except FileNotFoundError:
        return "File not found"

这个例子中，read_config_json 函数被注册为一个资源，用于读取 config.json 文件的内容。

资源读取

在客户端可以通过 Client 类读取注册的资源。示例如下：

from fastmcp.client.client import Client
from mcp.types import TextResourceContents

async def test_read_resource():
    async with Client(mcp) as client:
        result = await client.read_resource("file://test_dir/config.json")
        assert len(result) == 1
        assert isinstance(result[0], TextResourceContents)

3. 提示（Prompts）

提示是可重用的模板，用于定义与LLM的交互模式。在与大语言模型交互时，提示可以帮助引导模型生成更符合预期的结果。

提示注册

可以通过 @mcp.prompt() 装饰器将一个函数注册为提示。例如：

from fastmcp import FastMCP

mcp = FastMCP("Demo 🚀")

@mcp.prompt()
def explain_sql(query: str) -> str:
    return f"Explaining SQL query:\n{query}"

这里，explain_sql 函数被注册为一个提示，用于解释SQL查询。

提示使用

在客户端可以通过 Client 类获取注册的提示。示例如下：

from fastmcp.client.client import Client

async def test_get_prompt():
    async with Client(mcp) as client:
        result = await client.get_prompt("explain_sql", {"query": "SELECT * FROM users"})
        assert len(result.messages) == 1

4. 上下文（Context）

上下文提供了在工具执行过程中共享信息的机制。可以通过 get_context() 函数获取当前上下文。例如，在工具执行过程中记录日志：

from fastmcp import FastMCP
from fastmcp.server.context import get_context

mcp = FastMCP("Demo 🚀")

@mcp.tool()
async def some_tool():
    context = get_context()
    await context.info("Tool execution started")
    # 工具逻辑
    await context.info("Tool execution finished")
    return "Tool result"

5. MCP客户端

FastMCP v2提供了强大的客户端功能，用于与MCP服务器进行交互。客户端可以调用服务器上注册的工具、读取资源、获取提示等。

客户端使用示例

from fastmcp.client.client import Client
from fastmcp import FastMCP
from mcp.types import TextContent

async def test_client():
    mcp = FastMCP("Demo 🚀")

    @mcp.tool()
    def add(a: int, b: int) -> int:
        """Add two numbers"""
        return a + b

    async with Client(mcp) as client:
        result = await client.call_tool("add", {"a": 1, "b": 2})
        assert len(result) == 1
        assert isinstance(result[0], TextContent)
        assert result[0].text == "3"

高级特性

1. 代理服务器（Proxy Servers）

FastMCP支持代理服务器功能，允许将请求转发到其他MCP服务器。这在分布式系统中非常有用，可以实现负载均衡和服务拆分。

2. 组合MCP服务器（Composing MCP Servers）

可以将多个MCP服务器组合在一起，形成一个更大的服务器。例如：

from fastmcp import FastMCP

main_app = FastMCP("MainApp")
api_app = FastMCP("APIApp")

def fetch_data(query: str) -> str:
    return f"Data for query: {query}"

api_app.add_tool(fetch_data, name="get_data")
await main_app.import_server("api", api_app)