Doramagic 项目包 · 项目说明书
fastapi_mcp 项目
生成时间:2026-05-11 16:52:13 UTC
项目概览
FastAPI-MCP 是一个将 FastAPI 应用程序自动转换为 MCP(Model Context Protocol)服务器的框架。该项目由 Tadata Inc. 开发,采用 MIT 开源许可证,为开发者提供了一种无缝集成 AI 助手能力到现有 FastAPI 应用的解决方案。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
简介
FastAPI-MCP 是一个将 FastAPI 应用程序自动转换为 MCP(Model Context Protocol)服务器的框架。该项目由 Tadata Inc. 开发,采用 MIT 开源许可证,为开发者提供了一种无缝集成 AI 助手能力到现有 FastAPI 应用的解决方案。
资料来源:README.md:1
核心特性
FastAPI-MCP 具有以下设计目标和技术特点:
| 特性 | 描述 |
|---|---|
| 内置认证 | 支持使用现有的 FastAPI 依赖项进行身份验证 |
| FastAPI 原生 | 不仅仅是 OpenAPI 到 MCP 的简单转换器 |
| 零配置/最小配置 | 只需指向 FastAPI 应用即可运行 |
| Schema 保持 | 保留请求模型和响应模型的完整类型定义 |
| 文档保留 | 端点文档与 Swagger 中完全一致 |
| 灵活部署 | 支持将 MCP 服务器挂载到 FastAPI 应用 |
资料来源:README.md:24-31
技术架构
整体架构
graph TD
A[FastAPI 应用] --> B[FastApiMCP]
B --> C[MCP 服务器]
C --> D[HTTP 端点 /mcp]
B --> E[工具发现]
E --> F[OpenAPI Schema 解析]
F --> G[MCP Tools]
B --> H[请求处理]
H --> I[httpx.AsyncClient]
I --> A核心模块结构
fastapi_mcp/
├── __init__.py # 包入口,导出公共API
├── server.py # FastApiMCP 主类实现
├── types.py # 类型定义(AuthConfig、OAuthMetadata)
└── openapi/
└── convert.py # OpenAPI Schema 转换逻辑
资料来源:fastapi_mcp/__init__.py:1-17
核心 API
公开导出
fastapi_mcp/__init__.py 导出了以下核心组件:
from fastapi_mcp import FastApiMCP, AuthConfig, OAuthMetadata
| 组件 | 用途 |
|---|---|
FastApiMCP | 主类,用于创建和管理 MCP 服务器 |
AuthConfig | 认证配置类 |
OAuthMetadata | OAuth 2.0 元数据类 |
资料来源:fastapi_mcp/__init__.py:12-16
类型定义
fastapi_mcp/types.py 定义了以下关键类型:
#### HTTPRequestInfo
class HTTPRequestInfo(BaseType):
method: str
path: str
headers: Dict[str, str]
cookies: Dict[str, str]
query_params: Dict[str, str]
body: Any
#### AuthConfig
认证配置类,支持 OAuth 2.0 授权流程:
| 参数 | 类型 | 说明 |
|---|---|---|
version | Literal["2025-03-26"] | MCP 规范版本 |
dependencies | Optional[Sequence[Depends]] | FastAPI 认证依赖项 |
issuer | Optional[str] | OAuth 2.0 发行方标识 |
oauth_metadata_url | Optional[StrHttpUrl] | OAuth 元数据端点 URL |
authorize_url | Optional[StrHttpUrl] | 授权端点 URL |
token_endpoint | Optional[StrHttpUrl] | 令牌端点 URL |
资料来源:fastapi_mcp/types.py:80-120
快速开始
环境要求
| 要求 | 版本 |
|---|---|
| Python | 3.10+(推荐 3.12) |
| 包管理器 | uv |
资料来源:README.md:36-39
基础用法
from examples.shared.apps.items import app # FastAPI 应用
from fastapi_mcp import FastApiMCP
# 创建 MCP 服务器
mcp = FastApiMCP(app)
# 挂载到 FastAPI 应用
mcp.mount_http()
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
资料来源:examples/01_basic_usage_example.py:1-14
高级功能
工具过滤
支持通过操作 ID 和标签过滤暴露的 MCP 工具:
# 通过操作 ID 过滤(包含模式)
mcp = FastApiMCP(
app,
include_operations=["get_item", "list_items"],
)
# 通过操作 ID 过滤(排除模式)
mcp = FastApiMCP(
app,
exclude_operations=["create_item", "update_item", "delete_item"],
)
# 通过标签过滤
mcp = FastApiMCP(
app,
include_tags=["items"],
)
资料来源:examples/03_custom_exposed_endpoints_example.py:18-37
认证配置
FastAPI-MCP 支持基于 FastAPI 依赖项的认证机制:
from fastapi import Depends
from fastapi.security import HTTPBearer
from fastapi_mcp import FastApiMCP, AuthConfig
token_auth_scheme = HTTPBearer()
mcp = FastApiMCP(
app,
auth_config=AuthConfig(
dependencies=[Depends(token_auth_scheme)],
),
)
资料来源:examples/08_auth_example_token_passthrough.py:30-40
版本历史
| 版本 | 日期 | 主要变更 |
|---|---|---|
| 0.2.0 | - | 移除 CLI,改用运行时集成;添加 add_mcp_server 函数 |
| 0.1.1 | 2024-07-03 | 修复 PEP 604 联合类型语法兼容性 |
| 0.1.0 | 2024-03-08 | 初始版本发布 |
资料来源:CHANGELOG.md:1-30
开发指南
环境搭建
# 1. 安装 uv
curl -LsSf https://astral.sh/uv/install.sh | sh
# 2. 克隆仓库
git clone https://github.com/YOUR-USERNAME/fastapi_mcp.git
cd fastapi_mcp
# 3. 同步依赖
uv sync
# 4. 安装预提交钩子
uv run pre-commit install
代码质量检查
| 工具 | 用途 |
|---|---|
| ruff | 代码检查和格式化 |
| mypy | 类型检查 |
| pytest | 测试框架 |
# 运行所有测试
pytest
# 类型检查
mypy .
# 代码格式化检查
ruff check .
ruff format .
许可证
MIT License。Copyright (c) 2025 Tadata Inc.
资料来源:README.md:42
资料来源:[README.md:1](https://github.com/tadata-org/fastapi_mcp/blob/main/README.md)
快速开始
本文档面向希望快速上手 FastAPI-MCP 的开发者。FastAPI-MCP 是一个将 FastAPI 应用无缝转换为 MCP(Model Context Protocol)服务器的库,使 FastAPI 应用能够作为 MCP 工具被 AI 模型调用。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
环境要求
在开始之前,请确保您的开发环境满足以下要求:
| 要求 | 版本 | 说明 |
|---|---|---|
| Python | 3.10+(推荐 3.12) | 项目依赖类型注解等特性 |
| uv | 最新稳定版 | 高性能 Python 包管理器 |
资料来源:CONTRIBUTING.md:1
安装 uv
如果尚未安装 uv,请参考 uv 官方安装指南。
项目初始化
克隆项目
首先,从 GitHub fork 并克隆仓库:
git clone https://github.com/YOUR-USERNAME/fastapi_mcp.git
cd fastapi-mcp
# 添加上游远程仓库
git remote add upstream https://github.com/tadata-org/fastapi_mcp.git
资料来源:CONTRIBUTING.md:1
创建虚拟环境并安装依赖
使用 uv sync 自动创建虚拟环境并安装所有依赖:
uv sync
资料来源:CONTRIBUTING.md:1
安装预提交钩子
项目使用 pre-commit 工具自动运行代码检查:
uv run pre-commit install
uv run pre-commit run
资料来源:CONTRIBUTING.md:1
基础使用
基本示例
以下示例展示将 FastAPI 应用转换为 MCP 服务器的最简方式:
from examples.shared.apps.items import app # FastAPI 应用
from examples.shared.setup import setup_logging
from fastapi_mcp import FastApiMCP
setup_logging()
# 创建 MCP 服务器
mcp = FastApiMCP(app)
# 挂载 MCP 服务器
mcp.mount_http()
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
资料来源:examples/01_basic_usage_example.py:1
工作流程
graph TD
A[FastAPI 应用] --> B[创建 FastApiMCP 实例]
B --> C[挂载 MCP 端点]
C --> D[启动 Uvicorn 服务器]
D --> E[访问 MCP 端点]核心配置参数
FastApiMCP 类支持多种配置选项:
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
name | str | - | MCP 服务器名称 |
description | str | - | 服务器描述 |
describe_full_response_schema | bool | False | 是否在工具描述中包含完整响应 JSON Schema |
describe_all_responses | bool | False | 是否描述所有可能的响应(包括错误响应) |
include_operations | List[str] | None | 要包含的操作 ID 列表 |
exclude_operations | List[str] | None | 要排除的操作 ID 列表 |
include_tags | List[str] | None | 要包含的标签列表 |
exclude_tags | List[str] | None | 要排除的标签列表 |
headers | List[str] | ["authorization"] | 要转发到工具调用的 HTTP 头列表 |
完整 Schema 描述
如需在工具描述中包含完整的响应 Schema,使用以下配置:
mcp = FastApiMCP(
app,
name="Item API MCP",
description="MCP 服务器示例",
describe_full_response_schema=True,
describe_all_responses=True,
)
mcp.mount_http()
资料来源:examples/02_full_schema_description_example.py:1
操作过滤
按操作 ID 过滤
可以选择性地暴露特定操作:
# 仅包含指定操作
include_mcp = FastApiMCP(
app,
include_operations=["get_item", "list_items"],
)
# 排除指定操作
exclude_mcp = FastApiMCP(
app,
exclude_operations=["create_item", "update_item", "delete_item"],
)
资料来源:examples/03_custom_exposed_endpoints_example.py:1
按标签过滤
也可以按 FastAPI 的标签进行过滤:
# 仅包含 items 标签的操作
items_mcp = FastApiMCP(
app,
include_tags=["items"],
)
# 排除 search 标签的操作
exclude_search_mcp = FastApiMCP(
app,
exclude_tags=["search"],
)
资料来源:examples/03_custom_exposed_endpoints_example.py:1
过滤规则说明
| 规则 | 说明 |
|---|---|
include_operations 和 exclude_operations | 互斥,不能同时使用 |
include_tags 和 exclude_tags | 互斥,不能同时使用 |
| 操作过滤 + 标签过滤 | 可以组合使用,匹配任一条件的操作都会被包含 |
资料来源:examples/03_custom_exposed_endpoints_example.py:1
认证配置
令牌传递认证
FastAPI-MCP 支持通过 HTTP 头传递认证令牌:
from fastapi import Depends
from fastapi.security import HTTPBearer
from fastapi_mcp import FastApiMCP, AuthConfig
token_auth_scheme = HTTPBearer()
@app.get("/private")
async def private(token=Depends(token_auth_scheme)):
return token.credentials
mcp = FastApiMCP(
app,
name="Protected MCP",
auth_config=AuthConfig(
dependencies=[Depends(token_auth_scheme)],
),
)
mcp.mount_http()
资料来源:examples/08_auth_example_token_passthrough.py:1
AuthConfig 配置选项
| 参数 | 类型 | 说明 |
|---|---|---|
version | Literal["2025-03-26"] | MCP 规范版本 |
dependencies | Sequence[params.Depends] | FastAPI 认证依赖 |
示例 FastAPI 应用
以下是一个完整的 CRUD 示例应用结构:
from fastapi import FastAPI, HTTPException, Query
from pydantic import BaseModel
from typing import List, Optional
app = FastAPI()
class Item(BaseModel):
id: int
name: str
description: Optional[str] = None
price: float
tags: List[str] = []
items_db = {}
@app.post("/items/", response_model=Item)
async def create_item(item: Item):
items_db[item.id] = item
return item
@app.get("/items/{item_id}", response_model=Item)
async def get_item(item_id: int):
if item_id not in items_db:
raise HTTPException(status_code=404, detail="Item not found")
return items_db[item_id]
@app.get("/items/", response_model=List[Item])
async def list_items():
return list(items_db.values())
@app.put("/items/{item_id}", response_model=Item)
async def update_item(item_id: int, item: Item):
if item_id not in items_db:
raise HTTPException(status_code=404, detail="Item not found")
item.id = item_id
items_db[item_id] = item
return item
@app.delete("/items/{item_id}")
async def delete_item(item_id: int):
if item_id not in items_db:
raise HTTPException(status_code=404, detail="Item not found")
del items_db[item_id]
return {"message": "Item deleted successfully"}
@app.get("/items/search/", response_model=List[Item])
async def search_items(
q: Optional[str] = Query(None),
min_price: Optional[float] = Query(None),
max_price: Optional[float] = Query(None),
tags: List[str] = Query([]),
):
results = list(items_db.values())
# 搜索逻辑...
return results
资料来源:examples/shared/apps/items.py:1
运行测试
代码检查
在提交代码前,运行以下检查确保代码质量:
# 类型检查
mypy .
# 运行测试
pytest
# 代码格式化和检查
ruff check .
ruff format .
资料来源:CONTRIBUTING.md:1
使用预提交
如果已安装 pre-commit 钩子,这些检查会在每次提交时自动运行:
uv run pre-commit run
挂载 MCP 端点
自定义挂载路径
可以为 MCP 服务器指定不同的挂载路径:
mcp.mount_http(mount_path="/mcp")
多端点挂载
可以同时挂载多个 MCP 服务器实例:
mcp1 = FastApiMCP(app, name="Server 1")
mcp2 = FastApiMCP(app, name="Server 2")
mcp1.mount_http(mount_path="/mcp1")
mcp2.mount_http(mount_path="/mcp2")
下一步
资料来源:[CONTRIBUTING.md:1]()
系统架构
FastAPI-MCP 是一个将 FastAPI 应用自动转换为 MCP(Model Context Protocol)服务器的框架。其核心架构围绕 FastApiMCP 类展开,通过 OpenAPI schema 提取、工具过滤、HTTP 请求转发和认证集成四大核心模块实现功能。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
概述
FastAPI-MCP 是一个将 FastAPI 应用自动转换为 MCP(Model Context Protocol)服务器的框架。其核心架构围绕 FastApiMCP 类展开,通过 OpenAPI schema 提取、工具过滤、HTTP 请求转发和认证集成四大核心模块实现功能。
系统采用零配置设计理念,开发者只需将现有的 FastAPI 应用传入 FastApiMCP,即可自动生成兼容 MCP 协议的服务端点。框架支持灵活的工具过滤机制、多种认证配置方式以及自定义 HTTP 客户端注入。
核心组件
组件关系图
graph TD
A[FastAPI App] --> B[FastApiMCP]
B --> C[OpenAPI Schema 解析]
C --> D[工具转换引擎]
D --> E[MCP 工具列表]
E --> F[工具过滤器]
F --> G[最终 MCP 工具集]
H[HTTP 请求转发器] --> G
I[认证配置] --> H
J[自定义 HTTP Client] --> H主要类结构
| 组件 | 文件位置 | 职责 |
|---|---|---|
FastApiMCP | fastapi_mcp/server.py | 主服务器类,核心逻辑入口 |
AuthConfig | fastapi_mcp/types.py | 认证配置数据模型 |
OAuthMetadata | fastapi_mcp/types.py | OAuth 2.0 元数据 |
HTTPRequestInfo | fastapi_mcp/types.py | HTTP 请求信息封装 |
工具发现与过滤机制
工具过滤流程
graph TD
A[OpenAPI Schema] --> B{include_operations?}
B -->|有值| C[过滤至指定操作ID]
B -->|无值| D{exclude_operations?}
D -->|有值| E[排除指定操作ID]
D -->|无值| F{include_tags?}
F -->|有值| G[过滤至指定标签]
F -->|无值| H{exclude_tags?}
H -->|有值| I[排除指定标签]
H -->|无值| J[返回全部工具]
C --> J
E --> J
G --> J
I --> J过滤配置参数
| 参数名 | 类型 | 说明 | 互斥参数 |
|---|---|---|---|
include_operations | List[str] | 仅包含指定的操作 ID | exclude_operations |
exclude_operations | List[str] | 排除指定的操作 ID | include_operations |
include_tags | List[str] | 仅包含指定标签的端点 | exclude_tags |
exclude_tags | List[str] | 排除指定标签的端点 | include_tags |
资料来源:fastapi_mcp/server.py:50-75
过滤实现逻辑
def _filter_tools(self, tools: List[types.Tool], openapi_schema: Dict[str, Any]) -> List[types.Tool]:
if (
self._include_operations is None
and self._exclude_operations is None
and self._include_tags is None
and self._exclude_tags is None
):
return tools
# 构建按标签分组操作ID的映射
operations_by_tag: Dict[str, List[str]] = {}
for path, path_item in openapi_schema.get("paths", {}).items():
for method, operation in path_item.items():
operation_id = operation.get("operationId")
tags = operation.get("tags", [])
# 分类存储操作ID
资料来源:fastapi_mcp/server.py:35-60
HTTP 请求转发架构
请求转发流程
sequenceDiagram
participant MCP as MCP Client
participant MCP_Server as FastApiMCP Server
participant HTTP_Client as httpx.AsyncClient
participant FastAPI as FastAPI App
MCP_Server->>FastAPI: 提取 OpenAPI Schema
MCP->>MCP_Server: 调用 MCP Tool
MCP_Server->>MCP_Server: 解析工具参数
MCP_Server->>HTTP_Client: 构造 HTTP 请求
HTTP_Client->>FastAPI: 转发请求
FastAPI-->>HTTP_Client: 返回响应
HTTP_Client-->>MCP_Server: 返回结果
MCP_Server-->>MCP: 返回 Tool ResultHTTP 方法路由
| HTTP 方法 | 调用方式 | 说明 |
|---|---|---|
| GET | client.get(path, params=query, headers=headers) | 查询参数传递 |
| POST | client.post(path, params=query, headers=headers, json=body) | JSON 请求体 |
| PUT | client.put(path, params=query, headers=headers, json=body) | JSON 请求体 |
| DELETE | client.delete(path, params=query, headers=headers) | 查询参数传递 |
| PATCH | client.patch(path, params=query, headers=headers, json=body) | JSON 请求体 |
资料来源:fastapi_mcp/server.py:12-26
自定义 HTTP 客户端
框架支持注入自定义 httpx.AsyncClient 实例,用于满足特殊网络配置需求:
from fastapi_mcp import FastApiMCP
custom_client = httpx.AsyncClient(
timeout=60.0,
proxies={"http://": "http://proxy.example.com"},
)
mcp = FastApiMCP(
app,
http_client=custom_client,
)
认证与授权系统
认证配置架构
graph LR
A[OAuthMetadata] -->|issuer| B[授权服务器标识]
A -->|authorization_endpoint| C[授权端点]
A -->|token_endpoint| D[令牌端点]
A -->|scopes_supported| E[支持的权限范围]
F[AuthConfig] -->|dependencies| G[FastAPI 依赖项]
F -->|oauth_metadata| H[自定义 OAuth 元数据]AuthConfig 数据模型
| 字段 | 类型 | 默认值 | 说明 |
|---|---|---|---|
version | Literal["2025-03-26"] | "2025-03-26" | MCP 规范版本 |
dependencies | Sequence[Depends] | None | FastAPI 认证依赖链 |
issuer | str | None | OAuth 发行方 URL |
oauth_metadata_url | StrHttpUrl | None | OAuth 元数据端点 |
authorize_url | StrHttpUrl | None | 授权端点 URL |
token_endpoint | StrHttpUrl | None | 令牌端点 URL |
资料来源:fastapi_mcp/types.py:100-150
认证使用示例
from fastapi import Depends, HTTPException
from fastapi.security import HTTPBearer
from fastapi_mcp import FastApiMCP, AuthConfig
token_auth_scheme = HTTPBearer()
async def authenticate_request(request: Request, token: str = Depends(token_auth_scheme)):
payload = verify_token(request, token)
if payload is None:
raise HTTPException(status_code=401, detail="Unauthorized")
return payload
mcp = FastApiMCP(
app,
auth_config=AuthConfig(dependencies=[Depends(authenticate_request)]),
)
资料来源:examples/08_auth_example_token_passthrough.py:20-35
头部信息转发机制
默认转发配置
框架默认自动转发 Authorization 头部,可通过配置扩展:
mcp = FastApiMCP(
app,
headers=["authorization", "x-custom-header", "x-request-id"],
)
头部转发流程
graph TD
A[MCP Request] --> B[提取配置的 headers]
B --> C{headers 参数配置}
C -->|["authorization"]| D[转发 Authorization]
C -->|["x-custom-header"]| E[转发自定义头部]
C -->|未配置| F[默认转发 authorization]
D --> G[合并至 HTTP 请求]
E --> G
F --> G资料来源:fastapi_mcp/server.py:80-90
工具描述生成
描述信息构建流程
框架从 OpenAPI schema 自动提取端点描述信息:
graph TD
A[OpenAPI Schema] --> B[解析路径参数]
A --> C[解析查询参数]
A --> D[解析请求体]
A --> E[解析响应 schema]
B --> F[构建 Tool 描述]
C --> F
D --> F
E --> F参数分类处理
# 按位置分类参数
path_params = [] # 路径参数 (in: path)
query_params = [] # 查询参数 (in: query)
header_params = [] # 请求头参数 (in: header)
for param in operation.get("parameters", []):
param_name = param.get("name")
param_in = param.get("in")
required = param.get("required", False)
if param_in == "path":
path_params.append((param_name, param))
elif param_in == "query":
query_params.append((param_name, param))
elif param_in == "header":
header_params.append((param_name, param))
资料来源:fastapi_mcp/openapi/convert.py:30-50
目录结构
fastapi_mcp/
├── __init__.py # 公共 API 导出
├── server.py # FastApiMCP 主类实现
├── types.py # 数据模型定义
└── openapi/
└── convert.py # OpenAPI 转换逻辑
| 模块 | 导出内容 |
|---|---|
fastapi_mcp.FastApiMCP | 主服务器类 |
fastapi_mcp.AuthConfig | 认证配置类 |
fastapi_mcp.OAuthMetadata | OAuth 元数据类 |
资料来源:fastapi_mcp/__init__.py:1-25
使用模式
基础模式
from fastapi import FastAPI
from fastapi_mcp import FastApiMCP
app = FastAPI()
# 添加 MCP 服务器
mcp = FastApiMCP(app)
# 挂载到 FastAPI 应用
mcp.mount_http()
高级过滤模式
# 按操作 ID 过滤
mcp = FastApiMCP(
app,
include_operations=["get_item", "list_items"],
)
# 按标签过滤
mcp = FastApiMCP(
app,
exclude_tags=["internal"],
)
资料来源:examples/03_custom_exposed_endpoints_example.py:15-30
总结
FastAPI-MCP 的系统架构采用模块化设计,核心通过 FastApiMCP 类协调以下组件:
- 工具发现模块:解析 OpenAPI schema,自动转换为 MCP 工具
- 工具过滤模块:支持基于操作 ID 和标签的灵活过滤
- 请求转发模块:将 MCP 调用转换为标准 HTTP 请求
- 认证模块:集成 FastAPI 依赖注入系统,支持 OAuth 2.0
该架构实现了零配置集成,同时保留了高度的可定制性,满足从简单到复杂的各类应用场景需求。
资料来源:[fastapi_mcp/server.py:50-75]()
传输层
FastAPI-MCP 支持两种传输层协议,用于在 MCP 客户端和 FastAPI 应用之间进行通信。传输层是 MCP 服务器与外部客户端建立连接的核心组件,负责处理请求路由、消息传输和协议握手。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
传输层概述
FastAPI-MCP 实现了 MCP 协议的传输层抽象,提供两种传输方式:
| 传输类型 | 方法 | 说明 |
|---|---|---|
| HTTP Streamable | mount_http() | 基于 HTTP 的流式传输,推荐的生产环境方案 |
| SSE | mount_sse() | 基于 Server-Sent Events 的传输方式 |
两种传输方式都遵循 MCP 协议规范,但在连接管理和消息传输机制上有所不同。mount_http() 是当前推荐的默认选择,而 mount_sse() 则提供了基于事件的实时通信能力。
资料来源:server.py:203-250
HTTP Streamable 传输
HTTP Streamable 传输是 FastAPI-MCP 推荐的传输层实现。它基于 HTTP 协议,提供可靠的双向通信能力。
工作原理
HTTP Streamable 传输使用 FastAPI 的 ASGI 接口与 MCP 服务器进行交互。当 MCP 客户端发起连接请求时,HTTP 客户端会通过 httpx.AsyncClient 转发请求到 FastAPI 应用。
# 基础使用示例
from fastapi_mcp import FastApiMCP
mcp = FastApiMCP(app)
mcp.mount_http() # 使用 HTTP Streamable 传输
资料来源:examples/01_basic_usage_example.py:1-15
HTTP 客户端配置
在初始化 FastApiMCP 时,可以通过 http_client 参数传入自定义的 httpx.AsyncClient 实例:
import httpx
from fastapi_mcp import FastApiMCP
# 自定义 HTTP 客户端
custom_client = httpx.AsyncClient(
timeout=30.0,
follow_redirects=True
)
mcp = FastApiMCP(
app,
http_client=custom_client
)
mcp.mount_http()
超时配置
默认情况下,HTTP 请求超时设置为 10 秒。可以通过自定义客户端调整:
mcp = FastApiMCP(
app,
http_client=httpx.AsyncClient(
timeout=60.0 # 自定义超时时间
)
)
资料来源:server.py:120-130
SSE 传输
SSE (Server-Sent Events) 传输提供基于事件的单向服务器推送能力,适用于需要服务器主动推送消息的场景。
端点注册
SSE 传输在 FastAPI 应用中注册两个主要端点:
| 端点 | HTTP 方法 | 用途 |
|---|---|---|
{mount_path} | GET | MCP 连接建立端点 |
{mount_path}/messages/ | POST | 消息处理端点 |
工作流程
graph TD
A[MCP 客户端] -->|建立 SSE 连接| B[/mcp GET 端点]
A -->|发送消息| C[/mcp/messages/ POST 端点]
B -->|SSE 流| A
C -->|响应| A
B --> D[FastApiSseTransport]
C --> D资料来源:server.py:210-240
SSE 挂载方法
from fastapi_mcp import FastApiMCP
mcp = FastApiMCP(app)
mcp.mount_sse(router=app, mount_path="/mcp")
SSE 传输参数
| 参数 | 类型 | 说明 | |
|---|---|---|---|
router | FastAPI \ | APIRouter | FastAPI 应用或路由实例 |
mount_path | str | SSE 端点的挂载路径 | |
dependencies | Sequence[Depends] | 可选的依赖注入列表 |
资料来源:server.py:152-200
传输层选择指南
推荐场景
graph LR
A[需要可靠双向通信] --> B[选择 HTTP Streamable]
C[需要服务端推送] --> D[选择 SSE]
E[生产环境部署] --> B
F[简单集成场景] --> B| 场景 | 推荐传输 |
|---|---|
| 生产环境 | HTTP Streamable |
| 需要服务端推送 | SSE |
| 简单演示 | HTTP Streamable |
| 与现有 MCP 客户端兼容 | HTTP Streamable |
传输层与认证
传输层支持与认证系统集成。可以通过依赖注入为传输端点添加认证保护:
from fastapi import Depends
from fastapi.security import HTTPBearer
from fastapi_mcp import FastApiMCP, AuthConfig
token_auth_scheme = HTTPBearer()
mcp = FastApiMCP(
app,
auth_config=AuthConfig(
dependencies=[Depends(token_auth_scheme)]
)
)
mcp.mount_http()
这种方式确保所有通过传输层访问的端点都需要进行身份验证。
资料来源:examples/08_auth_example_token_passthrough.py:1-50
传输层初始化参数
FastApiMCP 类中与传输层相关的初始化参数:
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
http_client | httpx.AsyncClient | None | 自定义 HTTP 客户端 |
headers | List[str] | ["authorization"] | 转发到工具调用的 HTTP 头列表 |
mcp = FastApiMCP(
app,
http_client=custom_client,
headers=["authorization", "x-custom-header"] # 自定义转发头
)
资料来源:server.py:100-120
迁移指南
废弃的 mount() 方法
早期版本中的 mount() 方法已被废弃,应使用 mount_http() 或 mount_sse() 替代:
# ❌ 已废弃
mcp.mount(app, transport="sse")
# ✅ 推荐方式
mcp.mount_http()
# 或
mcp.mount_sse(app, mount_path="/mcp")
使用废弃方法时,系统会发出警告:
DeprecationWarning: mount() is deprecated and will be removed in a future version.
Use mount_http() for HTTP transport (recommended) or mount_sse() for SSE transport instead.
资料来源:server.py:250-270
总结
FastAPI-MCP 的传输层提供了灵活的双协议支持,开发者可以根据具体需求选择合适的传输方式。HTTP Streamable 传输是生产环境的推荐选择,提供稳定可靠的通信能力;SSE 传输则为需要服务端推送的场景提供了替代方案。
资料来源:[server.py:203-250]()
OpenAPI 到 MCP 转换
OpenAPI 到 MCP 转换是 FastAPI-MCP 框架的核心功能,负责将 FastAPI 应用的 OpenAPI schema 自动转换为 MCP (Model Context Protocol) 工具。该转换过程保留了 FastAPI 端点的完整类型信息、参数定义、请求体结构和响应模式,使得 MCP 客户端能够以类型安全的方式调用 FastAPI 服务。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
概述
OpenAPI 到 MCP 转换是 FastAPI-MCP 框架的核心功能,负责将 FastAPI 应用的 OpenAPI schema 自动转换为 MCP (Model Context Protocol) 工具。该转换过程保留了 FastAPI 端点的完整类型信息、参数定义、请求体结构和响应模式,使得 MCP 客户端能够以类型安全的方式调用 FastAPI 服务。
转换模块位于 fastapi_mcp/openapi/ 目录下,主要由 convert.py 实现核心转换逻辑,utils.py 提供辅助工具函数。
核心转换函数
convert_openapi_to_mcp_tools
convert_openapi_to_mcp_tools 是整个转换流程的核心入口函数,定义于 fastapi_mcp/openapi/convert.py:
def convert_openapi_to_mcp_tools(
openapi_schema: Dict[str, Any],
describe_all_responses: bool = False,
describe_full_response_schema: bool = False,
) -> Tuple[List[types.Tool], Dict[str, Dict[str, Any]]]:
参数说明:
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
openapi_schema | Dict[str, Any] | 必需 | 完整的 OpenAPI schema 对象 |
describe_all_responses | bool | False | 是否在工具描述中包含所有可能的响应模式 |
describe_full_response_schema | bool | False | 是否在工具描述中包含完整的 JSON 响应 schema |
返回值:
返回一个元组 (List[types.Tool], Dict[str, Dict[str, Any]]),包含:
- MCP tools 列表,用于 MCP 协议交互
- operation ID 到操作详情的映射,用于后续 HTTP 执行
资料来源:fastapi_mcp/openapi/convert.py:30-48
转换流程架构
graph TD
A[OpenAPI Schema] --> B[resolve_schema_references]
B --> C[遍历所有 paths]
C --> D{检查 HTTP 方法}
D -->|get/post/put/delete/patch| E[提取 operationId]
D -->|其他方法| F[跳过并记录警告]
E --> G[获取操作元数据]
G --> H[构建工具描述]
H --> I[组织参数类型]
I --> J[处理请求体]
J --> K[生成 MCP Tool]
K --> L[返回工具列表]
style F fill:#ffcccc
style L fill:#ccffcc详细的转换步骤
第一步:Schema 引用解析
在开始遍历操作之前,转换器首先调用 resolve_schema_references 函数解析 OpenAPI schema 中的所有 $ref 引用:
resolved_openapi_schema = resolve_schema_references(openapi_schema, openapi_schema)
这一步骤确保后续处理能够访问完整的 schema 定义,避免因引用未解析而导致的类型信息丢失。
资料来源:fastapi_mcp/openapi/convert.py:50
第二步:路径与方法遍历
转换器遍历 OpenAPI schema 中定义的所有路径,并为每个路径检查所有支持的 HTTP 方法:
for path, path_item in resolved_openapi_schema.get("paths", {}).items():
for method, operation in path_item.items():
if method not in ["get", "post", "put", "delete", "patch"]:
logger.warning(f"Skipping non-HTTP method: {method}")
continue
支持的 HTTP 方法:
| 方法 | 说明 | 工具调用方式 |
|---|---|---|
GET | 获取资源 | client.get(path, params=query, headers=headers) |
POST | 创建资源 | client.post(path, params=query, headers=headers, json=body) |
PUT | 完全更新资源 | client.put(path, params=query, headers=headers, json=body) |
DELETE | 删除资源 | client.delete(path, params=query, headers=headers) |
PATCH | 部分更新资源 | client.patch(path, params=query, headers=headers, json=body) |
资料来源:fastapi_mcp/openapi/convert.py:53-58
第三步:参数分类
转换器将操作参数按位置分为四类:
path_params = []
query_params = []
header_params = []
body_params = []
for param in operation.get("parameters", []):
param_name = param.get("name")
param_in = param.get("in")
required = param.get("required", False)
if param_in == "path":
path_params.append((param_name, param))
elif param_in == "query":
query_params.append((param_name, param))
elif param_in == "header":
header_params.append((param_name, param))
参数分类表:
| 参数位置 | 变量名 | 说明 | 必填检查 |
|---|---|---|---|
path | path_params | URL 路径参数 | 始终为 required=True |
query | query_params | URL 查询字符串参数 | 可选 |
header | header_params | HTTP 请求头参数 | 可选 |
body | body_params | 请求体参数 | 需检查 required 字段 |
资料来源:fastapi_mcp/openapi/convert.py:78-92
第四步:请求体处理
当操作包含 requestBody 时,转换器解析其内容类型和 schema 定义:
request_body = operation.get("requestBody", {})
if request_body and "content" in request_body:
content_type = next(iter(request_body["content"]), None)
if content_type and "schema" in request_body["content"][content_type]:
schema = request_body["content"][content_type]["schema"]
if "properties" in schema:
for prop_name, prop_schema in schema["properties"].items():
body_params.append((prop_name, prop_schema))
第五步:工具描述构建
转换器为每个 MCP 工具生成详细的描述信息,包括端点路径、方法、参数和响应模式:
tool_description += response_info
描述信息包含:
- 端点路径和 HTTP 方法
- 路径参数说明
- 查询参数说明
- 请求体结构(JSON Schema)
- 响应模式(可选,取决于配置)
响应描述控制
describe_all_responses 选项
当 describe_all_responses=True 时,工具描述将包含该端点所有可能的响应模式,包括不同的状态码对应的 schema:
{
"200": { "description": "Successful Response", "content": {...} },
"422": { "description": "Validation Error", "content": {...} }
}
describe_full_response_schema 选项
当 describe_full_response_schema=True 时,工具描述将包含完整的 JSON Schema 而不仅仅是摘要信息,便于 MCP 客户端进行详细的类型推断。
资料来源:fastapi_mcp/openapi/convert.py:35-40
工具函数模块
辅助函数列表
| 函数名 | 功能 |
|---|---|
resolve_schema_references | 递归解析 OpenAPI schema 中所有的 $ref 引用 |
clean_schema_for_display | 清理 schema,移除不适合显示的内部元数据 |
generate_example_from_schema | 根据 schema 定义生成示例数据 |
get_single_param_type_from_schema | 从 schema 中提取单一参数类型信息 |
这些工具函数确保转换过程中的类型安全和信息完整性。
过滤机制
FastApiMCP 类支持对转换后的工具进行细粒度过滤:
def _filter_tools(self, tools: List[types.Tool], openapi_schema: Dict[str, Any]) -> List[types.Tool]:
过滤选项:
| 选项 | 类型 | 说明 |
|---|---|---|
include_operations | List[str] | 仅包含指定 operation ID 的工具 |
exclude_operations | List[str] | 排除指定 operation ID 的工具 |
include_tags | List[str] | 仅包含指定标签的工具 |
exclude_tags | List[str] | 排除指定标签的工具 |
使用限制:
include_operations和exclude_operations不可同时使用include_tags和exclude_tags不可同时使用- 操作过滤和标签过滤可以组合使用(使用贪婪匹配)
资料来源:fastapi_mcp/server.py:75-90
完整使用示例
from fastapi import FastAPI
from fastapi_mcp import FastApiMCP
app = FastAPI()
# 默认转换所有端点
mcp = FastApiMCP(app)
# 自定义转换配置
mcp = FastApiMCP(
app,
name="Custom MCP Server",
describe_all_responses=True,
describe_full_response_schema=True,
include_operations=["get_item", "list_items"],
exclude_operations=["delete_item"],
headers=["authorization", "x-custom-header"],
)
# 挂载 MCP 服务器
mcp.mount()
资料来源:examples/02_full_schema_description_example.py
MCP Tool 结构
转换后的 MCP Tool 包含以下核心字段:
| 字段 | 说明 |
|---|---|
name | 工具名称(通常为 operationId) |
description | 工具描述,包含端点信息和参数说明 |
inputSchema | JSON Schema 格式的输入参数定义 |
annotations | 可选的注释信息 |
HTTP 执行映射
对于每个转换后的工具,系统维护一个 operation_map,将工具名称映射到实际的执行信息:
operation_map = {
"get_item": {
"method": "get",
"path": "/items/{item_id}",
"operation_id": "get_item"
},
"list_items": {
"method": "get",
"path": "/items",
"operation_id": "list_items"
}
}
此映射用于后续通过 httpx 客户端实际执行 HTTP 请求。
错误处理
转换过程中遇到的问题会被记录但不会中断流程:
- 非 HTTP 方法(如
options、head)会被跳过并记录警告 - 缺少
operationId的操作会被跳过 - Schema 解析错误会被捕获并记录日志
类型安全保证
FastAPI-MCP 的转换过程通过以下机制确保类型安全:
- Pydantic 模型验证:使用 Pydantic 进行类型验证
- 引用解析:确保所有
$ref被正确解析 - Schema 清理:移除不兼容的字段,确保 JSON Schema 有效性
- 示例生成:根据 schema 自动生成示例值,便于测试
资料来源:fastapi_mcp/types.py
总结
OpenAPI 到 MCP 转换是 FastAPI-MCP 框架的桥梁,它:
- 自动化 - 无需手动定义 MCP 工具,OpenAPI schema 自动转换
- 类型安全 - 保留完整的类型信息,支持强类型调用
- 灵活可控 - 支持细粒度的端点过滤和描述控制
- 文档完整 - 自动保留并增强 FastAPI 端点的文档信息
该模块使得开发者能够零配置地将现有的 FastAPI 应用暴露为 MCP 服务器,同时保持原有的类型安全性和开发体验。
资料来源:[fastapi_mcp/openapi/convert.py:30-48]()
认证与授权
FastAPI-MCP 内置了基于 FastAPI 依赖注入系统的认证与授权机制。这套机制允许开发者复用现有的 FastAPI 认证逻辑,无需额外配置即可保护 MCP 工具的访问。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
系统架构
graph TD
A[MCP Client] --> B[FastAPI-MCP Server]
B --> C{FastAPI Dependencies}
C -->|验证通过| D[MCP Tool]
C -->|验证失败| E[401/403 错误]
F[OAuth Provider] -->|OAuth Flow| B
subgraph "认证配置层"
G[AuthConfig]
H[OAuthMetadata]
I[Custom Metadata]
end核心类型
AuthConfig
AuthConfig 是配置 MCP 认证的核心类,位于 fastapi_mcp/types.py。支持 OAuth 2.0 授权流程和自定义元数据配置。
| 参数 | 类型 | 说明 |
|---|---|---|
version | Literal["2025-03-26"] | MCP 规范版本,当前仅支持 "2025-03-26" |
dependencies | Optional[Sequence[Depends]] | FastAPI 依赖列表,用于身份验证 |
setup_proxies | bool | 是否设置 OAuth 代理端点 |
client_id | Optional[str] | OAuth 客户端 ID |
client_secret | Optional[str] | OAuth 客户端密钥 |
issuer | Optional[str] | OAuth 服务器签发者 |
oauth_metadata_url | Optional[StrHttpUrl] | OAuth 提供商元数据端点 URL |
authorize_url | Optional[StrHttpUrl] | 授权端点 URL |
metadata_path | str | 元数据端点路径,默认 "/.well-known/oauth-authorization-server" |
audience | Optional[str] | OAuth 受众 |
default_scope | Optional[str] | 默认权限范围 |
custom_oauth_metadata | Optional[OAuthMetadataDict] | 自定义 OAuth 元数据 |
setup_fake_dynamic_registration | bool | 是否设置虚假动态客户端注册端点 |
资料来源:fastapi_mcp/types.py:100-150
OAuthMetadata
OAuth 2.0 服务器元数据结构,符合 RFC 8414 规范:
| 参数 | 类型 | 说明 |
|---|---|---|
issuer | StrHttpUrl | 授权服务器的签发者标识符 |
authorization_endpoint | Optional[StrHttpUrl] | 授权端点 URL |
token_endpoint | StrHttpUrl | 令牌端点 URL |
scopes_supported | List[str] | 支持的 OAuth 范围列表,默认 ["openid", "profile", "email"] |
response_types_supported | List[str] | 支持的响应类型 |
资料来源:fastapi_mcp/types.py:50-90
认证流程
基础 Token 验证
最简单的认证方式是使用 FastAPI 的 HTTPBearer 进行 Token 验证:
sequenceDiagram
participant C as MCP Client
participant S as FastAPI-MCP
participant D as FastAPI Dependencies
C->>S: 请求携带 Authorization Header
S->>D: 注入 dependencies 进行验证
D->>D: 验证 Bearer Token
alt Token 有效
D-->>S: 验证通过
S->>C: 执行 MCP Tool
else Token 无效
D-->>S: 401 Unauthorized
S->>C: 返回错误
end资料来源:examples/08_auth_example_token_passthrough.py
OAuth 授权流程
当配置了完整的 OAuth 设置时,流程如下:
graph LR
A[Client] -->|1. 请求工具| B[FastAPI-MCP]
B -->|2. Token 缺失/无效| C[OAuth Redirect]
C -->|3. 用户授权| D[Auth Provider]
D -->|4. Authorization Code| A
A -->|5. 交换 Token| D
D -->|6. Access Token| A
A -->|7. 请求工具 + Token| B
B -->|8. 执行工具| A使用方式
方式一:Token 验证(简单场景)
适用于内部服务或测试环境,直接验证 Authorization Header 中的 Token:
from fastapi import Depends
from fastapi.security import HTTPBearer
from fastapi_mcp import FastApiMCP, AuthConfig
token_auth_scheme = HTTPBearer()
@app.get("/private")
async def private(token=Depends(token_auth_scheme)):
return token.credentials
mcp = FastApiMCP(
app,
name="Protected MCP",
auth_config=AuthConfig(
dependencies=[Depends(token_auth_scheme)],
),
)
mcp.mount_http()
资料来源:examples/08_auth_example_token_passthrough.py:30-45
方式二:Auth0 集成(生产环境)
生产环境推荐使用 Auth0 等专业 OAuth 提供商:
from fastapi_mcp import FastApiMCP, AuthConfig
mcp = FastApiMCP(
app,
name="MCP With Auth0",
auth_config=AuthConfig(
issuer=f"https://{settings.auth0_domain}/",
authorize_url=f"https://{settings.auth0_domain}/authorize",
oauth_metadata_url=settings.auth0_oauth_metadata_url,
audience=settings.auth0_audience,
client_id=settings.auth0_client_id,
client_secret=settings.auth0_client_secret,
dependencies=[Depends(verify_auth)],
setup_proxies=True,
),
)
资料来源:examples/09_auth_example_auth0.py:45-60
方式三:自定义 OAuth 元数据
当需要使用特定的 OAuth 提供商而非自动发现时:
from fastapi_mcp.auth.proxy import setup_oauth_custom_metadata
setup_oauth_custom_metadata(
app=app,
auth_config=auth_config,
metadata=custom_oauth_metadata,
include_in_schema=False,
)
资料来源:fastapi_mcp/auth/proxy.py:25-45
OAuth 代理端点
FastAPI-MCP 提供多个 OAuth 代理函数,用于搭建完整的授权服务:
| 函数 | 功能 | 路径 |
|---|---|---|
setup_oauth_custom_metadata | 提供自定义 OAuth 元数据 | auth_config.metadata_path |
setup_oauth_metadata_proxy | 代理 OAuth 提供商的元数据端点 | 默认 /.well-known/oauth-authorization-server |
setup_oauth_authorize_proxy | 代理授权端点 | 默认 /oauth/authorize |
setup_oauth_fake_dynamic_register_endpoint | 提供动态客户端注册 | /oauth/register |
资料来源:fastapi_mcp/auth/proxy.py:48-120
元数据代理配置
setup_oauth_metadata_proxy(
app=app,
metadata_url="https://your-oauth-provider.com/.well-known/oauth-authorization-server",
path="/.well-known/oauth-authorization-server",
authorize_path="/oauth/authorize",
register_path="/oauth/register",
)
授权代理配置
setup_oauth_authorize_proxy(
app=app,
client_id="your-client-id",
authorize_url="https://your-provider.com/authorize",
audience="your-audience",
default_scope="openid profile",
)
动态客户端注册
对于需要动态注册客户端的场景:
setup_oauth_fake_dynamic_register_endpoint(
app=app,
client_id="your-client-id",
client_secret="your-client-secret",
)
在 FastApiMCP 中启用认证
将 AuthConfig 传入 FastApiMCP 构造函数即可启用认证:
mcp = FastApiMCP(
app,
name="Protected MCP Server",
auth_config=AuthConfig(
dependencies=[Depends(my_auth_dependency)],
setup_proxies=True,
client_id="...",
client_secret="...",
issuer="https://auth.example.com",
),
)
资料来源:fastapi_mcp/server.py:180-220
认证配置内部处理
在 FastApiMCP 初始化时,会根据 AuthConfig 自动配置认证端点:
def _setup_auth_2025_03_26(self):
if self._auth_config:
if self._auth_config.custom_oauth_metadata:
setup_oauth_custom_metadata(...)
elif self._auth_config.setup_proxies:
setup_oauth_metadata_proxy(...)
setup_oauth_authorize_proxy(...)
if self._auth_config.setup_fake_dynamic_registration:
setup_oauth_fake_dynamic_register_endpoint(...)
资料来源:fastapi_mcp/server.py:220-260
最佳实践
- 生产环境使用 OAuth:不要在生产环境使用简单的 Token 验证,应配置完整的 OAuth 流程
- 敏感信息管理:使用环境变量存储
client_id和client_secret - 依赖注入复用:直接复用现有的 FastAPI 认证依赖,保持代码一致性
- 元数据端点:确保
metadata_path符合 OAuth 规范,默认值为/.well-known/oauth-authorization-server
资料来源:[fastapi_mcp/types.py:100-150]()
端点过滤与自定义
端点过滤与自定义是 FastAPI-MCP 框架提供的重要功能,允许开发者精确控制哪些 FastAPI 端点被暴露为 MCP(Model Context Protocol)工具。通过灵活的配置选项,用户可以基于操作 ID(operationId)或标签(tags)对端点进行细粒度的筛选,从而实现按需暴露接口、保护敏感端点或构建特定的工具集。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
概述
端点过滤与自定义是 FastAPI-MCP 框架提供的重要功能,允许开发者精确控制哪些 FastAPI 端点被暴露为 MCP(Model Context Protocol)工具。通过灵活的配置选项,用户可以基于操作 ID(operationId)或标签(tags)对端点进行细粒度的筛选,从而实现按需暴露接口、保护敏感端点或构建特定的工具集。
这一功能在 CHANGELOG.md 中被首次引入,作为支持"refreshing"功能的一部分,专门用于解决动态添加 FastAPI 路由后的端点管理问题。资料来源:CHANGELOG.md
核心概念
操作 ID(Operation ID)
每个 FastAPI 端点在 OpenAPI 规范中都有一个唯一的 operationId,用于标识该操作。FastAPI-MCP 利用这个标识符来精确定位和过滤特定的端点。
标签(Tags)
标签是 OpenAPI 规范中用于分组和组织端点的机制。一个端点可以拥有多个标签,这为批量过滤提供了便利的条件。
过滤参数
FastAPI-MCP 在 FastApiMCP 类中提供了四个核心过滤参数:
| 参数名 | 类型 | 说明 | 互斥约束 |
|---|---|---|---|
include_operations | Optional[List[str]] | 仅包含指定 operation ID 的端点 | 与 exclude_operations 互斥 |
exclude_operations | Optional[List[str]] | 排除指定 operation ID 的端点 | 与 include_operations 互斥 |
include_tags | Optional[List[str]] | 仅包含具有指定标签的端点 | 与 exclude_tags 互斥 |
exclude_tags | Optional[List[str]] | 排除具有指定标签的端点 | 与 include_tags 互斥 |
资料来源:fastapi_mcp/server.py:60-85
过滤规则
互斥规则
FastAPI-MCP 对过滤参数实施了严格的互斥约束,以确保配置的一致性和可预测性:
- 操作 ID 级别:不能同时使用
include_operations和exclude_operations - 标签级别:不能同时使用
include_tags和exclude_tags - 跨级别组合:可以同时使用操作 ID 过滤和标签过滤
组合过滤行为
当同时使用多种过滤条件时,FastAPI-MCP 采用贪婪模式(greedy approach):任何满足任一条件的端点都会被包含在最终的工具列表中。
graph TD
A[开始过滤流程] --> B{include_operations<br/>存在?}
B -->|是| C[根据 operation ID 包含]
B -->|否| D{exclude_operations<br/>存在?}
D -->|是| E[根据 operation ID 排除]
D -->|否| F{include_tags<br/>存在?}
F -->|是| G[根据标签包含]
F -->|否| H{exclude_tags<br/>存在?}
H -->|是| I[根据标签排除]
H -->|否| J[不进行过滤]
C --> K{同时存在<br/>标签过滤?}
E --> K
G --> L[贪婪模式:<br/>满足任一条件即包含]
I --> L
J --> L
K -->|是| L
L --> M[返回过滤后的工具列表]资料来源:fastapi_mcp/server.py:48-70
过滤实现机制
核心方法:`_filter_tools`
FastApiMCP 类中的 _filter_tools 方法是执行过滤逻辑的核心实现。该方法接收两个参数:
tools:待过滤的工具列表openapi_schema:完整的 OpenAPI 架构定义
def _filter_tools(self, tools: List[types.Tool], openapi_schema: Dict[str, Any]) -> List[types.Tool]:
"""
Filter tools based on operation IDs and tags.
Args:
tools: List of tools to filter
openapi_schema: The OpenAPI schema
Returns:
Filtered list of tools
"""
if (
self._include_operations is None
and self._exclude_operations is None
and self._include_tags is None
and self._exclude_tags is None
):
return tools
资料来源:fastapi_mcp/server.py:35-56
构建标签到操作 ID 的映射
过滤过程中,框架首先构建一个从标签到操作 ID 列表的映射:
operations_by_tag: Dict[str, List[str]] = {}
for path, path_item in openapi_schema.get("paths", {}).items():
for method, operation in path_item.items():
if method not in ["get", "post", "put", "delete", "patch"]:
continue
operation_id = operation.get("operationId")
if not operation_id:
continue
tags = operation.get("tags", [])
for tag in tags:
if tag not in operations_by_tag:
operations_by_tag[tag] = []
operations_by_tag[tag].append(operation_id)
资料来源:fastapi_mcp/server.py:57-74
使用示例
按操作 ID 过滤
#### 仅包含特定操作
from fastapi_mcp import FastApiMCP
# 仅暴露 get_item 和 list_items 两个操作
include_operations_mcp = FastApiMCP(
app,
name="Item API MCP - Included Operations",
include_operations=["get_item", "list_items"],
)
资料来源:examples/03_custom_exposed_endpoints_example.py:18-24
#### 排除特定操作
# 排除 create_item、update_item 和 delete_item 操作
exclude_operations_mcp = FastApiMCP(
app,
name="Item API MCP - Excluded Operations",
exclude_operations=["create_item", "update_item", "delete_item"],
)
资料来源:examples/03_custom_exposed_endpoints_example.py:27-32
按标签过滤
#### 仅包含特定标签
# 仅暴露具有 'items' 标签的端点
include_tags_mcp = FastApiMCP(
app,
name="Item API MCP - Included Tags",
include_tags=["items"],
)
资料来源:examples/03_custom_exposed_endpoints_example.py:35-39
#### 排除特定标签
# 排除具有 'search' 标签的端点
exclude_tags_mcp = FastApiMCP(
app,
name="Item API MCP - Excluded Tags",
exclude_tags=["search"],
)
资料来源:examples/03_custom_exposed_endpoints_example.py:42-46
组合过滤
可以将操作 ID 过滤和标签过滤组合使用:
# 包含具有 'search' 标签或 delete_item 操作的所有端点
combined_include_mcp = FastApiMCP(
app,
name="Item API MCP - Combined Include",
include_operations=["delete_item"],
include_tags=["search"],
)
资料来源:examples/03_custom_exposed_endpoints_example.py:49-55
挂载与部署
过滤后的 MCP 服务器可以通过 mount_http 方法挂载到不同的路径:
# 为不同的过滤配置挂载独立的 MCP 端点
include_operations_mcp.mount_http(mount_path="/include-operations-mcp")
exclude_operations_mcp.mount_http(mount_path="/exclude-operations-mcp")
include_tags_mcp.mount_http(mount_path="/include-tags-mcp")
exclude_tags_mcp.mount_http(mount_path="/exclude-tags-mcp")
combined_include_mcp.mount_http(mount_path="/combined-include-mcp")
资料来源:examples/03_custom_exposed_endpoints_example.py:58-63
典型应用场景
场景一:构建只读 API
如果需要创建一个只读的 MCP 工具集,可以排除所有修改类操作:
mcp_readonly = FastApiMCP(
app,
name="Read-Only API",
exclude_operations=["create_item", "update_item", "delete_item"],
)
场景二:按功能模块隔离
根据业务模块隔离不同的 MCP 服务:
# 仅包含搜索相关功能
mcp_search = FastApiMCP(
app,
name="Search API",
include_tags=["search"],
)
# 仅包含物品管理功能
mcp_items = FastApiMCP(
app,
name="Items API",
include_tags=["items"],
)
场景三:权限分级
通过标签组织不同权限级别的端点:
# 管理员端点(排除公开端点)
mcp_admin = FastApiMCP(
app,
name="Admin API",
include_tags=["admin"],
exclude_operations=["get_public_stats"], # 排除特定敏感操作
)
注意事项
1. operationId 必需
过滤功能依赖 OpenAPI 规范中的 operationId。如果 FastAPI 端点未显式定义 operationId,FastAPI 会自动生成一个,但可能难以预测。建议显式定义:
@app.get("/items/{item_id}", operation_id="get_item")
async def get_item(item_id: int):
return {"item_id": item_id}
2. 标签的作用域
一个端点可以属于多个标签,只要满足任一过滤条件,该端点就会被包含。
3. 无过滤参数时的行为
如果未提供任何过滤参数(include_operations、exclude_operations、include_tags、exclude_tags 均为 None),则所有可发现的端点都将被暴露为 MCP 工具。
相关文档
更新日志
| 版本 | 更新内容 |
|---|---|
| 0.2.0 | 新增端点过滤能力,支持 include_operations、exclude_operations、include_tags、exclude_tags 参数 |
资料来源:[fastapi_mcp/server.py:60-85]()
部署选项
FastAPI-MCP 提供了多种部署方式,以适应不同的应用场景和架构需求。本页面详细介绍如何将 MCP 服务器与现有的 FastAPI 应用集成,以及可用的传输协议和挂载选项。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
概述
FastAPI-MCP 的部署架构基于以下几个核心组件:
graph TD
A[FastAPI 应用] --> B[FastApiMCP 实例]
B --> C[传输层]
C --> D[HTTP/SSE 传输]
C --> E[标准输入输出]
B --> F[MCP 工具注册]
F --> G[API 端点 → MCP 工具]HTTP 挂载部署
HTTP 挂载是最常用的部署方式,允许将 MCP 服务器作为 FastAPI 应用的子路由挂载。
基本挂载
通过 mount_http() 方法将 MCP 端点挂载到指定路径:
from fastapi import FastAPI
from fastapi_mcp import FastApiMCP
app = FastAPI(...)
mcp = FastApiMCP(app, name="My MCP Server")
mcp.mount_http(mount_path="/mcp")
自定义挂载路径
可以为不同的 MCP 实例配置不同的挂载路径,实现多租户或分类服务:
# 包含特定操作的 MCP
include_mcp = FastApiMCP(
app,
name="Item API MCP - Included Operations",
include_operations=["get_item", "list_items"],
)
# 排除特定操作的 MCP
exclude_mcp = FastApiMCP(
app,
name="Item API MCP - Excluded Operations",
exclude_operations=["create_item", "update_item"],
)
# 挂载到不同路径
include_mcp.mount_http(mount_path="/include-operations-mcp")
exclude_mcp.mount_http(mount_path="/exclude-operations-mcp")
资料来源:examples/03_custom_exposed_endpoints_example.py:1-10
传输协议
HTTP 挂载支持两种传输协议:
| 协议 | 说明 | 默认端口 |
|---|---|---|
| SSE (Server-Sent Events) | 适用于需要服务端推送的场景 | 是 |
| Streamable HTTP | 支持流式响应和双向通信 | 可选 |
SSE 传输通过 FastApiSseTransport 实现,提供以下端点:
GET /{mount_path}- MCP 连接端点POST /{mount_path}/messages/- 消息处理端点
资料来源:fastapi_mcp/server.py:140-170
独立服务器部署
对于需要独立运行的场景,FastAPI-MCP 支持将 MCP 服务器作为独立进程启动。
使用方式
独立服务器模式允许 MCP 直接作为独立服务运行:
from examples.shared.apps.items import app
from fastapi_mcp import FastApiMCP
mcp = FastApiMCP(
app,
name="Separate Server",
describe_all_responses=True,
)
if __name__ == "__main__":
mcp.run()
资料来源:examples/04_separate_server_example.py:1-20
独立运行流程
graph LR
A[启动进程] --> B[MCP.run 初始化]
B --> C[创建 STDIO 传输]
C --> D[MCP 协议处理]
D --> E[接收工具调用请求]
E --> F[调用 FastAPI 端点]
F --> G[返回结果到 MCP 客户端]工具动态注册
FastAPI-MCP 支持在服务器运行期间动态注册和更新工具。
reregister_tools 方法
reregister_tools() 方法允许在不重启服务器的情况下更新 MCP 工具列表:
from fastapi import FastAPI
from fastapi_mcp import FastApiMCP, Tool
app = FastAPI(...)
mcp = FastApiMCP(app, name="Dynamic Tools Server")
# 初始工具列表
initial_tools = [
Tool(
name="original_tool",
description="An original tool",
inputSchema={"type": "object", "properties": {}},
)
]
# 注册初始工具
mcp.add_mcp_server(tools=initial_tools)
# 动态添加新工具
new_tools = [
Tool(
name="dynamic_tool",
description="A dynamically added tool",
inputSchema={
"type": "object",
"properties": {
"message": {"type": "string", "description": "A message"}
},
},
)
]
mcp.reregister_tools(tools=new_tools)
资料来源:examples/05_reregister_tools_example.py:1-50
动态注册流程
graph TD
A[运行时] --> B[创建新工具定义]
B --> C[调用 reregister_tools]
C --> D[更新 MCP 服务器工具列表]
D --> E[新工具立即生效]
E --> F[下次 list_tools 调用返回新列表]部署配置参数
mount_http 参数
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
mount_path | str | "/mcp" | MCP 端点挂载的基础路径 |
transport | FastApiSseTransport | 自动创建 | 传输协议实例 |
dependencies | Sequence[Depends] | None | 挂载路径的 FastAPI 依赖项 |
资料来源:fastapi_mcp/server.py:110-130
运行时参数
| 参数 | 类型 | 说明 |
|---|---|---|
http_client | httpx.AsyncClient | 自定义 HTTP 客户端 |
headers | List[str] | 转发的 HTTP 头列表,默认 ["authorization"] |
include_operations | List[str] | 仅包含指定的操作 ID |
exclude_operations | List[str] | 排除指定的操作 ID |
资料来源:fastapi_mcp/server.py:70-100
认证配置部署
在部署时可以通过 auth_config 配置认证功能:
from fastapi import Depends
from fastapi.security import HTTPBearer
from fastapi_mcp import FastApiMCP, AuthConfig
token_auth_scheme = HTTPBearer()
mcp = FastApiMCP(
app,
name="Protected MCP",
auth_config=AuthConfig(
dependencies=[Depends(token_auth_scheme)],
),
)
mcp.mount_http()
资料来源:examples/08_auth_example_token_passthrough.py:1-30
多实例部署架构
graph TD
A[FastAPI 应用] --> B[主 MCP 实例<br/>包含全部工具]
A --> C[过滤 MCP 实例 1<br/>include_operations]
A --> D[过滤 MCP 实例 2<br/>include_tags]
B --> E[/mcp]
C --> F[/filtered-operations]
D --> G[/filtered-tags]
style E fill:#90EE90
style F fill:#87CEEB
style G fill:#DDA0DD部署环境要求
| 要求 | 最低版本 | 推荐版本 |
|---|---|---|
| Python | 3.10+ | 3.12 |
| FastAPI | - | 最新稳定版 |
| httpx | - | 最新稳定版 |
| MCP SDK | - | 兼容版本 |
常见部署场景
场景一:单一路由挂载
适用于简单的微服务架构:
mcp = FastApiMCP(app, name="Simple MCP")
mcp.mount_http(mount_path="/mcp")
场景二:多实例隔离
适用于需要按功能模块隔离访问权限的场景:
admin_mcp = FastApiMCP(app, include_tags=["admin"])
user_mcp = FastApiMCP(app, include_tags=["user"])
admin_mcp.mount_http(mount_path="/admin-mcp")
user_mcp.mount_http(mount_path="/user-mcp")
场景三:认证保护部署
适用于需要保护敏感 API 的场景:
mcp = FastApiMCP(
app,
auth_config=AuthConfig(dependencies=[Depends(verify_token)])
)
mcp.mount_http()
下一步
资料来源:[examples/03_custom_exposed_endpoints_example.py:1-10](https://github.com/tadata-org/fastapi_mcp/blob/main/examples/03_custom_exposed_endpoints_example.py)
配置与自定义
FastAPI-MCP 提供了丰富的配置选项,使开发者能够根据实际需求灵活定制 MCP 服务器的行为。本文详细介绍 FastAPI-MCP 的各项配置参数、挂载方式、认证配置以及工具过滤机制。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
核心配置参数
FastApiMCP 类的构造函数是主要的配置入口,支持多种参数来控制 MCP 服务器的功能和行为。
基础配置
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
name | str | 必填 | MCP 服务器名称 |
description | str | None | 服务器描述信息 |
describe_all_responses | bool | False | 是否在工具描述中包含所有可能的响应 schema |
describe_full_response_schema | bool | False | 是否包含完整的 JSON schema 响应信息 |
基础配置允许开发者为 MCP 服务器设置标识信息和响应描述的详细程度。当 describe_all_responses 设置为 True 时,工具描述将包含端点的所有可能响应状态码对应的 schema;当 describe_full_response_schema 设置为 True 时,会输出完整的 JSON schema 结构。资料来源:fastapi_mcp/server.py:54-73
HTTP 客户端配置
http_client: Annotated[
Optional[httpx.AsyncClient],
Doc("用于向 FastAPI 应用发起 API 调用的自定义 HTTP 客户端"),
] = None
通过 http_client 参数,开发者可以传入自定义的 httpx.AsyncClient 实例来自定义超时、代理、连接池等 HTTP 行为。资料来源:fastapi_mcp/server.py:79-85
标头转发配置
headers: Annotated[
List[str],
Doc("从 MCP 请求中转发到每个工具调用的 HTTP 标头名称列表"),
] = ["authorization"]
headers 参数控制哪些 HTTP 标头会被转发到后端 API 调用。默认情况下只转发 authorization 标头,开发者可以根据需要添加其他标头。资料来源:fastapi_mcp/server.py:119-126
挂载方式
FastAPI-MCP 支持多种挂载方式,以适应不同的部署场景。
graph TD
A[FastAPI 应用] --> B[选择挂载方式]
B --> C[mount_http]
B --> D[mount_sse]
B --> E[mount - 已废弃]
C --> F[HTTP 传输<br/>推荐方式]
D --> G[SSE 传输]
E --> H[兼容性保留]
F --> I[高性能场景]
G --> J[实时推送场景]HTTP 传输(推荐)
mount_http() 是推荐的挂载方式,提供更高的性能和更简单的部署体验:
from fastapi_mcp import FastApiMCP
mcp = FastApiMCP(app, name="My MCP Server")
mcp.mount_http()
该方式会自动注册必要的端点来处理 MCP 协议的通信。资料来源:fastapi_mcp/server.py:217-224
SSE 传输
mount_sse() 使用 Server-Sent Events 进行通信:
mcp.mount_sse(router=app, mount_path="/mcp")
SSE 传输方式会在指定路径下注册 MCP 连接端点,适用于需要服务端推送功能的场景。资料来源:examples/06_custom_mcp_router_example.py:1-25
自定义路由器挂载
可以通过 mount_http() 方法将 MCP 服务器挂载到特定的 APIRouter 上,从而实现自定义挂载路径:
from fastapi import APIRouter
from fastapi_mcp import FastApiMCP
other_router = APIRouter(prefix="/other/route")
app.include_router(other_router)
mcp = FastApiMCP(app)
mcp.mount_http(other_router)
执行上述代码后,MCP 服务器将只在 /other/route/mcp 路径下可用,而不会挂载到应用根路径。资料来源:examples/06_custom_mcp_router_example.py:11-18
超时配置
通过自定义 httpx.AsyncClient 可以配置 HTTP 请求超时时间:
import httpx
from fastapi_mcp import FastApiMCP
custom_client = httpx.AsyncClient(
timeout=30.0 # 设置 30 秒超时
)
mcp = FastApiMCP(app, http_client=custom_client)
mcp.mount_http()
默认超时时间为 10 秒,开发者应根据实际业务需求调整该值。资料来源:fastapi_mcp/server.py:93-96
认证配置
FastAPI-MCP 利用 FastAPI 原生的依赖注入系统来实现认证功能。
AuthConfig 配置
AuthConfig 类用于配置 MCP 服务器的认证行为:
| 参数 | 类型 | 说明 |
|---|---|---|
dependencies | List[Depends] | FastAPI 依赖项列表,用于验证请求身份 |
Bearer Token 认证示例
以下示例展示如何配置基于 Bearer Token 的认证:
from fastapi import Depends
from fastapi.security import HTTPBearer
from fastapi_mcp import FastApiMCP, AuthConfig
token_auth_scheme = HTTPBearer()
@app.get("/private")
async def private(token=Depends(token_auth_scheme)):
return token.credentials
mcp = FastApiMCP(
app,
name="Protected MCP",
auth_config=AuthConfig(
dependencies=[Depends(token_auth_scheme)],
),
)
mcp.mount_http()
在此配置下,所有 MCP 工具调用都必须携带有效的 Authorization 标头。资料来源:examples/08_auth_example_token_passthrough.py:1-45
认证标头转发
在 MCP 客户端配置中,需要通过标头传递认证信息:
{
"mcpServers": {
"remote-example": {
"command": "npx",
"args": [
"mcp-remote",
"http://localhost:8000/mcp",
"--header",
"Authorization:${AUTH_HEADER}"
]
}
}
}
工具过滤机制
FastAPI-MCP 提供了灵活的端点过滤功能,支持按操作 ID 和标签两种方式进行筛选。
graph TD
A[OpenAPI Schema] --> B{过滤条件判断}
B --> C{include_operations<br/>已设置?}
C -->|是| D[按操作 ID 包含]
C -->|否| E{exclude_operations<br/>已设置?}
E -->|是| F[按操作 ID 排除]
E -->|否| G{include_tags<br/>已设置?}
G -->|是| H[按标签包含]
G -->|否| I{exclude_tags<br/>已设置?}
I -->|是| J[按标签排除]
I -->|否| K[返回全部工具]
D --> L[过滤后的工具列表]
F --> L
H --> L
J --> L
K --> L过滤参数说明
| 参数 | 类型 | 说明 | 互斥约束 |
|---|---|---|---|
include_operations | List[str] | 仅包含指定操作 ID 的工具 | 与 exclude_operations 互斥 |
exclude_operations | List[str] | 排除指定操作 ID 的工具 | 与 include_operations 互斥 |
include_tags | List[str] | 仅包含指定标签的工具 | 与 exclude_tags 互斥 |
exclude_tags | List[str] | 排除指定标签的工具 | 与 include_tags 互斥 |
资料来源:fastapi_mcp/server.py:86-118
使用示例
按操作 ID 过滤:
# 仅包含特定操作
include_mcp = FastApiMCP(
app,
name="Item API MCP",
include_operations=["get_item", "list_items"],
)
# 排除特定操作
exclude_mcp = FastApiMCP(
app,
name="Item API MCP",
exclude_operations=["create_item", "update_item", "delete_item"],
)
按标签过滤:
# 仅包含特定标签
include_tags_mcp = FastApiMCP(
app,
name="Item API MCP",
include_tags=["items"],
)
# 排除特定标签
exclude_tags_mcp = FastApiMCP(
app,
name="Item API MCP",
exclude_tags=["search"],
)
组合过滤:
# 可以组合操作 ID 和标签过滤
combined_mcp = FastApiMCP(
app,
name="Item API MCP - Combined",
include_operations=["get_item"],
include_tags=["search"],
)
组合过滤采用贪婪策略,只要满足任一条件的端点都会被包含。资料来源:examples/03_custom_exposed_endpoints_example.py:1-60
响应 Schema 配置
描述全部响应
当 describe_all_responses=True 时,工具描述将包含端点的所有可能响应:
mcp = FastApiMCP(
app,
name="Full Response MCP",
describe_all_responses=True,
)
完整 Schema 描述
当 describe_full_response_schema=True 时,工具描述将包含完整的 JSON Schema:
mcp = FastApiMCP(
app,
name="Full Schema MCP",
describe_full_response_schema=True,
)
依赖管理
运行时依赖
添加运行时所需的依赖包:
uv add new-package
开发依赖
添加开发、测试或 CI 所需的依赖包:
uv add --group dev new-package
添加依赖后需要同时提交 pyproject.toml 和 uv.lock 文件。资料来源:CONTRIBUTING.md:85-96
配置最佳实践
生产环境配置
- 显式设置超时:根据业务需求配置合理的 HTTP 超时时间
- 配置认证:生产环境应启用认证机制
- 过滤敏感端点:使用
exclude_operations排除管理类接口 - 配置标头转发:仅转发必要的认证标头
开发环境配置
- 启用完整 Schema:开发时可设置
describe_all_responses=True以获得更详细的文档 - 包含全部工具:开发阶段不过滤任何端点
- 调整超时时间:开发环境可设置更长的超时便于调试
常用配置组合
# 最小配置(仅必需参数)
mcp = FastApiMCP(app, name="My API")
# 完整配置
mcp = FastApiMCP(
app,
name="My API",
description="Production MCP Server",
describe_all_responses=True,
describe_full_response_schema=True,
include_tags=["api"],
exclude_operations=["internal_debug"],
auth_config=AuthConfig(dependencies=[Depends(token_auth_scheme)]),
headers=["authorization", "x-api-key"],
)
mcp.mount_http()资料来源:[fastapi_mcp/server.py:86-118]()
示例集
fastapimcp 项目提供了一套完整的示例代码,帮助开发者快速上手并深入理解该库的各项功能。示例集涵盖了从基础使用到高级配置的各种场景,是学习和实践 fastapimcp 的重要资源。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
继续阅读本节完整说明和来源证据。
示例概览
示例目录位于仓库根目录的 examples/ 文件夹下,每个示例都针对一个特定的功能点进行演示。所有示例都依赖共享的 FastAPI 应用组件,这些组件位于 examples/shared/apps/items.py 中定义了一个包含增删改查操作的 Items API。
示例文件清单
| 文件名 | 功能描述 | 难度等级 |
|---|---|---|
| 01_basic_usage_example.py | 基础使用:如何将 MCP 服务器添加到 FastAPI 应用 | ⭐ 入门 |
| 02_full_schema_description_example.py | 完整的 OpenAPI Schema 描述配置 | ⭐ 入门 |
| 03_custom_exposed_endpoints_example.py | 自定义暴露的端点:通过 operation ID 和 tag 过滤 | ⭐⭐ 进阶 |
| 04_separate_server_example.py | 独立 MCP 服务器运行方式 | ⭐⭐ 进阶 |
| 05_reregister_tools_example.py | 重新注册 MCP 工具 | ⭐⭐ 进阶 |
| 06_custom_mcp_router_example.py | 自定义 MCP 路由器配置 | ⭐⭐ 进阶 |
| 07_configure_http_timeout_example.py | 配置 HTTP 请求超时时间 | ⭐⭐ 进阶 |
| 08_auth_example_token_passthrough.py | 令牌透传认证示例 | ⭐⭐⭐ 高级 |
| 09_auth_example_auth0.py | Auth0 认证集成示例 | ⭐⭐⭐ 高级 |
核心概念
MCP 工具生成流程
fastapi_mcp 的核心功能是将 FastAPI 应用中的 API 端点自动转换为 MCP(Model Context Protocol)工具。转换过程中会保留原有的请求模型、响应模型和文档描述。
graph TD
A[FastAPI 应用] --> B[FastApiMCP 初始化]
B --> C[解析 OpenAPI Schema]
C --> D[提取端点信息]
D --> E[转换端点为 MCP 工具]
E --> F[MCP 服务器]基础使用示例
最简化的使用方式只需三行代码即可将 MCP 功能集成到现有的 FastAPI 应用中:
from fastapi_mcp import FastApiMCP
# 添加 MCP 服务器到 FastAPI app
mcp = FastApiMCP(app)
# 将 MCP 服务器挂载到 HTTP 传输层
mcp.mount_http()
资料来源:examples/01_basic_usage_example.py:1-13
这种零配置的设计理念使得开发者无需修改现有业务代码,即可快速获得 MCP 能力。
端点过滤配置
在实际应用中,可能只需要暴露部分 API 端点作为 MCP 工具。fastapi_mcp 提供了多种过滤机制来控制暴露的端点。
按 Operation ID 过滤
通过 include_operations 和 exclude_operations 参数可以指定包含或排除特定的操作 ID:
# 仅包含指定的操作 ID
include_mcp = FastApiMCP(
app,
include_operations=["get_item", "list_items"],
)
# 排除指定的操作 ID
exclude_mcp = FastApiMCP(
app,
exclude_operations=["create_item", "update_item", "delete_item"],
)
按 Tag 过滤
通过 include_tags 和 exclude_tags 参数可以按 OpenAPI 中的 tag 进行过滤:
# 仅包含 items tag 下的端点
include_tags_mcp = FastApiMCP(
app,
include_tags=["items"],
)
# 排除 search tag 下的端点
exclude_tags_mcp = FastApiMCP(
app,
exclude_tags=["search"],
)
过滤规则说明
| 规则 | 说明 |
|---|---|
| include_operations 与 exclude_operations | 互斥,不能同时使用 |
| include_tags 与 exclude_tags | 互斥,不能同时使用 |
| 混合使用 | 可以组合使用 operation 过滤和 tag 过滤,满足任一条件的端点都会被包含 |
资料来源:examples/03_custom_exposed_endpoints_example.py:1-55
认证配置
fastapi_mcp 支持基于 OAuth 2.0 的认证功能,可以利用现有的 FastAPI 依赖注入机制来保护 MCP 端点。
令牌透传认证
最简单的认证方式是通过 Authorization 头部透传令牌:
from fastapi import Depends
from fastapi.security import HTTPBearer
from fastapi_mcp import FastApiMCP, AuthConfig
token_auth_scheme = HTTPBearer()
@app.get("/private")
async def private(token=Depends(token_auth_scheme)):
return token.credentials
mcp = FastApiMCP(
app,
auth_config=AuthConfig(
dependencies=[Depends(token_auth_scheme)],
),
)
资料来源:examples/08_auth_example_token_passthrough.py:1-47
MCP 客户端配置
使用令牌透传时,需要在 MCP 客户端配置文件中设置头部:
{
"mcpServers": {
"remote-example": {
"command": "npx",
"args": [
"mcp-remote",
"http://localhost:8000/mcp",
"--header",
"Authorization:${AUTH_HEADER}"
]
},
"env": {
"AUTH_HEADER": "Bearer <your-token>"
}
}
}
Auth0 集成
对于需要集成第三方 OAuth 提供者的场景,fastapi_mcp 提供了完整的 Auth0 集成示例:
from fastapi_mcp import FastApiMCP, AuthConfig
mcp = FastApiMCP(
app,
auth_config=AuthConfig(
issuer="https://your-tenant.auth0.com",
# 其他 OAuth 配置...
),
)
资料来源:examples/09_auth_example_auth0.py
高级配置
独立 MCP 服务器
除了挂载到 FastAPI 应用外,fastapi_mcp 也支持作为独立服务器运行,适用于需要单独部署 MCP 服务的场景。
HTTP 超时配置
通过自定义 httpx.AsyncClient 实例,可以配置 HTTP 请求的超时时间:
import httpx
from fastapi_mcp import FastApiMCP
custom_client = httpx.AsyncClient(timeout=30.0)
mcp = FastApiMCP(
app,
http_client=custom_client,
)
资料来源:examples/07_configure_http_timeout_example.py
工具重新注册
对于需要动态更新 MCP 工具的场景,示例 05 演示了如何重新注册工具的方法。
自定义 MCP 路由器
示例 06 展示了如何使用自定义的 MCP 路由器来扩展功能。
共享应用组件
所有示例都依赖位于 examples/shared/apps/items.py 的共享 FastAPI 应用,这是一个包含 CRUD 操作的 Items RESTful API:
| 端点 | 方法 | 功能 |
|---|---|---|
| /items | GET | 获取所有 Items |
| /items/{item_id} | GET | 获取单个 Item |
| /items | POST | 创建 Item |
| /items/{item_id} | PUT | 更新 Item |
| /items/{item_id} | DELETE | 删除 Item |
| /items/search | GET | 搜索 Items |
该应用使用 Pydantic 模型定义请求和响应结构,确保数据验证的完整性。
运行示例
环境准备
示例代码需要以下环境条件:
- Python 3.10+(推荐 3.12)
- uv 包管理器
启动命令
每个示例都可以直接运行:
cd examples
uv run python 01_basic_usage_example.py
服务器启动后,可以通过访问 http://localhost:8000/docs 查看 Swagger UI,或访问 http://localhost:8000/mcp 查看 MCP 端点。
架构总览
graph LR
subgraph "FastAPI Application"
A[REST Endpoints] --> B[Pydantic Models]
A --> C[Dependencies]
end
subgraph "FastApiMCP"
D[OpenAPI Schema Parser] --> E[Tool Generator]
E --> F[MCP Server]
end
G[MCP Clients] --> F
F --> H[HTTP Transport]
H --> Afastapi_mcp 充当了 FastAPI 应用与 MCP 协议之间的桥梁,将原生 FastAPI 特性完整地保留到 MCP 工具定义中。
总结
示例集提供了从入门到高级的完整学习路径。开发者应当:
- 从
01_basic_usage_example.py开始了解基础集成方式 - 参考
03_custom_exposed_endpoints_example.py学习端点过滤 - 查看
08_auth_example_token_passthrough.py和09_auth_example_auth0.py了解认证功能 - 根据具体需求查阅其他高级示例
所有示例都遵循"最小配置"原则,开箱即用,同时提供充分的定制化能力满足生产环境需求。
资料来源:[examples/01_basic_usage_example.py:1-13]()
失败模式与踩坑日记
保留 Doramagic 在发现、验证和编译中沉淀的项目专属风险,不把社区讨论只当作装饰信息。
可能影响升级、迁移或版本选择。
可能增加新用户试用和生产接入成本。
可能影响升级、迁移或版本选择。
可能影响升级、迁移或版本选择。
Pitfall Log / 踩坑日志
项目:tadata-org/fastapi_mcp
摘要:发现 22 个潜在踩坑项,其中 1 个为 high/blocking;最高优先级:配置坑 - 来源证据:[BUG] MCP session 404 in multi worker production environment。
1. 配置坑 · 来源证据:[BUG] MCP session 404 in multi worker production environment
- 严重度:high
- 证据强度:source_linked
- 发现:GitHub 社区证据显示该项目存在一个配置相关的待验证问题:[BUG] MCP session 404 in multi worker production environment
- 对用户的影响:可能影响升级、迁移或版本选择。
- 建议检查:来源问题仍为 open,Pack Agent 需要复核是否仍影响当前版本。
- 防护动作:不得脱离来源链接放大为确定性结论;需要标注适用版本和复核状态。
- 证据:community_evidence:github | cevd_f318cbe8fc55407da8cb88f5418cce0d | https://github.com/tadata-org/fastapi_mcp/issues/189 | 来源讨论提到 python 相关条件,需在安装/试用前复核。
2. 安装坑 · 来源证据:v0.1.8
- 严重度:medium
- 证据强度:source_linked
- 发现:GitHub 社区证据显示该项目存在一个安装相关的待验证问题:v0.1.8
- 对用户的影响:可能增加新用户试用和生产接入成本。
- 建议检查:来源显示可能已有修复、规避或版本变化,说明书中必须标注适用版本。
- 防护动作:不得脱离来源链接放大为确定性结论;需要标注适用版本和复核状态。
- 证据:community_evidence:github | cevd_11a827f3808141e4bd7b0541a8628af0 | https://github.com/tadata-org/fastapi_mcp/releases/tag/v0.1.8 | 来源类型 github_release 暴露的待验证使用条件。
3. 安装坑 · 来源证据:v0.2.0
- 严重度:medium
- 证据强度:source_linked
- 发现:GitHub 社区证据显示该项目存在一个安装相关的待验证问题:v0.2.0
- 对用户的影响:可能影响升级、迁移或版本选择。
- 建议检查:来源显示可能已有修复、规避或版本变化,说明书中必须标注适用版本。
- 防护动作:不得脱离来源链接放大为确定性结论;需要标注适用版本和复核状态。
- 证据:community_evidence:github | cevd_a145fff6c53f4e709ef1bb7bc291216c | https://github.com/tadata-org/fastapi_mcp/releases/tag/v0.2.0 | 来源类型 github_release 暴露的待验证使用条件。
4. 安装坑 · 来源证据:v0.3.4
- 严重度:medium
- 证据强度:source_linked
- 发现:GitHub 社区证据显示该项目存在一个安装相关的待验证问题:v0.3.4
- 对用户的影响:可能影响升级、迁移或版本选择。
- 建议检查:来源显示可能已有修复、规避或版本变化,说明书中必须标注适用版本。
- 防护动作:不得脱离来源链接放大为确定性结论;需要标注适用版本和复核状态。
- 证据:community_evidence:github | cevd_6dcb58f1897f46a188514e2714e5896d | https://github.com/tadata-org/fastapi_mcp/releases/tag/v0.3.4 | 来源类型 github_release 暴露的待验证使用条件。
5. 配置坑 · 可能修改宿主 AI 配置
- 严重度:medium
- 证据强度:source_linked
- 发现:项目面向 Claude/Cursor/Codex/Gemini/OpenCode 等宿主,或安装命令涉及用户配置目录。
- 对用户的影响:安装可能改变本机 AI 工具行为,用户需要知道写入位置和回滚方法。
- 建议检查:列出会写入的配置文件、目录和卸载/回滚步骤。
- 防护动作:涉及宿主配置目录时必须给回滚路径,不能只给安装命令。
- 证据:capability.host_targets | github_repo:944976593 | https://github.com/tadata-org/fastapi_mcp | host_targets=mcp_host, claude, cursor
6. 配置坑 · 来源证据:Suggestion: MCPWatch observability example for fastapi_mcp users
- 严重度:medium
- 证据强度:source_linked
- 发现:GitHub 社区证据显示该项目存在一个配置相关的待验证问题:Suggestion: MCPWatch observability example for fastapi_mcp users
- 对用户的影响:可能增加新用户试用和生产接入成本。
- 建议检查:来源问题仍为 open,Pack Agent 需要复核是否仍影响当前版本。
- 防护动作:不得脱离来源链接放大为确定性结论;需要标注适用版本和复核状态。
- 证据:community_evidence:github | cevd_dfa72f41f3094dd5b2ffe188889f2b4f | https://github.com/tadata-org/fastapi_mcp/issues/303 | 来源类型 github_issue 暴露的待验证使用条件。
7. 配置坑 · 来源证据:clean_schema_for_display() strips anyOf and loses items for Optional[List[X]] parameters
- 严重度:medium
- 证据强度:source_linked
- 发现:GitHub 社区证据显示该项目存在一个配置相关的待验证问题:clean_schema_for_display() strips anyOf and loses items for Optional[List[X]] parameters
- 对用户的影响:可能增加新用户试用和生产接入成本。
- 建议检查:来源问题仍为 open,Pack Agent 需要复核是否仍影响当前版本。
- 防护动作:不得脱离来源链接放大为确定性结论;需要标注适用版本和复核状态。
- 证据:community_evidence:github | cevd_74e4280da33d49e1a3a8d576c7bb78a6 | https://github.com/tadata-org/fastapi_mcp/issues/304 | 来源讨论提到 python 相关条件,需在安装/试用前复核。
8. 配置坑 · 来源证据:v0.3.6
- 严重度:medium
- 证据强度:source_linked
- 发现:GitHub 社区证据显示该项目存在一个配置相关的待验证问题:v0.3.6
- 对用户的影响:可能增加新用户试用和生产接入成本。
- 建议检查:来源显示可能已有修复、规避或版本变化,说明书中必须标注适用版本。
- 防护动作:不得脱离来源链接放大为确定性结论;需要标注适用版本和复核状态。
- 证据:community_evidence:github | cevd_bdc90006d16a437798ff2766d514f3d4 | https://github.com/tadata-org/fastapi_mcp/releases/tag/v0.3.6 | 来源类型 github_release 暴露的待验证使用条件。
9. 能力坑 · 能力判断依赖假设
- 严重度:medium
- 证据强度:source_linked
- 发现:README/documentation is current enough for a first validation pass.
- 对用户的影响:假设不成立时,用户拿不到承诺的能力。
- 建议检查:将假设转成下游验证清单。
- 防护动作:假设必须转成验证项;没有验证结果前不能写成事实。
- 证据:capability.assumptions | github_repo:944976593 | https://github.com/tadata-org/fastapi_mcp | README/documentation is current enough for a first validation pass.
10. 维护坑 · 来源证据:[BUG] Description incorrectly passed as version to MCP Server
- 严重度:medium
- 证据强度:source_linked
- 发现:GitHub 社区证据显示该项目存在一个维护/版本相关的待验证问题:[BUG] Description incorrectly passed as version to MCP Server
- 对用户的影响:可能增加新用户试用和生产接入成本。
- 建议检查:来源问题仍为 open,Pack Agent 需要复核是否仍影响当前版本。
- 防护动作:不得脱离来源链接放大为确定性结论;需要标注适用版本和复核状态。
- 证据:community_evidence:github | cevd_2e599a8b03d649d8a47efb6b4d49f5ca | https://github.com/tadata-org/fastapi_mcp/issues/293 | 来源讨论提到 python 相关条件,需在安装/试用前复核。
11. 维护坑 · 来源证据:v0.3.0
- 严重度:medium
- 证据强度:source_linked
- 发现:GitHub 社区证据显示该项目存在一个维护/版本相关的待验证问题:v0.3.0
- 对用户的影响:可能影响升级、迁移或版本选择。
- 建议检查:来源显示可能已有修复、规避或版本变化,说明书中必须标注适用版本。
- 防护动作:不得脱离来源链接放大为确定性结论;需要标注适用版本和复核状态。
- 证据:community_evidence:github | cevd_bc6b7bd2988b48d48920e4ffb259f147 | https://github.com/tadata-org/fastapi_mcp/releases/tag/v0.3.0 | 来源类型 github_release 暴露的待验证使用条件。
12. 维护坑 · 来源证据:v0.3.3 - Fix OpenAPI Conversion Regression
- 严重度:medium
- 证据强度:source_linked
- 发现:GitHub 社区证据显示该项目存在一个维护/版本相关的待验证问题:v0.3.3 - Fix OpenAPI Conversion Regression
- 对用户的影响:可能增加新用户试用和生产接入成本。
- 建议检查:来源显示可能已有修复、规避或版本变化,说明书中必须标注适用版本。
- 防护动作:不得脱离来源链接放大为确定性结论;需要标注适用版本和复核状态。
- 证据:community_evidence:github | cevd_79b96b9d35b9444c938c355c081410ac | https://github.com/tadata-org/fastapi_mcp/releases/tag/v0.3.3 | 来源类型 github_release 暴露的待验证使用条件。
13. 维护坑 · 来源证据:v0.4.0
- 严重度:medium
- 证据强度:source_linked
- 发现:GitHub 社区证据显示该项目存在一个维护/版本相关的待验证问题:v0.4.0
- 对用户的影响:可能影响升级、迁移或版本选择。
- 建议检查:来源显示可能已有修复、规避或版本变化,说明书中必须标注适用版本。
- 防护动作:不得脱离来源链接放大为确定性结论;需要标注适用版本和复核状态。
- 证据:community_evidence:github | cevd_4382c9c951e14187b76777ad8561ded9 | https://github.com/tadata-org/fastapi_mcp/releases/tag/v0.4.0 | 来源类型 github_release 暴露的待验证使用条件。
14. 维护坑 · 维护活跃度未知
- 严重度:medium
- 证据强度:source_linked
- 发现:未记录 last_activity_observed。
- 对用户的影响:新项目、停更项目和活跃项目会被混在一起,推荐信任度下降。
- 建议检查:补 GitHub 最近 commit、release、issue/PR 响应信号。
- 防护动作:维护活跃度未知时,推荐强度不能标为高信任。
- 证据:evidence.maintainer_signals | github_repo:944976593 | https://github.com/tadata-org/fastapi_mcp | last_activity_observed missing
15. 安全/权限坑 · 下游验证发现风险项
- 严重度:medium
- 证据强度:source_linked
- 发现:no_demo
- 对用户的影响:下游已经要求复核,不能在页面中弱化。
- 建议检查:进入安全/权限治理复核队列。
- 防护动作:下游风险存在时必须保持 review/recommendation 降级。
- 证据:downstream_validation.risk_items | github_repo:944976593 | https://github.com/tadata-org/fastapi_mcp | no_demo; severity=medium
16. 安全/权限坑 · 存在安全注意事项
- 严重度:medium
- 证据强度:source_linked
- 发现:No sandbox install has been executed yet; downstream must verify before user use.
- 对用户的影响:用户安装前需要知道权限边界和敏感操作。
- 建议检查:转成明确权限清单和安全审查提示。
- 防护动作:安全注意事项必须面向用户前置展示。
- 证据:risks.safety_notes | github_repo:944976593 | https://github.com/tadata-org/fastapi_mcp | No sandbox install has been executed yet; downstream must verify before user use.
17. 安全/权限坑 · 存在评分风险
- 严重度:medium
- 证据强度:source_linked
- 发现:no_demo
- 对用户的影响:风险会影响是否适合普通用户安装。
- 建议检查:把风险写入边界卡,并确认是否需要人工复核。
- 防护动作:评分风险必须进入边界卡,不能只作为内部分数。
- 证据:risks.scoring_risks | github_repo:944976593 | https://github.com/tadata-org/fastapi_mcp | no_demo; severity=medium
18. 安全/权限坑 · 来源证据:v0.3.1 - Authorization
- 严重度:medium
- 证据强度:source_linked
- 发现:GitHub 社区证据显示该项目存在一个安全/权限相关的待验证问题:v0.3.1 - Authorization
- 对用户的影响:可能影响授权、密钥配置或安全边界。
- 建议检查:来源显示可能已有修复、规避或版本变化,说明书中必须标注适用版本。
- 防护动作:不得脱离来源链接放大为确定性结论;需要标注适用版本和复核状态。
- 证据:community_evidence:github | cevd_41bf79ee5bd04da1b943d22449a0d649 | https://github.com/tadata-org/fastapi_mcp/releases/tag/v0.3.1 | 来源类型 github_release 暴露的待验证使用条件。
19. 安全/权限坑 · 来源证据:v0.3.2
- 严重度:medium
- 证据强度:source_linked
- 发现:GitHub 社区证据显示该项目存在一个安全/权限相关的待验证问题:v0.3.2
- 对用户的影响:可能影响授权、密钥配置或安全边界。
- 建议检查:来源显示可能已有修复、规避或版本变化,说明书中必须标注适用版本。
- 防护动作:不得脱离来源链接放大为确定性结论;需要标注适用版本和复核状态。
- 证据:community_evidence:github | cevd_aa60828a6a6c4cc2b0b28fb72bf2ddad | https://github.com/tadata-org/fastapi_mcp/releases/tag/v0.3.2 | 来源类型 github_release 暴露的待验证使用条件。
20. 安全/权限坑 · 来源证据:v0.3.7
- 严重度:medium
- 证据强度:source_linked
- 发现:GitHub 社区证据显示该项目存在一个安全/权限相关的待验证问题:v0.3.7
- 对用户的影响:可能影响授权、密钥配置或安全边界。
- 建议检查:来源显示可能已有修复、规避或版本变化,说明书中必须标注适用版本。
- 防护动作:不得脱离来源链接放大为确定性结论;需要标注适用版本和复核状态。
- 证据:community_evidence:github | cevd_31faa7de6a364174958daaffa9d9204b | https://github.com/tadata-org/fastapi_mcp/releases/tag/v0.3.7 | 来源类型 github_release 暴露的待验证使用条件。
21. 维护坑 · issue/PR 响应质量未知
- 严重度:low
- 证据强度:source_linked
- 发现:issue_or_pr_quality=unknown。
- 对用户的影响:用户无法判断遇到问题后是否有人维护。
- 建议检查:抽样最近 issue/PR,判断是否长期无人处理。
- 防护动作:issue/PR 响应未知时,必须提示维护风险。
- 证据:evidence.maintainer_signals | github_repo:944976593 | https://github.com/tadata-org/fastapi_mcp | issue_or_pr_quality=unknown
22. 维护坑 · 发布节奏不明确
- 严重度:low
- 证据强度:source_linked
- 发现:release_recency=unknown。
- 对用户的影响:安装命令和文档可能落后于代码,用户踩坑概率升高。
- 建议检查:确认最近 release/tag 和 README 安装命令是否一致。
- 防护动作:发布节奏未知或过期时,安装说明必须标注可能漂移。
- 证据:evidence.maintainer_signals | github_repo:944976593 | https://github.com/tadata-org/fastapi_mcp | release_recency=unknown
来源:Doramagic 发现、验证与编译记录