Give an AI agent the user's identity, not a god-mode service account

An MCP server exposes tools that call real backends — read this, write that, run the other thing. The path of least resistance is to give the server one service account with broad rights and let every user's agent act through it. That's a disaster in slow motion: every user's agent can touch every user's data, your backend's per-user permissions are bypassed, and your audit log says "the bot did it."

I wanted the opposite: the agent acts strictly as the human driving it, with exactly their permissions. So the MCP server holds no data-access identity of its own. It authenticates each caller, captures the caller's identity token, and forwards that same token to every downstream backend — which authorize it as the user, using the same permission checks they already have. (This is the identity half of the story; I wrote about the audit and approval half separately.)

There are two client shapes, so two auth modes:

• Resource-server mode — the MCP client already holds a bearer token minted elsewhere; the server just validates it against the identity provider's JWKS (JWTVerifier(jwks_uri=..., issuer=..., audience=...)).
• Proxy mode — for clients that speak MCP's own OAuth flow (dynamic client registration); the server brokers tokens against the corporate OIDC provider (an OIDCProxy).

And here's the bit that cost me the most, specific to proxy mode: the access token the client presents is the proxy's own token — the backends don't trust it. Only the upstream id_token carries the identity the backends will accept. So the server overrides a private library hook to grab that upstream token at verification time and stash it in a contextvars.ContextVar, and a single resolver feeds it to every tool's HTTP client:

import contextvars
USER_ID_TOKEN: contextvars.ContextVar[str | None] = contextvars.ContextVar("user_id_token", default=None)
 
def build_auth(cfg):
    if cfg.mode == "oidc-proxy":
        if not hasattr(OIDCProxy, "_get_verification_token"):   # fail LOUD on a lib upgrade
            raise RuntimeError("OIDCProxy hook renamed; re-point the id_token capture")
 
        class CapturingProxy(OIDCProxy):
            def _get_verification_token(self, upstream):        # private hook, overridden on purpose
                tok = super()._get_verification_token(upstream)
                if tok:
                    USER_ID_TOKEN.set(tok)                      # stash the USER's upstream id_token
                return tok
        return CapturingProxy(config_url=cfg.oidc_url, client_id=cfg.cid,
                              client_secret=cfg.secret, base_url=cfg.base_url,
                              required_scopes=["openid", "user"], verify_id_token=True)
    return JWTVerifier(jwks_uri=cfg.jwks_uri, issuer=cfg.issuer,
                       audience=cfg.audience or None, algorithm="RS256")
 
def user_token() -> str:                # every tool's client calls this — never a service account
    if (t := USER_ID_TOKEN.get()):
        return t                        # proxy mode: the captured upstream id_token
    if (a := get_access_token()) and a.token:
        return a.token                  # resource-server mode: the verified bearer
    authz = (get_http_headers() or {}).get("authorization", "")
    return authz[7:] if authz.lower().startswith("bearer ") else ""

Things that bit, or nearly did:

• "Just forward the bearer" works in resource-server mode and breaks only in proxy mode — which means it passes every test until the day someone uses a proxy client. Capturing the upstream token is what makes both modes behave.
• The capture point is a private library method. Guard it with hasattr at startup and raise a loud, descriptive error on a version bump — otherwise a routine upgrade silently downgrades you to "no user identity forwarded," and everything keeps working as the service account.
• Use a ContextVar, not a thread-local or a global. It's what keeps per-request identity correct under async concurrency.

The hard part: federated SSO, and two services agreeing on who you are

The bug that forced all of this into focus: users who signed in through a corporate SSO provider could authenticate to the agent fine, yet every call died with "couldn't retrieve your groups." The web UI logs those users in directly against the SSO provider — but the agent's path brokers the login through a token service that re-mints its own token, and that minted token's username claim was bound to a legacy-directory alias a federated SSO user simply doesn't have. Blank username → no user resolved → no groups. And a never-before-seen SSO user had no account at all.

Fixing it drove home how much care identity needs when it's derived in more than one place — here a Python API and a separate gateway each extract it independently, so any disagreement is authorization drift (a permission cache keyed on one username, a resource check resolving a different one). Three moves keep it honest:

