API Reference

This page provides detailed API documentation for mcpgate.

Main Module

mcpgate

A stateless gateway that turns any OpenAPI spec into MCP tools on the fly.

Classes

OpenAPIMiddleware

OpenAPIMiddleware(*, ttl: Seconds = 300.0)

Bases: Middleware

Middleware that builds per-request MCP tools from an OpenAPI spec.

Headers extracted from each request

• x-openapi-url: URL of the OpenAPI JSON specification.
• x-api-url: Base URL of the target API.
• x-cookies (optional): Cookie string forwarded to the API.

The middleware uses ContextVar to isolate the per-request OpenAPIProvider, so concurrent requests never share state.

Providers and their HTTP clients are cached by (openapi_url, api_url) with a configurable TTL to avoid redundant spec fetches and parsing.

Initialize the middleware with a provider cache TTL in seconds.

Source code in src/mcpgate/__init__.py

def __init__(self, *, ttl: Seconds = 300.0) -> None:
    """Initialize the middleware with a provider cache TTL in seconds."""
    self._ttl = ttl
    self._cache: dict[CacheKey, _CachedProvider] = {}
    self._lock = asyncio.Lock()
    self._spec_client: AsyncClient | None = None

Functions

on_list_tools async

on_list_tools(context: MiddlewareContext[ListToolsRequest], call_next: CallNext[ListToolsRequest, Sequence[Tool]]) -> Sequence[Tool]

Prepend tools from the per-request OpenAPI provider.

Source code in src/mcpgate/__init__.py

async def on_list_tools(
    self,
    context: MiddlewareContext[mt.ListToolsRequest],
    call_next: CallNext[mt.ListToolsRequest, Sequence[Tool]],
) -> Sequence[Tool]:
    """Prepend tools from the per-request OpenAPI provider."""
    if provider := self._provider.get():
        return list(chain(await provider.list_tools(), await call_next(context)))
    return await call_next(context)

on_call_tool async

on_call_tool(context: MiddlewareContext[CallToolRequestParams], call_next: CallNext[CallToolRequestParams, ToolResult]) -> ToolResult

Intercept tool calls destined for the per-request OpenAPI provider.

Source code in src/mcpgate/__init__.py

async def on_call_tool(
    self,
    context: MiddlewareContext[mt.CallToolRequestParams],
    call_next: CallNext[mt.CallToolRequestParams, ToolResult],
) -> ToolResult:
    """Intercept tool calls destined for the per-request OpenAPI provider."""
    if provider := self._provider.get():
        tool = await provider.get_tool(context.message.name)
        if tool:
            return await tool.run(arguments=context.message.arguments or {})
    return await call_next(context)

close async

close() -> None

Close all cached clients and the spec-fetching client.

Source code in src/mcpgate/__init__.py

async def close(self) -> None:
    """Close all cached clients and the spec-fetching client."""
    for cached in self._cache.values():
        await cached.client.aclose()
    self._cache.clear()
    if self._spec_client is not None:
        await self._spec_client.aclose()
        self._spec_client = None

Functions

create_mcp

create_mcp() -> FastMCP

Create and return a new FastMCP instance with OpenAPI middleware.

Source code in src/mcpgate/__init__.py

def create_mcp() -> FastMCP:
    """Create and return a new FastMCP instance with OpenAPI middleware."""
    server = FastMCP()
    server.add_middleware(OpenAPIMiddleware())
    return server

run

run(port: int = 8000, log_level: Level = Level.INFO) -> None

Run the MCP gateway server.

Source code in src/mcpgate/__init__.py

@app.command()
def run(port: int = 8000, log_level: Level = Level.INFO) -> None:
    """Run the MCP gateway server."""
    setup_logging(level=log_level)
    mcp.run(transport="http", port=port)