fastapi-mcp网关
零样板库，将REST API和GraphQL API作为MCP（模型上下文协议）服务器公开。
概述
fastapi-mcp-gateway 提供FastAPI、OpenAPI 3.x REST API和GraphQL API到MCP服务器的运行时转换，允许AI代理轻松地与您的API交互。无需代码生成，无需额外的构建步骤，只需纯粹的运行时魔法。
特性
- 🚀 零沸点板：所需设置最少
- 🔄 运行时转换：不需要代码生成步骤
- 🎯 FastAPI优先：针对FastAPI进行了优化，但可与任何OpenAPI 3.x API配合使用
- 🔷 GraphQL支持：完全支持GraphQL查询和突变
- 🔌 多种模式：正在执行HTTP或GraphQL
- 🛠️ MCP客户端：用于程序化工具调用的内置客户端
- ⚙️ 可配置的：对暴露的端点进行细粒度控制
- 🧪 测试良好：综合单元和集成测试
安装
pip install fastapi-mcp-gateway对于FastAPI支持：
pip install "fastapi-mcp-gateway[fastapi]"对于GraphQL支持：
pip install "fastapi-mcp-gateway[graphql]"发展：
pip install "fastapi-mcp-gateway[dev]"快速开始
基本FastAPI示例
from fastapi import FastAPI
from fastapi_mcp_gateway import create_mcp_server_from_fastapi
# Create your FastAPI app
app = FastAPI()
@app.get("/users")
def get_users(page: int = 1, limit: int = 10):
return [
{"id": 1, "name": "Alice"},
{"id": 2, "name": "Bob"}
]
@app.get("/users/{user_id}")
def get_user(user_id: int):
return {"id": user_id, "name": f"User {user_id}"}
@app.post("/users")
def create_user(name: str, email: str):
return {"id": 3, "name": name, "email": email}
# Create and run MCP server
if __name__ == "__main__":
import asyncio
mcp_server = create_mcp_server_from_fastapi(app)
asyncio.run(mcp_server.run())使用外部OpenAPI URL
import asyncio
from fastapi_mcp_gateway import create_mcp_server_from_openapi
from fastapi_mcp_gateway.config import GatewayConfig
async def main():
# Create MCP server from external API
mcp_server = await create_mcp_server_from_openapi(
"http://localhost:8000/openapi.json",
config=GatewayConfig(
base_url="http://localhost:8000"
)
)
await mcp_server.run()
if __name__ == "__main__":
asyncio.run(main())使用MCP客户端
import asyncio
from fastapi_mcp_gateway import MCPClient
async def main():
# Connect to an MCP server
client = MCPClient()
await client.connect(
command="python",
args=["server.py"]
)
# List available tools
tools = await client.list_tools()
print(f"Available tools: {[t['name'] for t in tools]}")
# Call a tool
result = await client.call_tool(
"get_users",
{"page": 1, "limit": 10}
)
print(f"Result: {result}")
# Cleanup
await client.disconnect()
if __name__ == "__main__":
asyncio.run(main())使用GraphQL API
import asyncio
from fastapi_mcp_gateway import create_mcp_server_from_graphql
from fastapi_mcp_gateway.config import GatewayConfig, GraphQLOperationFilter
async def main():
# Create MCP server from GraphQL endpoint
config = GatewayConfig(
# Filter operations: ALL, QUERIES_ONLY, or MUTATIONS_ONLY
graphql_operation_filter=GraphQLOperationFilter.ALL,
)
mcp_server = await create_mcp_server_from_graphql(
"http://localhost:8000/graphql",
config=config
)
await mcp_server.run()
if __name__ == "__main__":
asyncio.run(main())配置
这 GatewayConfig 类提供了广泛的配置选项：
REST API配置
from fastapi_mcp_gateway import create_mcp_server_from_fastapi, GatewayConfig
from fastapi_mcp_gateway.config import ExecutionMode
import logging
config = GatewayConfig(
# Execution mode: IN_PROCESS (direct calls), HTTP (via requests), or GRAPHQL
mode=ExecutionMode.IN_PROCESS,
# Base URL for HTTP/GraphQL mode
base_url="http://localhost:8000",
# Include only specific paths (None = all)
include_paths=["/api/*", "/public"],
# Exclude specific paths
exclude_paths=["/admin/*", "/internal"],
# Custom authentication
auth_factory=lambda: {"Authorization": "Bearer token123"},
# Logging level
log_level=logging.DEBUG,
# HTTP timeout in seconds
timeout=30.0,
# Maximum retries for HTTP requests
max_retries=3
)
mcp_server = create_mcp_server_from_fastapi(app, config=config)GraphQL配置
from fastapi_mcp_gateway import create_mcp_server_from_graphql, GatewayConfig
from fastapi_mcp_gateway.config import GraphQLOperationFilter
config = GatewayConfig(
# Filter which operations to expose
graphql_operation_filter=GraphQLOperationFilter.ALL, # or QUERIES_ONLY, MUTATIONS_ONLY
# Include/exclude specific operations by name
include_operations=["user", "users", "createUser"],
exclude_operations=["deleteUser"],
# GraphQL endpoint path (default: "/graphql")
graphql_endpoint="/graphql",
)
mcp_server = await create_mcp_server_from_graphql(
"http://localhost:8000/graphql",
config=config
)运作原理
REST API
- 架构加载：库加载您的OpenAPI模式（从FastAPI应用程序或URL）
- 工具映射：每个API端点都转换为MCP工具，具有：
- 工具名称（来源 operationId 或方法+路径） - 描述（来自 summary 和 description) - 输入模式（来自路径/查询/正文参数）
- 执行：当调用工具时，库：
- 解析路径参数 - 将查询参数和请求正文分开 - 进行HTTP调用或直接调用FastAPI - 返回JSON响应
GraphQL API
- 架构加载：库会反思你的GraphQL模式
- 工具映射：每个查询/变异都转换为MCP工具，该工具具有：
- 工具名称（来自操作名称） - 描述（来自字段描述） - 输入模式（来自GraphQL参数）
- 执行：当调用工具时，库：
- 使用变量构建GraphQL查询/变异 - 向GraphQL端点发送HTTP POST - 解析响应并处理错误 - 返回JSON数据
路径筛选
使用包含/排除模式控制暴露的端点：
config = GatewayConfig(
# Only expose /api/* endpoints
include_paths=["/api/*"],
# Exclude admin endpoints
exclude_paths=["/api/admin/*"]
)模式匹配支持：
- 精确匹配：
/users - 前缀通配符：
/api/* - 后缀通配符：
*/admin
执行模式
进程模式（默认）
使用直接调用您的FastAPI应用程序 TestClient:
config = GatewayConfig(mode=ExecutionMode.IN_PROCESS)
mcp_server = create_mcp_server_from_fastapi(app, config=config)优势：更快，无网络开销，更容易调试\ 使用时间：在FastAPI应用程序旁边运行MCP服务器
HTTP模式
向正在运行的API发出实际HTTP请求：
config = GatewayConfig(
mode=ExecutionMode.HTTP,
base_url="http://localhost:8000"
)
mcp_server = await create_mcp_server_from_openapi(
"http://localhost:8000/openapi.json",
config=config
)优势：适用于任何HTTP API，支持远程服务器\ 使用时间：API已经部署，或者您需要实际的HTTP行为
GraphQL模式
通过HTTP POST执行GraphQL操作：
config = GatewayConfig(
mode=ExecutionMode.GRAPHQL,
base_url="http://localhost:8000/graphql"
)
mcp_server = await create_mcp_server_from_graphql(
"http://localhost:8000/graphql",
config=config
)优势：原生GraphQL支持、自省、类型安全\ 使用时间：使用GraphQL API，需要查询和突变作为工具
认证
使用添加身份验证标头 auth_factory:
def get_auth_headers():
return {
"Authorization": f"Bearer {get_current_token()}",
"X-API-Key": "secret-key"
}
config = GatewayConfig(auth_factory=get_auth_headers)发展
设置
git clone https://github.com/yourusername/fastapi-mcp-gateway.git
cd fastapi-mcp-gateway
pip install -e ".[dev]"运行测试
# Run all tests
pytest
# Run with coverage
pytest --cov=fastapi_mcp_gateway --cov-report=html
# Run only unit tests
pytest tests/unit/
# Run only integration tests
pytest tests/integration/代码质量
# Format code
black .
# Lint
ruff check .
# Type checking
mypy fastapi_mcp_gateway项目结构
fastapi-mcp-gateway/
├── fastapi_mcp_gateway/
│ ├── __init__.py # Public API
│ ├── config.py # Configuration classes
│ ├── openapi_loader.py # OpenAPI schema loading
│ ├── graphql_loader.py # GraphQL schema loading
│ ├── mapping.py # OpenAPI to MCP mapping
│ ├── graphql_mapping.py # GraphQL to MCP mapping
│ ├── execution.py # REST tool execution logic
│ ├── graphql_execution.py # GraphQL tool execution
│ ├── mcp_server.py # MCP server implementation
│ └── mcp_client.py # MCP client wrapper
├── tests/
│ ├── unit/ # Unit tests
│ └── integration/ # Integration tests
├── examples/
│ ├── simple_server.py # FastAPI REST example
│ ├── graphql_server.py # GraphQL server example
│ ├── graphql_mcp_server.py # GraphQL MCP server
│ └── graphql_client.py # GraphQL MCP client
├── spec/
│ └── design.md # Design specification
├── pyproject.toml # Package configuration
└── README.md # This file限制（v1）
- 仅JSON内容类型（无多部分/表单数据等）
- 不支持WebSocket或SSE
- 无类型化客户端生成
- 仅支持基本身份验证
- 不支持GraphQL订阅
- 不支持GraphQL片段和指令
路线图
- \[\]支持WebSocket和SSE
- \[\]GraphQL订阅（通过WebSocket）
- \[\]GraphQL片段和指令
- \[\]GraphQL联邦支持
- \[\]语义工具描述增强
- \[\]类型化客户端生成（REST+GraphQL）
- \[\]CLI启动器，便于部署
- \[\]支持更多内容类型
- \[\]高级身份验证流程（OAuth2等）
- \[\]混合REST+GraphQL端点
- \[\]GraphQL查询优化和批处理
贡献
欢迎投稿！拜托：
- 复刻仓库
- 创建要素分支
- 添加新功能的测试
- 确保所有测试通过
- 提交拉取请求
许可证
MIT许可证-有关详细信息，请参阅许可证文件
鸣谢
内置：
Python库，使用FastAPI（或任何与OpenAPI兼容的）应用程序作为事实来源，从REST端点动态生成MCP服务器和客户端。