Skip to content

Authentication Middleware, JWT & Google IAP

Authentication for the /mcp endpoint is a native Litestar middleware concern. Whatever authentication middleware your application installs—a custom AbstractAuthenticationMiddleware, Litestar's OAuth2PasswordBearerAuth, litestar-security, or an identity-proxy JWT assertion validator—executes before any route handler. When MCPHandlerService dispatches a tool, resource, or route-bound prompt through the internal ASGI pipeline, the target handler inherits request.user and request.auth identically to a direct HTTP request.

litestar-mcp ships no proprietary token validator, JWKS cache, or OAuth discovery route of its own so your HTTP API and MCP surface never diverge.

1. Bring Your Own Litestar Middleware (AbstractAuthenticationMiddleware)

Subclass litestar.middleware.AbstractAuthenticationMiddleware and register it with DefineMiddleware on your Litestar application. The middleware inspects the inbound ASGIConnection (headers, cookies, or client certificates) and returns an AuthenticationResult(user=..., auth=...):

docs/examples/snippets/auth_middleware.py
"""Snippet: authenticate MCP with the app's own Litestar middleware. Referenced from docs/usage/auth.rst."""

from typing import Any

from litestar import Litestar, Request, get
from litestar.connection import ASGIConnection
from litestar.exceptions import NotAuthorizedException
from litestar.middleware import AbstractAuthenticationMiddleware, AuthenticationResult, DefineMiddleware

from litestar_mcp import LitestarMCP

API_KEYS = {"let-me-in": {"sub": "demo-user", "roles": ["reader"]}}


# start-example
class ApiKeyAuthenticationMiddleware(AbstractAuthenticationMiddleware):
    """Populate ``request.user`` / ``request.auth`` from an ``X-API-Key`` header."""

    async def authenticate_request(self, connection: "ASGIConnection[Any, Any, Any, Any]") -> "AuthenticationResult":
        claims = API_KEYS.get(connection.headers.get("X-API-Key", ""))
        if claims is None:
            msg = "Invalid API key"
            raise NotAuthorizedException(msg)
        return AuthenticationResult(user=claims["sub"], auth=claims)


@get("/whoami", mcp_tool=True, sync_to_thread=False)
def whoami(request: "Request[Any, Any, Any]") -> "dict[str, Any]":
    return {"user": request.user, "roles": request.auth["roles"]}


app = Litestar(
    route_handlers=[whoami],
    middleware=[DefineMiddleware(ApiKeyAuthenticationMiddleware)],
    plugins=[LitestarMCP()],
)
# end-example

When a client sends POST /mcp with X-API-Key: let-me-in:

  1. ApiKeyAuthenticationMiddleware.authenticate_request validates the header on the /mcp request and populates scope["user"] and scope["auth"].
  2. During tools/call for whoami, execute_tool synthesizes an internal ASGI scope that copies request.user, request.auth, and request headers from the parent /mcp request.
  3. The whoami handler reads request.user ("demo-user") and request.auth["roles"] (["reader"]) without any MCP-specific branching.

2. JWT & OAuth2PasswordBearerAuth

Litestar's built-in JWT backends (litestar[jwt]) work unchanged with LitestarMCP. Install litestar[jwt] alongside litestar-mcp:

uv add "litestar[jwt]" litestar-mcp

pip install "litestar[jwt]" litestar-mcp

pdm add "litestar[jwt]" litestar-mcp

poetry add "litestar[jwt]" litestar-mcp

Configure OAuth2PasswordBearerAuth with your retrieve_user_handler, token_secret, accepted_issuers, and accepted_audiences, then attach oauth_backend.on_app_init to Litestar(on_app_init=[...]):

docs/examples/notes/shared/auth.py
def build_oauth_backend(
    *,
    secret: "str",
    issuer: "str" = DEFAULT_ISSUER,
    audience: "str" = DEFAULT_AUDIENCE,
    token_url: "str" = DEFAULT_TOKEN_PATH,
    exclude: "list[str] | None" = None,
) -> "OAuth2PasswordBearerAuth[AuthenticatedIdentity, Token]":
    """Build the Litestar OAuth2 bearer backend for the reference notes examples."""
    return OAuth2PasswordBearerAuth[AuthenticatedIdentity, Token](
        token_secret=secret,
        token_url=token_url,
        retrieve_user_handler=_retrieve_identity_from_token,
        algorithm=DEFAULT_ALGORITHM,
        exclude=exclude or ["^/.well-known/"],
        accepted_issuers=[issuer],
        accepted_audiences=[audience],
    )
docs/examples/notes/sqlspec/jwt_auth.py
def _sign(sub: "str") -> "str":
    return mint_hs256_token(sub, secret=token_secret, issuer=issuer, audience=audience)

oauth_backend = build_oauth_backend(secret=token_secret, issuer=issuer, audience=audience)

async def _provide_resolved_user(request: "Request[Any, Any, Any]") -> "AuthenticatedIdentity":
    user = request.user
    if not isinstance(user, AuthenticatedIdentity):
        msg = "Authenticated identity is required"
        raise NotAuthorizedException(msg)
    return user

