MCP Tools for Strawberry GraphQL — async context, per-request DataLoader, permission classes

Async resolvers and the event loop constraint

Strawberry integrates with ASGI servers (FastAPI, Starlette, Django with ASGI). In an ASGI app, all requests share a single asyncio event loop. Synchronous functions run in a thread pool executor — CPython's default is min(32, os.cpu_count() + 4) threads. When a sync resolver calls a blocking I/O function (sqlite3, requests, blocking file read), it holds a thread pool thread until I/O completes. Under concurrent requests, thread pool exhaustion causes new sync resolvers to queue, adding latency proportional to parallelism.

import strawberry
from strawberry.fastapi import GraphQLRouter
from typing import Optional
import asyncpg  # asyncpg: async Postgres driver — no thread pool needed

# WRONG: sync resolver with blocking I/O
@strawberry.type
class QueryBad:
    @strawberry.field
    def user(self, id: str) -> Optional["UserType"]:
        import psycopg2  # blocking driver
        conn = psycopg2.connect(DATABASE_URL)  # blocks thread pool thread
        with conn.cursor() as cur:
            cur.execute("SELECT id, name FROM users WHERE id = %s", (id,))
            row = cur.fetchone()
        return UserType(id=row[0], name=row[1]) if row else None

# RIGHT: async resolver with async driver
@strawberry.type
class Query:
    @strawberry.field
    async def user(self, id: str, info: strawberry.types.Info) -> Optional["UserType"]:
        # asyncpg: async queries don't block the event loop
        pool: asyncpg.Pool = info.context.db_pool
        row = await pool.fetchrow(
            "SELECT id, name, email FROM users WHERE id = $1", id
        )
        return UserType(id=row["id"], name=row["name"]) if row else None

@strawberry.type
class UserType:
    id: str
    name: str

schema = strawberry.Schema(query=Query)

# Startup: create asyncpg pool at startup, share via app state
from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.db_pool = await asyncpg.create_pool(DATABASE_URL, min_size=5, max_size=20)
    yield
    await app.state.db_pool.close()

app = FastAPI(lifespan=lifespan)

async def get_context(request: Request) -> dict:
    return {"db_pool": request.app.state.db_pool, "viewer_id": extract_viewer_id(request)}

graphql_router = GraphQLRouter(schema, context_getter=get_context)
app.include_router(graphql_router, prefix="/graphql")

If you must use a synchronous library (e.g., an SDK that has no async variant), run it explicitly in the thread pool with asyncio.get_event_loop().run_in_executor(None, sync_fn, *args) inside an async resolver. This is the same thing Strawberry does for sync resolvers — you make it explicit and avoid the hidden per-resolver overhead of Strawberry's wrapper.

Per-request DataLoader isolation

Strawberry's strawberry.dataloader is a Python port of the original JavaScript DataLoader. It batches load(key) calls within the same asyncio event loop tick and caches results by key. The caching is intentional for within-request deduplication, but becomes a security issue across requests if the DataLoader instance is shared.

import strawberry
from strawberry.dataloader import DataLoader
from typing import List, Optional
from fastapi import Request

# WRONG: module-level DataLoader — cache persists across all requests
shared_user_loader = DataLoader(load_fn=lambda keys: load_users_from_db(keys))

# RIGHT: DataLoader created per-request inside context_getter
async def load_users(keys: List[str]) -> List[Optional[dict]]:
    """Batch function: called once per asyncio tick with all pending keys."""
    # keys is a list of user IDs — query all at once
    async with db_pool.acquire() as conn:
        rows = await conn.fetch(
            "SELECT id, name, email FROM users WHERE id = ANY($1::text[])",
            keys,
        )
    # CRITICAL: return in same order as keys, same length
    by_id = {row["id"]: dict(row) for row in rows}
    return [by_id.get(k) for k in keys]  # None for missing keys

async def load_posts_by_author(author_ids: List[str]) -> List[List[dict]]:
    """Returns a list of lists — one list of posts per author ID."""
    async with db_pool.acquire() as conn:
        rows = await conn.fetch(
            "SELECT * FROM posts WHERE author_id = ANY($1::text[]) ORDER BY created_at DESC",
            author_ids,
        )
    by_author: dict = {}
    for row in rows:
        by_author.setdefault(row["author_id"], []).append(dict(row))
    # Return list in same order as author_ids
    return [by_author.get(aid, []) for aid in author_ids]

@strawberry.type
class Context:
    user_loader: DataLoader
    post_loader: DataLoader
    viewer_id: Optional[str]

# context_getter is called once per HTTP request — fresh DataLoaders per request
async def get_context(request: Request) -> Context:
    return Context(
        # New DataLoader instances — cache is scoped to this single request
        user_loader=DataLoader(load_fn=load_users),
        post_loader=DataLoader(load_fn=load_posts_by_author),
        viewer_id=extract_viewer_id(request),
    )

@strawberry.type
class PostType:
    id: str
    title: str
    author_id: str

    @strawberry.field
    async def author(self, info: strawberry.types.Info["Context", None]) -> Optional["UserType"]:
        # All author lookups within this request are batched into a single DB query
        return await info.context.user_loader.load(self.author_id)

DataLoader's batch function must return a list of the same length as the input keys list, in the same order. If a key has no corresponding result, return None for that position. Returning a shorter list or a dict causes a DataLoader error that surfaces as a resolver exception. For keys that map to multiple results (like "posts by author"), return a list of lists — one inner list per input key.

Permission classes — sync vs async and the truthy-coroutine trap

