unique_mcp

Shared auth and context wiring for FastMCP servers in the Unique platform. Used as a dependency by MCP servers in this repo to handle per-request authentication against Zitadel and to build UniqueSettings / UniqueServiceFactory for tool handlers.

Problem → Solution

MCP tools must call Unique APIs on behalf of the requesting user — every tool invocation needs a UniqueSettings with the correct user_id and company_id. Hard-coding a single identity in env vars breaks multi-tenant deployments and leaks credentials.

The MCP server acts as an OAuth proxy: clients receive a FastMCP-issued JWT, which the server swaps server-side for the stored Zitadel token on every request. The Zitadel token should contain sub and the company claim, but this depends on token configuration and can't be assumed.

You wire a normal FastMCP instance with the Zitadel OAuth proxy (create_zitadel_oauth_proxy), then inject get_unique_settings / get_unique_settings_async (and optionally get_unique_userinfo / get_unique_service_factory) via Depends() into each tool.

get_unique_settings (sync) — three-priority strategy:

Both user-id and company-id must be present for priority 1 or 2 to apply. The sync helper does not call Zitadel /userinfo — incomplete JWTs fall through to env UNIQUE_AUTH_*.

Why _meta ranks below the token: _meta is caller-supplied and not bound to the bearer token. If it outranked the JWT, any client able to set tools/call._meta could assert an arbitrary user_id/company_id and read another tenant's data. It is therefore ignored entirely once an access token is present.

get_unique_settings_async — same as above, but inserts Zitadel /userinfo before the env fallback. Prefer this in tools that must act as the logged-in user. If an access token is present but neither JWT nor userinfo yield both IDs, it raises instead of using the fixed service user.

get_unique_userinfo is also available on its own when you need profile fields (e.g. email).

flowchart TD
    A([Tool call arrives]) --> D{Zitadel JWT has sub\n+ company claim?}
    D -- yes --> E[Use Zitadel JWT claims]
    D -- no --> H{Async resolver?\nget_unique_settings_async}
    H -- yes --> I{Zitadel /userinfo\nyields sub + company?}
    I -- yes --> J[Use userinfo identity]
    I -- no --> K{Access token present?}
    K -- yes --> L([Raise: refuse env fallback])
    K -- no --> B{_meta contains\nuser-id + company-id?}
    B -- yes --> C[Use _meta identity]
    B -- no --> F[Use env-loaded UniqueSettings auth]
    H -- no, sync --> K
    C & E & J --> G[Build UniqueSettings → tool executes]
    F --> G

OAuth scopes

The OAuthProxy advertises these valid scopes:

Usage

Construct the MCP server yourself: ServerSettings + ZitadelOAuthProxySettings, then create_zitadel_oauth_proxy, then register tools that depend on the injectors.

from fastmcp import FastMCP
from fastmcp.dependencies import Depends
from key_value.aio.stores.memory import MemoryStore

from unique_mcp import get_unique_settings, get_unique_service_factory, get_unique_userinfo
from unique_mcp.auth.zitadel.oauth_proxy import (
    ZitadelOAuthProxySettings,
    create_zitadel_oauth_proxy,
)
from unique_mcp.settings import ServerSettings
from unique_toolkit.app.unique_settings import UniqueSettings

server_settings = ServerSettings()
zitadel_settings = ZitadelOAuthProxySettings()

oauth_proxy = create_zitadel_oauth_proxy(
    client_storage=MemoryStore(),  # swap for shared durable store in prod
    mcp_server_base_url=server_settings.base_url.encoded_string(),
    zitadel_oauth_proxy_settings=zitadel_settings,
)

mcp = FastMCP("my-server", auth=oauth_proxy)


@mcp.tool()
async def search(query: str, settings: UniqueSettings = Depends(get_unique_settings)) -> str:
    # `settings` carries the correct user_id + company_id for this request
    return await some_unique_api_call(settings, query)


if __name__ == "__main__":
    s = server_settings
    mcp.run(
        transport=s.transport_scheme,
        host=s.local_base_url.host,
        port=s.local_base_url.port,
    )

Public exports (from unique_mcp import …)

Scenarios

1 — Normal OAuth flow (JWT with full claims)

The common case. The MCP server acts as an OAuth Authorization Server and proxies the login to Zitadel using the token swap pattern:

1. The client authenticates against the MCP server's OAuth endpoints (not Zitadel directly).
2. The MCP server proxies to Zitadel, obtains a Zitadel token, and stores it server-side.
3. The MCP server issues its own short-lived FastMCP JWT to the client.
4. On every tool call, the MCP server swaps the FastMCP JWT for the stored Zitadel token, validates it against Zitadel's JWKS, and extracts claims — no extra network call needed when the Zitadel JWT contains sub + urn:zitadel:iam:user:resourceowner:id.