async def note_service_provider() -> "AsyncIterator[SQLSpecNoteService]":
    async with provide_note_service(sqlspec, config) as service:
        yield service

class NoteController(Controller):
    path = "/notes"
    dependencies = {
        "note_service": Provide(note_service_provider),
        "resolved_user": Provide(_provide_resolved_user),
    }

    @get("/", mcp_tool=True)
    async def list_notes(
        self, note_service: "SQLSpecNoteService", resolved_user: "AuthenticatedIdentity"
    ) -> "list[Note]":
        rows = await note_service.list_for_owner(resolved_user.sub)
        return [msgspec.convert(note_row_to_public(row), Note) for row in rows]

    @post("/", mcp_tool=True)
    async def create_note(
        self, data: "CreateNoteInput", note_service: "SQLSpecNoteService", resolved_user: "AuthenticatedIdentity"
    ) -> "Note":
        row = await note_service.create(title=data.title, body=data.body, owner_sub=resolved_user.sub)
        return msgspec.convert(note_row_to_public(row), Note)

    @delete("/{note_id:str}", status_code=HTTP_200_OK, mcp_tool=True)
    async def delete_note(
        self, note_id: "str", note_service: "SQLSpecNoteService", resolved_user: "AuthenticatedIdentity"
    ) -> "DeleteNoteResult":
        deleted = await note_service.delete_for_owner(note_id, resolved_user.sub)
        return DeleteNoteResult(deleted=deleted, note_id=note_id)

@get("/notes/schema", mcp_resource=True, sync_to_thread=False)
def notes_schema() -> "NotesSchema":
    return NotesSchema()

@get("/app/info", mcp_resource=APP_INFO_RESOURCE_NAME, sync_to_thread=False)
def get_api_info() -> "AppInfo":
    return build_app_info(backend="sqlspec", auth_mode="jwt")

async def on_startup() -> "None":
    await bootstrap_schema(sqlspec, config)

return Litestar(
    route_handlers=[
        build_login_controller(user_directory=user_directory or DEFAULT_USER_DIRECTORY, token_signer=_sign),
        NoteController,
        notes_schema,
        get_api_info,
    ],
    on_startup=[on_startup],
    plugins=[SQLSpecPlugin(sqlspec), LitestarMCP()],
    on_app_init=[oauth_backend.on_app_init],
)

Excluding Public Discovery Paths

Pass exclude=["^/.well-known/", "^/healthz"] to OAuth2PasswordBearerAuth (or your custom AbstractAuthenticationMiddleware) when your application serves unauthenticated health checks, RFC 9728 /.well-known/oauth-protected-resource documents, or public A2A /.well-known/agent-card.json metadata alongside protected /mcp endpoints.

3. Identity Proxies & Google Cloud IAP (x-goog-iap-jwt-assertion)

Zero-trust identity proxies such as Google Cloud Identity-Aware Proxy (IAP) authenticate the user upstream at the load balancer and forward a signed ES256 JWT assertion in the x-goog-iap-jwt-assertion HTTP header.

To authenticate MCP and REST calls behind Google IAP, verify the x-goog-iap-jwt-assertion header against Google's public JWKS (https://www.gstatic.com/iap/verify/public_key-jwk), enforcing both issuer="https://cloud.google.com/iap" and your Cloud Run / Backend Service audience:

docs/examples/notes/shared/auth.py
class IAPAuthenticationMiddleware(AbstractAuthenticationMiddleware):
    """Validate the Google IAP assertion header and populate ``request.user`` / ``request.auth``."""

    def __init__(
        self,
        app: "ASGIApp",
        *,
        audience: "str",
        issuer: "str" = DEFAULT_IAP_ISSUER,
        jwks_url: "str" = DEFAULT_IAP_JWKS_URL,
        **kwargs: "Any",
    ) -> "None":
        super().__init__(app, **kwargs)
        self._validate = build_iap_token_validator(audience=audience, issuer=issuer, jwks_url=jwks_url)

    async def authenticate_request(self, connection: "ASGIConnection[Any, Any, Any, Any]") -> "AuthenticationResult":
        token = connection.headers.get(IAP_HEADER_NAME)
        if not token:
            msg = "Missing IAP assertion"
            raise NotAuthorizedException(msg)
        claims = await self._validate(token)
        if claims is None:
            msg = "Invalid IAP assertion"
            raise NotAuthorizedException(msg)
        return AuthenticationResult(user=identity_from_claims(claims), auth=claims)


def build_iap_auth_middleware(
    *, audience: "str", issuer: "str" = DEFAULT_IAP_ISSUER, jwks_url: "str" = DEFAULT_IAP_JWKS_URL
) -> "DefineMiddleware":
    """Build the IAP authentication middleware for the reference notes example."""
    return DefineMiddleware(IAPAuthenticationMiddleware, audience=audience, issuer=issuer, jwks_url=jwks_url)
docs/examples/notes/sqlspec/google_iap.py
async def _provide_resolved_user(request: "Request[Any, Any, Any]") -> "AuthenticatedIdentity":
    user = request.user
    if isinstance(user, AuthenticatedIdentity):
        return user
    auth = request.scope.get("auth")
    if isinstance(auth, dict):
        return identity_from_claims(auth)
    msg = "Missing IAP identity"
    raise NotAuthorizedException(msg)

