fastapi-gql-mcp

Turn any FastAPI router into a GraphQL query layer + MCP server — 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

FastAPIMCP was named RouterMCP through 0.8. The old import still works as a deprecated alias (removed at 1.0) — switch to FastAPIMCP.

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

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

Compared to the alternatives

Project	Tool count	Field selection	Composition	Setup
fastapi-mcp (Tadata)	one per endpoint	✗	✗	none
FastMCP.from_openapi	one per endpoint	✗	✗	none
fastapi-gql-mcp	2-6, constant	✓	✓	none

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

What the comparison measures (numbers below are real, from comparison/bench/results.json)

One app (bench/shared_app.py, a notes CRUD API), wired into both bridges, driven from two venvs (they can't share one — fastmcp 4 needs mcp>=2, fastapi-mcp 0.4.0 breaks on mcp 2.x):

Measurement	fastapi-mcp	fastapi-gql-mcp
tool catalog, 100 routes	~10,950 tok (grows linearly)	~2,480 tok simple / ~1,480 tok progressive (constant)
composed task (notes+stats)	2 tool calls = 2 agent turns	1 graphql_query
same list response	2,875 B whole payload	610 B with field projection
single trivial call (p50, same client stack, 3 runs)	0.93 ms	1.25 ms — GraphQL layer costs; agent turns dominate, not ms

Honest counterexamples included: below ~5 endpoints the one-tool-per-endpoint catalog is actually smaller (685 vs 893 tok), and per-call latency favors them — the GraphQL route pays off as the API grows. Every number is reproducible (comparison/README.md → Reproduce); environment, versions and run counts are recorded in results.json.

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.

Rules worth knowing:

• Mutations are off by default (allow_mutation=True to expose writes; mutation_include=[...] globs to whitelist specific write routes); graphql_query also refuses mutation documents.
• Mutation ordering: per the GraphQL spec, mutation fields at the ROOT execute serially in declaration order — with the grouped schema that means writes in DIFFERENT domains are ordered; writes grouped under the SAME domain run in parallel like query fields. When write order matters, put the operations in separate domains or send separate mutation documents.
• Recursive models return their true depth. Selecting a recursive field means "the whole subtree": the selection shape where you stop repeats to whatever depth the data has — the route already computed the full tree, and the bridge hands it over complete instead of truncating at the document's depth (truncation there was invisible: children: [] on a leaf is indistinguishable from a cut-off subtree). max_depth still guards the document you write; per-level field filtering applies at every depth. The recursive type's schema description states this contract in one line per type, so agents discover it from the SDL itself.
• Dynamic shapes pass through as JSON — dict/Any annotations bridge as the JSON scalar in both directions (a JSON argument lands as the raw request body); untyped routes, serialization-filtered responses, unions and unmappable model fields degrade the same way, each with a field note and a startup notice naming the fix — see Capability boundaries. Only routes that cannot be called correctly at all are skipped, with a logged reason; handler.skips lists them programmatically, and readiness() returns the full skip + degradation audit for CI assertions.
• include/exclude fnmatch globs scope which routes enter the schema. include_tags/exclude_tags do the same over route tags: a route matches when ANY of its string tags matches ANY pattern (include_tags=["iam:*"] keeps tags=["iam:users"]), exclude_tags wins, untagged routes drop under a tag whitelist, and tag filters AND with path filters. exclude_deprecated=True drops deprecated=True routes the same way; routes that stay carry their GraphQL-native @deprecated mark.
• Route tags form a domain tree (tags=["billing:invoice"]); large apps switch to progressive disclosure (below).
• A lone Annotated[FilterModel, Query()] flattens into individual query arguments (FastAPI Query Parameter Models).
• Same-named Pydantic classes from different modules get qualified type names.
• Descriptions flow into the schema — model docstrings → type descriptions, Field(description=...) → field descriptions, endpoint docstrings (or summary=) → field descriptions, Query()/Path()/Body(description=...) → argument descriptions. They surface in GraphiQL hover, introspection and every MCP discovery tool.

Capability boundaries

The bridging promise: if a route works over HTTP, it stays callable here. Every route lands in one of four buckets.

Structured — the default

A typed response_model (or return annotation) over Pydantic models 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]), aliases.

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

Your route	What happens
returns dict / Any	author-declared dynamic shape → the JSON scalar, no sub-selection
no return annotation, no response_model	same, plus the field description and a startup notice tell you to add one
response_model_exclude_unset / _exclude_defaults / _include / _exclude / by_alias=False	filtering runs after validation, so per-field promises cannot hold; the notice names the kwarg
returns a union (Item | Error)	the union bridges as JSON; nested inside a model, only that field degrades
a model field no GraphQL type can express (metadata: SomeCustomClass)	only that field bridges as JSON — the model keeps its structured fields; readiness().degraded_fields names it
takes a union parameter (value: A | B)	the argument bridges as JSON; FastAPI validates whichever member arrives (a 422 surfaces as a field error)

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. response_model_exclude_none stays structured — it only drops keys that are nullable anyway.

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.