Strawberry's BasePermission has a has_permission method that resolvers call before executing. The default signature is synchronous. If you write async def has_permission without registering it as an async permission class, Strawberry calls it as a regular function — which in Python returns a coroutine object, not the result of awaiting it. A coroutine object is always truthy, so every permission check passes regardless of what the async function actually returns.

import strawberry
from strawberry.permission import BasePermission
from strawberry.types import Info
from typing import Any
import jwt

# WRONG: async def inside a sync BasePermission subclass
class IsAuthenticatedBroken(BasePermission):
    message = "User is not authenticated"

    # This LOOKS like it works but is broken:
    # Strawberry calls has_permission() as a regular function.
    # Python returns a coroutine object without awaiting it.
    # The coroutine object is truthy — all checks pass.
    async def has_permission(  # type: ignore
        self, source: Any, info: Info, **kwargs: Any
    ) -> bool:
        token = info.context.request.headers.get("authorization", "").removeprefix("Bearer ")
        payload = await verify_token_async(token)  # never actually awaited!
        return payload is not None

# RIGHT: sync permission class with pre-resolved context
class IsAuthenticated(BasePermission):
    message = "User is not authenticated"

    def has_permission(self, source: Any, info: Info, **kwargs: Any) -> bool:
        # viewer_id is set in context_getter during request setup (sync-safe)
        return info.context.viewer_id is not None

class HasRole(BasePermission):
    message = "Insufficient role"

    def __init__(self, required_role: str):
        self.required_role = required_role

    def has_permission(self, source: Any, info: Info, **kwargs: Any) -> bool:
        return info.context.viewer_role == self.required_role

# Using permission classes on fields
@strawberry.type
class SecureQuery:
    @strawberry.field(permission_classes=[IsAuthenticated])
    def my_profile(self, info: Info) -> "UserType":
        return fetch_user(info.context.viewer_id)

    @strawberry.field(permission_classes=[IsAuthenticated, HasRole("admin")])
    def admin_dashboard(self, info: Info) -> "DashboardType":
        return build_admin_dashboard()

# RIGHT: if you need async permission checks, use strawberry's experimental async support
# or resolve the async check in context_getter before the resolvers run
async def get_context(request: Request) -> "AppContext":
    # Do the async auth check here — once per request, not per field
    token = request.headers.get("authorization", "").removeprefix("Bearer ")
    viewer = None
    if token:
        try:
            payload = await verify_token_async(token)  # awaited here in context
            viewer = await fetch_viewer_from_db(payload["sub"])
        except Exception:
            pass  # viewer stays None — permission class will reject the field

    return AppContext(viewer=viewer, db_pool=request.app.state.db_pool)

@strawberry.type
class AppContext:
    viewer: Optional["UserType"]
    db_pool: Any

    @property
    def viewer_id(self) -> Optional[str]:
        return self.viewer.id if self.viewer else None

Health probe — Strawberry GraphQL reachability from an MCP server

Strawberry mounts a GraphQL endpoint that responds to POST requests with JSON. The minimal probe sends { __typename } — a GraphQL introspection meta-field available on all schemas without authentication. A 200 response with {"data":{"__typename":"Query"}} confirms the endpoint is up, the schema is loaded, and the execution pipeline is functional.

import httpx
import asyncio

async def check_strawberry_health(graphql_url: str) -> dict:
    """Health probe for Strawberry GraphQL — minimal introspection query."""
    async with httpx.AsyncClient(timeout=5.0) as client:
        try:
            response = await client.post(
                graphql_url,
                json={"query": "{ __typename }"},
                headers={"Content-Type": "application/json"},
            )

            if response.status_code != 200:
                return {"healthy": False, "error": f"HTTP {response.status_code}"}

            body = response.json()
            if "errors" in body:
                return {"healthy": False, "error": body["errors"][0].get("message")}

            if body.get("data", {}).get("__typename") != "Query":
                return {"healthy": False, "error": f"Unexpected response: {body}"}

            return {
                "healthy": True,
                "latency_ms": int(response.elapsed.total_seconds() * 1000),
            }

        except httpx.TimeoutException:
            return {"healthy": False, "error": "timeout after 5s"}
        except Exception as e:
            return {"healthy": False, "error": str(e)}

# Startup check for Python MCP server
async def start_mcp_server():
    health = await check_strawberry_health(os.environ["GRAPHQL_URL"])
    if not health["healthy"]:
        print(f"Strawberry GraphQL unreachable at startup: {health['error']}")
        raise SystemExit(1)
    print(f"GraphQL reachable ({health['latency_ms']}ms). Starting MCP server.")

# Run on startup: asyncio.run(start_mcp_server())

# Deeper probe: include an authenticated query to check permission pipeline
async def check_authenticated_probe(graphql_url: str, token: str) -> dict:
    async with httpx.AsyncClient(timeout=5.0) as client:
        response = await client.post(
            graphql_url,
            json={"query": "query HealthProbe { viewer { id } }"},
            headers={
                "Content-Type": "application/json",
                "Authorization": f"Bearer {token}",
            },
        )
        body = response.json()
        # Permission class failure returns errors, not HTTP 4xx
        auth_ok = "errors" not in body or not any(
            "authenticated" in e.get("message", "").lower()
            for e in body.get("errors", [])
        )
        return {"healthy": response.status_code == 200 and auth_ok, "body": body}

AliveMCP monitors Python Strawberry endpoints by tracking both the lightweight { __typename } probe and a deeper authenticated probe for endpoints where auth configuration changes are likely — catching asyncio event loop stalls (probe times out) and permission class regressions (probe returns auth errors) before they affect production MCP tool calls.