async def note_service_provider() -> "AsyncIterator[SQLSpecNoteService]":
    async with provide_note_service(sqlspec, config) as service:
        yield service

class NoteController(Controller):
    path = "/notes"
    dependencies = {
        "note_service": Provide(note_service_provider),
        "resolved_user": Provide(_provide_resolved_user),
    }

    @get("/", mcp_tool=True)
    async def list_notes(
        self, note_service: "SQLSpecNoteService", resolved_user: "AuthenticatedIdentity"
    ) -> "list[Note]":
        rows = await note_service.list_for_owner(resolved_user.sub)
        return [msgspec.convert(note_row_to_public(row), Note) for row in rows]

    @post("/", mcp_tool=True)
    async def create_note(
        self, data: "CreateNoteInput", note_service: "SQLSpecNoteService", resolved_user: "AuthenticatedIdentity"
    ) -> "Note":
        row = await note_service.create(title=data.title, body=data.body, owner_sub=resolved_user.sub)
        return msgspec.convert(note_row_to_public(row), Note)

    @delete("/{note_id:str}", status_code=HTTP_200_OK, mcp_tool=True)
    async def delete_note(
        self, note_id: "str", note_service: "SQLSpecNoteService", resolved_user: "AuthenticatedIdentity"
    ) -> "DeleteNoteResult":
        deleted = await note_service.delete_for_owner(note_id, resolved_user.sub)
        return DeleteNoteResult(deleted=deleted, note_id=note_id)

@get("/notes/schema", mcp_resource=True, sync_to_thread=False)
def notes_schema() -> "NotesSchema":
    return NotesSchema()

@get("/app/info", mcp_resource=APP_INFO_RESOURCE_NAME, sync_to_thread=False)
def get_api_info() -> "AppInfo":
    return build_app_info(backend="sqlspec", auth_mode="google_iap")

async def on_startup() -> "None":
    await bootstrap_schema(sqlspec, config)

return Litestar(
    route_handlers=[NoteController, notes_schema, get_api_info],
    on_startup=[on_startup],
    plugins=[SQLSpecPlugin(sqlspec), LitestarMCP()],
    middleware=[build_iap_auth_middleware(audience=audience, issuer=issuer, jwks_url=jwks_url)],
)

See Notes Reference Examples Matrix for runnable SQLSpec and Advanced Alchemy variants covering no-auth, JWT, Cloud Run, and Google IAP with both Litestar Provide and Dishka (FromDishka[T]) dependency injection.

4. Authenticating Stdio & HTTP Bridge Sessions

In-Process Stdio (MCPStdioContext)

When running in-process over standard input/output (litestar mcp stdio, run_stdio, run_stdio_async, or MCP.run(transport="stdio")), every JSON-RPC frame is dispatched through the same ASGI application. You can authenticate stdio sessions in two ways:

  1. Header Injection via CLI Flags: Pass --bearer-env, --bearer-cmd, or --header to litestar mcp stdio so your application's HTTP authentication middleware validates the synthesized Authorization (or custom --header-name) header on every request: 
    litestar --app my_app:app mcp stdio --bearer-env MCP_API_TOKEN
  2. Pre-Seeded Identity via MCPStdioContext: Standalone local applications that run without HTTP token middleware can resolve identity from the host OS or environment at startup and pass it to MCPStdioContext: 
    from litestar_mcp import MCPStdioContext
from litestar_mcp.mcp.stdio import run_stdio

context = MCPStdioContext(
    client_id="desktop-host",
    owner_id="user:alice",
    user={"sub": "alice", "roles": ["editor"]},
    auth={"sub": "alice", "scopes": ["notes:read", "notes:write"]},
)
run_stdio(app, stdio_context=context)

Anonymous Stdio Has No Shared Owner ID

When MCPStdioContext.owner_id is None and no authenticated principal is present on scope["user"] or scope["auth"], litestar-mcp assigns owner_id=None rather than a shared "stdio" string. Always supply an authenticated principal or explicit owner_id when running multi-user or persisted task stores.

HTTP-to-Stdio Bridge (litestar mcp bridge)

When desktop MCP hosts connect to a remote Litestar deployment through litestar mcp bridge, pass bearer credentials or custom proxy headers at the CLI boundary:

litestar --app my_app:app mcp bridge \
  --base-url https://api.example.com \
  --bearer-env MCP_BEARER_TOKEN

litestar --app my_app:app mcp bridge \
  --base-url https://api.example.com \
  --bearer-cmd "gcloud auth print-identity-token"

litestar --app my_app:app mcp bridge \
  --base-url https://iap.example.com \
  --header-name "x-goog-iap-jwt-assertion" \
  --token-prefix "" \
  --bearer-env IAP_ASSERTION_JWT

When --bearer-cmd is configured and the remote Litestar server responds with HTTP 401 Unauthorized, the bridge automatically re-executes the command once to mint a fresh token and retries the request.

Next Steps