使用说明

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许可证-有关详细信息，请参阅许可证文件

鸣谢

内置：

快速API -现代、快速的web框架
主控程序 -模型上下文协议
httpx -HTTP客户端
派丹蒂克 -数据验证 Python库，使用FastAPI（或任何与OpenAPI兼容的）应用程序作为事实来源，从REST端点动态生成MCP服务器和客户端。