New in version: 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 without path paramsGET /statsResourceSimple resources for fetching data
GET with path paramsGET /users/{id}Resource TemplatePath parameters become template parameters
POST, PUT, PATCH, DELETE, etc.POST /usersToolOperations that modify data

在内部,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. When an MCP client calls a tool or accesses a resource:
    • 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()