Every extractor follows the same recipe, and the username claim is configured identically everywhere. Decode the token unverified just to read the issuer, look up that issuer's keys from a trusted-issuers map, verify the signature, then read the username from a per-issuer configurable claim — the same claim name on every service. A default of "username" on one service and an explicit "sub" on another is a silent divergence bug.

Alias coalescing has to run identically in each extractor — you can't fully push it to the minter. The clean instinct is "resolve one canonical username when the token is minted." But the token broker couldn't express "first non-empty alias," so instead it emits all of a person's aliases — one per login method (the SSO alias, the legacy-directory alias, …) — as arrays in the token, and each service picks the first non-empty one. The load-bearing part: because two services extract identity independently, that "first non-empty" rule has to be duplicated faithfully in both (a Go gateway and a Python API), or they resolve different users from the same token.

Provision on first sight. A federated user who never logged into the web UI has no account at all — so the bearer path just-in-time provisions one on first call, creating the user and deriving their groups and admin flags from the groups already carried in their token. No prior web login required.

def username_from_token(token: str) -> str:
    unv = jwt.decode(token, options={"verify_signature": False})  # unverified: only to read iss
    prov = TRUSTED_ISSUERS.get(unv["iss"])
    if prov is None:
        raise PermissionError("untrusted issuer")
    key = prov.jwks.get_signing_key_from_jwt(token).key
    claims = jwt.decode(token, key, algorithms=["RS256"], options={"verify_aud": False})
    # One alias per login method rides in the token; pick the first non-empty —
    # and BOTH services that extract identity must run this identical rule.
    if scalar := claims.get(prov.user_claim):
        return scalar
    for alias in claims.get(prov.alias_claim, []):
        if alias and alias.strip():
            return alias.strip()
    raise PermissionError("no username in token")  # fail closed on a blank identity
 
# Authorization: positive-only, TTL-bounded, fail-closed.
def accessible_ids(user: str, ttl: float, fetch) -> set[str]:
    hit = _cache.get(user)
    if hit and monotonic() - hit.ts <= ttl:
        return hit.ids
    try:
        ids = fetch(user)                 # forwards the USER's token downstream
    except Exception:
        raise PermissionError("authz check failed; deny")  # fail closed — do NOT cache
    _cache[user] = Entry(ids, monotonic())                 # cache ALLOW only
    return ids

Two deliberate choices in that cache: it's positive-only (caching a denial would turn a transient blip into a lockout), and every check fails closed (an error denies rather than allows). A TTL'd allow-cache also means a revoked membership stops being honored within roughly the TTL, not for the whole session.

The other door: the in-app chat authenticates completely differently

Everything above is the external door — the MCP server, for third-party AI clients that show up holding an OAuth bearer. The same assistant is also reachable inside our own web app, and that path shares none of this authentication.

In the app, the browser is already signed in, so a chat turn just calls our backend over the existing web session (cookie + CSRF). That backend swaps the session for a short-lived, single-use ticket — an opaque handle to a server-side record holding a user-scoped credential and the request context — and the streaming service redeems the ticket server-to-server (shared service secret, atomic single-use claim) to get the credential. No OAuth, no bearer from the client, and it never touches the MCP server; the two are even wired as separate mounted apps.

So there are two front doors with entirely separate caller authentication — cookie-session-then-ticket for the in-app chat, OAuth-bearer-or-proxy for MCP — that converge only at the shared tool layer, where both resolve to the same thing: the operation runs as the end user, never a shared or elevated identity. Different doors, one identity model behind them. And that convergence is exactly what makes two doors safe to have: whichever one you came through, the backend's per-user permissions are the final word.

The reframe: "give the bot a service account" is the easy default and it's wrong — it collapses every user into one super-identity and makes your audit trail meaningless. Hand the agent the user's own token, capture the upstream one specifically when you're proxying, and your backend's existing per-user authorization simply applies — no parallel permission system for the agent, and "who did this?" finally has a real answer. The tax you pay is that identity has to be derived identically everywhere it's derived — so run the same alias-coalescing in every service that extracts it, configure the same claim everywhere, and fail closed on every check.