Condition	Why / what to do
Form() / File() body	MCP tool arguments are JSON; the protocol has no file channel yet (SEP-2631 draft). The route stays available over plain HTTP
required header/cookie parameter	headers are not GraphQL arguments — make it optional; caller credentials ride passthrough_headers
query-parameter model mixed with plain params	FastAPI itself cannot serve that shape on the wire; move the plain params into the model
write verbs with allow_mutation=False (the default)	opt in with allow_mutation=True or mutation_include
hidden route (include_in_schema=False)	opt in with include_hidden=True
returns a raw Response (streaming, plain text)	no typed body to expose
response/parameter annotation with no mapping at all (-> bytes, a bare custom class)	no per-field shape exists to degrade to — annotate with a JSON-compatible type

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.

# 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()

Pass the same filters your deployment uses, and assert report.ready in CI to pin the exposure you expect.

Installation

Requires Python >= 3.10.

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

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."
    ),
)

Writing it well:

• Lead with the domain map — the same names your tags produce, so the agent can aim graphql_query without reading the whole SDL first.
• Say the composition rule — one query can combine routes; that is the feature agents most often fail to discover on their own.
• Name the write channel — graphql_mutation (if enabled).
• Keep it under ~120 tokens — it is resident context for the whole session; the SDL (via get_schema) already carries the details.
• Progressive mode profits most: a domain map in instructions lets the agent skip list_domains and go straight to list_queries("iam").

Progressive disclosure (large apps)

Above progressive_threshold routes (default 25, mode="auto", counted after path and tag filtering), the toolset switches to a 4-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)

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

Multi-worker deployments: streamable HTTP sessions are stateful and live in one process — behind several workers or pods, either pin the MCP path to one worker / use sticky routing, or mount stateless:

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

Stateless mode runs one transport per request: it survives any load balancer, at the cost of per-request session setup. /graphql (plain GraphQL face) is stateless already.

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.
• Without credentials, protected routes fail — field errors like HTTP_401 in query results; nothing falls back to a server-side identity.
• Machines without a user context configure the service credential on the MCP client side (or, for programmatic use, pass handler.execute(..., headers={...}) directly).
• MCP endpoint OAuth (optional): pass a fastmcp auth provider — auth=GitHubProvider(client_id=..., client_secret=..., base_url=...) — 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. Claude Code 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.

Expose the MCP endpoint only behind an entrance you control (network, or a FastAPI Depends on the mounted route) — the bridge authenticates no one itself, and combine with allow_mutation=False / mutation_include to keep writes out of reach.

Observability (OpenTelemetry)

Install an OpenTelemetry SDK next to your app — that's the whole setup. The spans are emitted natively from both ends, and the bridge stitches them into one waterfall:

• fastmcp emits the tool level (tools/call graphql_query);
• the bridge emits graphql.execute (the GraphQL orchestration layer) and injects W3C traceparent into every in-process route call — independent of passthrough_headers, a no-op without an SDK (opentelemetry-api only, non-recording by default);
• FastAPI >= 0.142 emits the route level (GET /things plus fastapi.dependencies/endpoint/serialization) and extracts the injected context — so route spans nest under graphql.execute, one trace per query.

Route-call timeouts and concurrency queue waits surface as span events (route.timeout, route.queue) on graphql.execute. A runnable proof (plus the Jaeger walkthrough): examples/otel_smoke.md; a live wired app: examples/notes_oauth (env-gated app/observability.py). Metrics (per-URL QPS/p99) are out of scope here — derive them from spans with an OTel Collector spanmetrics connector.

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).

For rate limiting and response caps on the MCP face, FastMCP's middleware suite attaches with zero bridge code — FastAPIMCP.mcp is the underlying FastMCP instance:

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))

RateLimitingMiddleware limits per client by default (pass get_client_id= to customize the key or global_limit=True for a shared bucket); ResponseLimitingMiddleware truncates oversized tool responses (default 1 MB, configurable suffix). The POST /graphql face does not go through fastmcp — attach your own middleware to the host app for that endpoint.

Demo

examples/shop runs a small shop app (users / catalog / orders / stats, auth via x-token: demo-secret) with every feature in play — including full documentation coverage so all four description chains are inspectable in GraphiQL:

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

python -m examples.shop prints all endpoint URLs and serves the grouped schema; /now is untyped on purpose so the skip warning is visible at startup. GET /categories is the recursive showcase — CategoryOut.children is self-referencing, so over MCP one level of children selection returns the whole tree (the type's schema description states the contract).

For the full consumer experience — a real app with GitHub OAuth login, session cookies, and MCP OAuth (Claude Code's browser login flow) — see examples/notes_oauth: three interchangeable credential carriers resolved in one place, the MCP endpoint protected by an OAuth 2.1 proxy, and a smoke script that walks the protected paths headlessly. For observability, examples/otel_smoke.md walks the one-waterfall-per-query proof in Jaeger.

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: GraphQL subscriptions over SSE routes, response header pass-through, per-domain auth scopes.

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

License

MIT