fastapi-gql-mcp

The agent access layer for FastAPI — agents call one graphql_query tool with full composition and per-caller credentials, instead of one tool per endpoint. The GraphQL layer + MCP server are derived from your existing app: zero decorators, zero model changes.

from fastapi import FastAPI
from fastapi_gql_mcp import FastAPIMCP

app = FastAPI()

# ... your existing routes ...

mcp = FastAPIMCP(app, name="my-app")
mcp.run()  # HTTP MCP server with get_schema + graphql_query tools

Contents — Installation · Why · How it works · Capability boundaries · Usage · Authentication · Observability · Hardening · Demo · Development · Status

Installation

Requires Python >= 3.10.

uv add fastapi-gql-mcp            # core: GraphQL handler
uv add 'fastapi-gql-mcp[mcp]'     # + MCP server (fastmcp)

Why

Existing FastAPI→MCP bridges map one tool per endpoint: dozens of tools, no composition, whole-payload responses. fastapi-gql-mcp instead derives a GraphQL schema from your routes (Apollo's "GraphQL as the MCP contract" pattern), so agents get:

• a constant tool set (2 in simple mode, up to 6 with progressive disclosure) — never one tool per endpoint; discovery tools and graphql_query carry the readOnlyHint MCP annotation, graphql_mutation the destructiveHint, so agents can auto-run reads and confirm writes
• field-level selection — fetch { id name }, not the whole payload
• composition — combine several routes in one query; a failing route nulls only its own field
• your docs, verbatim — docstrings and description= metadata travel into the schema the agent reads
• real auth — queries run through the actual ASGI app, so Depends, middleware and headers apply; pass credentials via per-caller header passthrough
• nothing disappears — if a route works over HTTP, it stays callable here: untyped responses, serialization-filtered responses, unions and model fields whose type has no GraphQL mapping degrade to a documented raw-JSON field (with a startup notice naming the cause and the fix), and -> None routes become Boolean success fields — migration keeps its feel instead of routes silently vanishing
• recursive data comes back complete — selecting a recursive field means "the whole subtree": one level of children selection returns the tree at its true depth (the route already computed it; there is no invisible truncation), with your selection shape repeating per level (up to the unroll limit — see Rules worth knowing)

Where it wins, and where it doesn't: with many routes and long-lived agent sessions, the constant catalog dominates (at 100 routes: ~15,400 tok for a one-tool-per-endpoint OpenAPI bridge vs ~2,480 simple / ~1,510 progressive here — smaller at every measured size). Single-call latency is statistically tied with the flat bridge (1.39 vs 1.41 ms p50, same client stack), and the flat bridge needs no GraphQL mental model and works on any OpenAPI spec — but agent turns dominate, and composition collapses turns.

Compared to the alternatives

Full head-to-head — context-growth curves, latency, auth models, selection guidance, all measured on one shared app: comparison/.

How it works

No decorators, no model changes — everything is derived from the app you already have, in three steps:

1. Scan — RouterScanner reads app.routes: verb, path, params, response_model, tags, docstrings.
2. Build — a graphql-core schema is assembled: tags become domain groups, endpoint function names become field names, Pydantic models become GraphQL types, and your documentation becomes schema descriptions.
3. Execute — each field's resolver calls its route in-process through the real ASGI app, so Depends, middleware and auth behave exactly as over HTTP. Sibling fields resolve concurrently; a failing route nulls only its own field.

One route, end to end:

# your code — unchanged
@app.get("/products", response_model=list[ProductOut], tags=["shop:catalog"])
async def list_products(filters: Annotated[ProductFilter, Query()]) -> list[ProductOut]:
    """Browse the product catalog."""

# the schema the agent discovers (excerpt)
type Query { shop: ShopQuery! }
type ShopQuery { catalog: ShopCatalogQuery! }
type ShopCatalogQuery {
  list_products(category: String, in_stock: Boolean, limit: Int = 10): [ProductOut!]
}

# what the agent asks — field-level selection, routes combined in one query
{
  shop     { catalog { list_products(in_stock: true) { name } } }
  analytics { shop_stats { revenue_cents } }
}

The shape follows two rules:

• Tags group the tree — tags=["shop:catalog"] answers at { shop { catalog { … } } }. Untagged routes join the domain of their first path segment, so every field has a group.
• Function names name the leaves — async def list_products becomes list_products, your own vocabulary with no URL reconstruction. Two routes sharing a function name fail fast with DuplicateFieldError.

Capability boundaries

The bridging promise: if a route works over HTTP, it stays callable here. Every route lands in one of four buckets — classified once at scan time, so the SDL field note, the startup warning, and readiness() always tell the same story.

Structured — the default

Inputs — every argument shape FastAPI validates maps to a GraphQL argument:

Outputs — a typed response_model (or return annotation) becomes a selectable GraphQL type: { id name }, nested models, TypedDicts, stdlib @dataclasses, lists and sets, enums, Literals (enum members normalize to their values), custom scalars (UUID, Decimal, datetime…), generics (Page[Item]), constrained/bound TypeVars (they normalize to their constraint/bound), recursive models (true depth — see above), and output aliases (serialization_alias first, then alias).

Raw JSON fallback — still callable, just not field-selectable

response_model_exclude_none stays structured — it only drops keys that are nullable anyway. Every fallback names its cause in the field description and in a startup warning (bridged N route(s) as raw JSON …) that says how to get field selection back.

Boolean success — -> None routes

Deletes and other side-effect calls annotated -> None (204-style) become gone(id): Boolean: true on 2xx, failures surface as field errors.

Skipped — only when the route cannot be called correctly

Skips are logged at startup with the reason and the fix; handler.skips exposes them (path, method, reason) for CI assertions, and readiness() (below) wraps them into the fuller exposure audit.

include/exclude globs — and include_tags/exclude_tags — also remove routes by configuration; that is filtering you asked for, not a skip.

Readiness checklist

The same audit the startup notices come from is callable as data — which routes the bridge would skip, which it would degrade to raw JSON, and which model fields would degrade. One classifier backs both, so the report and the warnings can never drift apart. Pass the same filters your deployment uses, and assert report.ready in CI to pin the exposure you expect.

# standalone: scan + classify only — no schema build, no MCP server
from fastapi_gql_mcp import RouterScanner

report = RouterScanner(app, include_tags=["iam:*"]).readiness()
report.ready            # False when anything is skipped or degraded
report.skips            # tuple[SkipRecord(path, method, reason, tags), ...]
report.degraded         # tuple[DegradedRecord(path, method, field_name, reason, tags), ...]
report.degraded_fields  # tuple[(Model.field, reason), ...] — unions and unmappable field types

# or over an already-built deployment (stored scan results, no re-scan)
mcp.handler.readiness()

Usage

MCP server (HTTP)

run() serves streamable HTTP (the only transport — the wrapped app is a service, and per-caller credential passthrough needs an HTTP request context). Use mount_to(app, "/mcp") to serve MCP on the app's own port.

mcp = FastAPIMCP(
    app,
    name="my-app",
    allow_mutation=False,
    include=["/api/*"],
    include_tags=["iam:*"],  # keep only routes tagged iam:… (untagged drop)
    # The caller's own Authorization header travels to the routes by default;
    # an empty list disables forwarding entirely.
    # passthrough_headers=["authorization"],
)
mcp.run()  # streamable HTTP, 127.0.0.1:8000 — mcp.run(host="0.0.0.0", port=9000)

instructions: the agent's first read

instructions rides the initialize handshake — clients inject it into the conversation once per connection, before any tool call. A short guide saves the agent its discovery round-trips (and you the tokens):

mcp = FastAPIMCP(
    app,
    name="my-app",
    instructions=(
        "Notes service exposed as GraphQL. Domains: iam (users, roles), "
        "shop (catalog, orders). Call get_schema once, then compose "
        "several routes in a single graphql_query — a failing route nulls "
        "only its own field. Writes go through graphql_mutation."
    ),
)

Progressive disclosure (large apps)

Above progressive_threshold routes (default 25, mode="auto", counted after path and tag filtering), the toolset switches to a 5-layer walkthrough of the tag tree:

list_domains ──▶ list_queries("billing:invoice") ──▶ get_query_schema("billing:invoice") ──▶ graphql_query
                     (list_mutations with allow_mutation=True)
                     search_fields("billing", "invoice late fee")   ← BM25 shortcut for wide domains

Each domain SDL fragment re-wraps the real group types along the path, so it shows exactly the grouped query the agent must write — nothing more, and with every description attached. Discovery is scoped; execution is not: graphql_query always runs against the full schema, so fields from different domains combine freely. Force either mode with mode="simple" | "progressive".

Mounted into the same app

mcp.mount_to(app, "/mcp")            # streamable HTTP at /mcp/
mcp.handler.mount_graphql(app)       # GraphiQL at /graphiql + POST /graphql

mcp.mount_to(app, "/mcp", stateless_http=True)  # no session affinity needed
mcp.run(stateless_http=True)                    # same flag on run()

Multiple MCP deployments over one app

Different MCP consumers often need different slices of the same app. Build one FastAPIMCP per use case, each scoped by its own tag filter, and mount each at its own path — the instances share nothing but the wrapped app:

iam = FastAPIMCP(app, name="iam-api", include_tags=["iam:*"])
billing = FastAPIMCP(app, name="billing-api", include_tags=["billing:*"])
iam.mount_to(app, "/mcp-iam")        # streamable HTTP at /mcp-iam/
billing.mount_to(app, "/mcp-billing")

Plain GraphQL (no MCP)

from fastapi_gql_mcp import RouterGraphQLHandler

handler = RouterGraphQLHandler(app)
print(handler.get_sdl())
result = await handler.execute(
    "query($id: Int!) { iam { get_user(user_id: $id) { name } } }",
    variables={"id": 1},
)

Authentication

Route calls travel through the real ASGI app in-process, so Depends, middleware and security schemes behave exactly as over HTTP. Credentials have a single source: the caller — the FastAPI security schemes are the only verifiers, and this bridge never holds or manages tokens of its own.

• Per-caller passthrough (default): each MCP/GraphQL client connects with its own credentials and passthrough_headers (default ("authorization",)) forwards them to the routes — queries run as the caller, exactly as they would over HTTP. An explicitly empty list disables forwarding; headers are matched case-insensitively and only whitelisted names ever reach a route (no smuggling x-internal-token past the bridge).
• Session cookies ride a header — whitelist "cookie" (the common trip-up): on the wire a browser session is the Cookie: request header, so cookie-authenticated apps forward it with passthrough_headers=["authorization", "cookie"]. The default forwards authorization only — with a session-cookie app, protected routes answer HTTP_401 over MCP until cookie is listed. FastAPI's Cookie() route parameters are the other, unrelated cookie: like Header() parameters they never become GraphQL arguments. The POST /graphql face (and GraphiQL) share the same whitelist, so browser sessions flow through it the same way. Wired example: examples/notes_oauth.

Client side, the credential is just a header on every MCP request:

# Claude Code
claude mcp add --transport http my-api http://localhost:8000/mcp \
  --header "authorization: Bearer <token>"
# generic clients (mcp-remote bridge)
npx mcp-remote http://localhost:8000/mcp --header "Authorization:${AUTH_HEADER}"

Gating the MCP endpoint

auth= also accepts a bare token verifier — subclass fastmcp.server.auth.auth.TokenVerifier, override verify_token, and the endpoint answers 401 (a WWW-Authenticate: Bearer challenge) for every request without a valid token:

from fastmcp.server.auth.auth import TokenVerifier
from mcp.server.auth.provider import AccessToken

class SharedSecretVerifier(TokenVerifier):
    """One shared token guards the MCP endpoint — no OAuth machinery."""

    async def verify_token(self, token: str):
        if token != "demo-secret":
            return None  # None -> 401
        return AccessToken(token=token, client_id="local", scopes=[], expires_at=None)

mcp = FastAPIMCP(app, auth=SharedSecretVerifier())

One token, two checks: the gate verifies the endpoint, and the routes' own security schemes verify business calls — the same Bearer reaches them via passthrough_headers. A perimeter you already control (network ACL, gateway auth on the /mcp path) works too. Combine with allow_mutation=False / mutation_include to keep writes out of reach.

MCP endpoint OAuth

Pass a fastmcp auth provider and the MCP endpoint speaks OAuth 2.1: 401 discovery, dynamic client registration, PKCE, a consent page, and its own reference tokens verifying every call:

mcp = FastAPIMCP(app, auth=GitHubProvider(client_id=..., client_secret=..., base_url=...))

Ready-made providers: GitHub, Google, Auth0, Keycloak, AWS, Azure, Clerk, Discord, Supabase, WorkOS, and more — plus JWTVerifier (verify an external IdP's JWTs via JWKS) and OAuthProxy (adapt any upstream OAuth server). Full list and configuration: fastmcp auth docs.

With a full provider, no --header is needed: Claude Code follows the 401 OAuth discovery, opens a browser, the user logs in, and the agent's queries run as that user. mount_to(app, "/mcp", auth_at_root=True) hosts the OAuth routes at the app root (for reusing an IdP app whose registered callback lives there). The bridge itself still verifies nothing. Full wired flow: examples/notes_oauth.

Observability (OpenTelemetry)

Install an OpenTelemetry SDK next to your app — that's the whole setup. One waterfall per query: fastmcp emits the tool span, the bridge emits graphql.execute and injects W3C traceparent into every in-process route call, and FastAPI >= 0.142 nests its route spans under it. Timeouts and queue waits surface as span events.

Hardening the bridge

Four knobs are built in and on by default:

• request_timeout (default 30s, None disables) — per-route-call deadline. The in-process ASGI call bypasses httpx's own timeout machinery, so enforcement lives in asyncio.wait_for; a timed-out field surfaces as a TIMEOUT error (http_status 504) while its siblings survive.
• max_depth (default 10, None disables) — maximum selection-set nesting per document. Recursive models make depth unbounded and an MCP caller is an LLM that can emit runaway nesting; overly deep documents are rejected with a validation-style error before anything executes.
• max_concurrency (default 16, None disables) — bound on in-flight route calls across all queries. Sibling fields resolve concurrently, so one wide query fans out; this protects the wrapped app's upstream from being hammered by its own bridge (queueing counts against request_timeout, default 30s).
• document_cache_size (default 128, 0 disables) — LRU capacity for the parse + depth-guard + validate front half of execution, keyed by the query string. Agents repeat documents constantly; a hit skips straight to execution (measured 1.52ms → 0.69ms on a 2-field query). Execution results are never cached — per-call credentials run for real every time.

All four are parameters of RouterGraphQLHandler and FastAPIMCP. For anything policy-shaped, validation_rules= on the handler passes extra graphql-core validation rules through (they extend the standard set).

from fastmcp.server.middleware.rate_limiting import RateLimitingMiddleware
from fastmcp.server.middleware.response_limiting import ResponseLimitingMiddleware

mcp = FastAPIMCP(app)
mcp.mcp.add_middleware(RateLimitingMiddleware(max_requests_per_second=10))
mcp.mcp.add_middleware(ResponseLimitingMiddleware(max_size=1_000_000))

Demo

examples/shop — a small shop app (users / catalog / orders / stats, auth via x-token: demo-secret) with every feature in play:

uv run --extra mcp python -m examples.shop                     # REST + /mcp/ + /graphiql + /graphql on :8010
uv run --extra mcp python -m examples.shop.mcp_walkthrough     # agent's-eye MCP walkthrough, no client needed

Development

uv sync && uv run pytest        # tests
uv run ruff check src tests
uv run mypy src

Status

0.x — breaking changes can land in minor bumps; 1.0 will freeze the public API. See CHANGELOG.md. Ideas welcome, agent-first: a get_schema variant that returns only the subgraph an agent asks about, response shaping for token budgets, per-team scoped deployments (one MCP deployment per tag set), response header pass-through.

Design extracted from nexusx (SQLModel → GraphQL → MCP), rebuilt on graphql-core standard execution.

License

MIT