sequenceDiagram
    participant Client
    participant MCP as MCP Server
    participant Zitadel

    Client->>MCP: GET /.well-known/oauth-authorization-server
    MCP-->>Client: OAuth metadata (authorize/token endpoints)
    Client->>MCP: GET /authorize
    MCP->>Zitadel: redirect (proxy OAuth flow)
    Zitadel-->>Client: login page
    Client->>Zitadel: authenticate
    Zitadel-->>MCP: authorization code (callback)
    MCP->>Zitadel: POST /oauth/v2/token (exchange code)
    Zitadel-->>MCP: Zitadel JWT (stored server-side, never sent to client)
    MCP-->>Client: FastMCP JWT (reference token)

    Client->>MCP: tools/call + Authorization: Bearer <FastMCP JWT>
    MCP->>MCP: verify FastMCP JWT signature → look up JTI → retrieve stored Zitadel JWT
    MCP->>MCP: validate Zitadel JWT via JWKS, extract sub + company_id claims
    MCP->>MCP: build UniqueSettings
    MCP-->>Client: tool result

2 — JWT without company claim (userinfo before env)

If the Zitadel JWT carries sub but not the company claim, get_unique_settings (sync) falls back to environment identity (UNIQUE_AUTH_*). That is wrong for multi-user servers.

Use await get_unique_settings_async() (or call get_unique_userinfo) so identity comes from Zitadel /userinfo instead. Configure Zitadel so JWTs embed the resourceowner claim when possible — see docs/zitadel/README.md — to avoid the extra userinfo round-trip.

sequenceDiagram
    participant Client
    participant MCP as MCP Server
    participant Zitadel

    Client->>MCP: tools/call + Authorization: Bearer <FastMCP JWT>
    MCP->>MCP: token swap → retrieve Zitadel JWT
    Note over MCP: JWT incomplete for get_unique_settings (sync) → env auth
    MCP->>Zitadel: get_unique_settings_async: GET /oidc/v1/userinfo (Bearer Zitadel JWT)
    alt userinfo has sub + company
        Zitadel-->>MCP: sub, urn:zitadel:...:id, email, ...
        Note over MCP: Use userinfo identity
    else userinfo incomplete
        Note over MCP: Raise — refuse env fallback for a logged-in request
    end
    MCP-->>Client: tool result

3 — Platform-internal caller supplying identity via _meta

An internal service calls the tool on behalf of a known user by passing identity directly in the MCP _meta field. Both unique.app/auth/user-id and unique.app/auth/company-id must be present; if either is missing the provider falls through to env resolution.

Security: _meta values are taken as-is, with no validation and no binding to the bearer token. They are therefore honoured only when the request carries no access token. Sending _meta identity alongside a Bearer token has no effect — the token wins, and on the async resolver an unresolvable token raises rather than falling back. Use _meta only from callers you fully trust, and never expose it to external users.

{
  "method": "tools/call",
  "params": {
    "name": "search",
    "arguments": { "query": "hello" },
    "_meta": {
      "unique.app/auth/user-id": "user-abc123",
      "unique.app/auth/company-id": "company-xyz456"
    }
  }
}

sequenceDiagram
    participant InternalSvc as Internal Service
    participant MCP as MCP Server

    InternalSvc->>MCP: tools/call + _meta (no Authorization header)
    MCP->>MCP: no access token → _meta identity is eligible
    alt _meta has both user-id + company-id
        MCP->>MCP: build UniqueSettings from _meta
        MCP->>MCP: call Unique API with provided identity
        alt identity is valid
            MCP-->>InternalSvc: tool result
        else user-id or company-id not recognised by Unique
            MCP-->>InternalSvc: error (API rejects identity)
        end
    else _meta incomplete or absent
        MCP->>MCP: fall through to env auth
        MCP-->>InternalSvc: result or misconfiguration
    end

Configuration

UNIQUE_MCP_* — server settings:

ZITADEL_* — OAuth proxy settings:

Env-based user/company identity for tools (when JWT/_meta do not supply auth) comes from unique-toolkit / UniqueSettings.from_env_auto_with_sdk_init() (for example UNIQUE_AUTH_* where applicable in your deployment).

Zitadel setup

See docs/zitadel/README.md for step-by-step instructions: creating the OAuth app, enabling JWT token type with embedded org claims, configuring redirect URIs (including ngrok for local dev), and required scopes.

Development

cd unique_mcp && uv run pytest tests/ -q

Platform helpers (logging + metrics)

On import, unique_mcp sets FASTMCP_CHECK_FOR_UPDATES=off (unless already set). Call configure_tracing from unique-toolkit[otel] when you want Tempo/OTLP export.

from unique_toolkit.monitoring import configure_tracing
from unique_mcp.logging import configure_logging
from unique_mcp.monitoring import setup_ops

configure_tracing(service_name="my-mcp")
configure_logging()

mcp = FastMCP("my-server")
middleware = [...]
middleware.append(setup_ops(mcp))

mcp.run(transport="http", middleware=middleware)