Skip to content

Latest commit

 

History

History
647 lines (480 loc) · 32.4 KB

File metadata and controls

647 lines (480 loc) · 32.4 KB

authplane-fastmcp User Guide

OAuth 2.1 JWT authentication for FastMCP servers, powered by the Authplane Python SDK.

Table of Contents


Installation

pip install authplane-fastmcp

Requires Python 3.11+.

Quick Start

import asyncio

from fastmcp import FastMCP
from authplane_fastmcp import authplane_auth


async def main() -> None:
    result = await authplane_auth(
        issuer="https://auth.company.com",
        base_url="https://mcp.company.com",
        scopes=["tools/query", "tools/write"],
    )
    mcp = FastMCP("My Server", **result)

    @mcp.tool()
    def query(sql: str) -> str:
        """Execute a query."""
        return f"Ran: {sql}"  # replace with your real handler

    try:
        await mcp.run_async(transport="http", port=8080)
    finally:
        await result.aclose()


asyncio.run(main())

authplane_auth() performs RFC 8414 metadata discovery, fetches the JWKS, and wires up all authentication components. The result unpacks directly into FastMCP().

Configuration Reference

All parameters of authplane_auth():

Parameter Type Default Description
issuer str required Authorization server URL
base_url str required Root URL of this FastMCP server
scopes list[str] [] Scopes this server supports
mcp_path str "/mcp" Mount path of the MCP endpoint. The JWT audience is derived as base_url + mcp_path
as_credentials ASCredentials None Client credentials for introspection and token exchange
dpop DPoPProvider None DPoP provider for outbound calls to the AS (introspection, token exchange)
allowed_algorithms list[str] ["RS256", "ES256"] Allowed JWT signature algorithms (asymmetric only)
jwks_refresh_seconds int 300 JWKS cache TTL
metadata_refresh_seconds int 3600 AS metadata cache TTL
cache_ttl_buffer_seconds float 30.0 Buffer subtracted from token TTLs before cache expiry
default_ttl_seconds float 3600.0 Fallback token cache TTL when responses omit expiry metadata
circuit_breaker_threshold int 5 Transient failures before opening the AS circuit breaker
circuit_breaker_cooldown_seconds float 30.0 Cooldown before allowing a half-open probe
clock_skew_seconds int 30 Leeway for exp/nbf/iat validation
dev_mode bool False Relaxes SSRF checks for local development
revocation_checker see below None Token revocation strategy
fail_closed bool False Reject tokens when the revocation check itself fails, instead of accepting them (see below)
fetch_settings FetchSettings None Full SSRF / fetch settings applied to both metadata and JWKS fetches (overrides dev_mode)
resource_metadata_url str None URL to advertise as RFC 9728 §5.1 resource_metadata instead of the derivation of the resource, for an AS-hosted PRM document. Does not reach the challenge FastMCP emits — see below
inbound_dpop InboundDPoPOptions None Per-resource inbound DPoP policy (replay store, max proof age, clock skew, accepted proof algorithms, required). When set, the resource advertises DPoP support in PRM (RFC 9728 §2). See Inbound DPoP through the FastMCP adapter below for current limitations.

Inbound DPoP through the FastMCP adapter

FastMCP's TokenVerifier builds on the upstream MCP BearerAuthBackend, which matches only Authorization: Bearer ... headers and exposes a bearer-only verify_token(token: str) protocol. RFC 9449 §7.1 DPoP-bound requests use the Authorization: DPoP <token> scheme and would be rejected at the framework layer before reaching this adapter.

Setting inbound_dpop=InboundDPoPOptions(...) therefore affects only Protected Resource Metadata advertising in the standard adapter integration; verify-time DPoP enforcement requires custom request-aware middleware that extracts the proof header and threads a DPoPRequestContext into AuthplaneResource.verify(...).

Scope Enforcement

Use FastMCP's built-in require_scopes decorator to enforce per-tool scope requirements:

from fastmcp.server.auth import require_scopes


@mcp.tool(auth=require_scopes("tools/query"))
def query(sql: str) -> str:
    """Requires the tools/query scope."""
    return f"Ran: {sql}"  # replace with your real handler


@mcp.tool(auth=require_scopes("tools/admin", "tools/delete"))
def delete_all() -> str:
    """Requires BOTH tools/admin AND tools/delete scopes."""
    return clear_database()

FastMCP enforces scopes before the handler runs by filtering tools the caller cannot use out of the catalog. If the token is missing a required scope, the tool is hidden from tools/list, and a tools/call for that tool returns HTTP 200 with {"isError": true, "content": [{"text": "Unknown tool: '<name>'"}]} — not a 403. UX layers expecting a 403 to prompt for re-auth will not see one; key off isError + the tool-not-found content text instead.

Accessing Token Claims

Dependency Injection (Recommended)

from fastmcp.dependencies import CurrentAccessToken
from fastmcp.server.auth import AccessToken


@mcp.tool()
async def my_tool(data: str, token: AccessToken = CurrentAccessToken()) -> str:
    # Standard JWT claims
    sub = token.claims.get("sub")  # Subject (user ID)
    jti = token.claims.get("jti")  # JWT ID
    iss = token.claims.get("iss")  # Issuer
    aud = token.claims.get("aud")  # Audience
    exp = token.claims.get("exp")  # Expiration (Unix timestamp)
    nbf = token.claims.get("nbf")  # Not before
    iat = token.claims.get("iat")  # Issued at

    # OAuth claims
    client_id = token.client_id  # Client ID
    scopes = token.scopes  # List of granted scopes
    expires_at = token.expires_at  # Expiration (Unix timestamp)

    # Custom claims
    tenant = token.claims.get("tenant_id")
    org = token.claims.get("organization")

    return f"Hello {sub} from tenant {tenant}"

The claims dict contains the full JWT payload including all standard and custom claims.

Imperative Access

from fastmcp.server.dependencies import get_access_token


@mcp.tool()
async def my_tool(data: str) -> str:
    token = get_access_token()  # Returns None if unauthenticated
    if token:
        user = token.claims.get("sub")
    return f"Processing {data}"

Protected Resource Metadata (PRM)

The adapter automatically serves RFC 9728 Protected Resource Metadata at the well-known URI. This enables MCP clients to discover the authorization server and supported scopes.

The PRM endpoint location depends on the resource URL:

Resource URL PRM Endpoint
https://mcp.company.com GET /.well-known/oauth-protected-resource
https://mcp.company.com/api/v1 GET /.well-known/oauth-protected-resource/api/v1

The response includes:

  • Authorization server URL (issuer)
  • Supported scopes
  • Bearer token methods
  • Resource identifier

No additional configuration is needed; PRM is served automatically.

Where the PRM document lives

Two topologies, and the adapter supports both:

(a) Resource-hosted — the default. This server serves the document itself at the well-known path derived from the resource (base_url + mcp_path, RFC 9728 §3.1), exactly as the table above shows. Nothing to configure.

(b) AS-hosted. The authorization server serves the document for every registered Resource — authserver >= 0.2.0 serves one at <issuer>/.well-known/oauth-protected-resource/{ref}, where ref is the RFC 9728 §3.1 path suffix of the Resource URI (or its slug) — and this server only points clients at it. Useful when the resource server cannot host well-known paths. Pass resource_metadata_url=:

mcp = FastMCP(
    "My MCP Server",
    **await authplane_auth(
        issuer="https://auth.company.com",
        base_url="https://mcp.company.com",
        resource_metadata_url="https://auth.company.com/.well-known/oauth-protected-resource/mcp",
    ),
)

FastMCP's own challenge keeps the derived URL. No upstream parameter accepts a metadata URL: AuthProvider takes base_url and resource_base_url, both resource identifiers, and fastmcp.server.http derives the challenge URL from AuthProvider._get_resource_url(path) through mcp.server.auth.routes.build_resource_metadata_url before passing it to fastmcp.server.auth.middleware.RequireAuthMiddleware. So the 401 and the 403 insufficient_scope that middleware sends carry the resource-hosted URL whatever you configure here — and resource_base_url is not the lever either, since it also moves the PRM route this adapter serves. What the option does reach is every challenge this SDK composes: AuthplaneTokenVerifier.resource_metadata_url(), and through it any middleware of your own calling response_headers_for(...). Until upstream accepts a URL, option (b) is only fully effective for a server whose 401/403 you emit yourself.

RFC 9728 §3.3 constrains topology (b). The rule binds the document's resource value to the URL the document was fetched from, not to the API URL the client called: the value "MUST be identical to the protected resource's resource identifier value into which the well-known URI path suffix was inserted to create the URL used to retrieve the metadata", and otherwise "MUST NOT be used". Those two readings coincide only when the metadata URL is the §3.1 derivation of the resource — i.e. in topology (a). An AS-hosted document on a different origin is therefore usable only against clients that do not enforce §3.3, and that check is what stops a resource server from pointing a client at metadata for somebody else's resource. See the core user guide for the worked example.

Whichever topology you pick, the Resource URI registered at the AS, the resource derived here from base_url + mcp_path, and this server's public URL must be one identical string — a trailing slash or a :443 spelled out on one side only is a mismatch.

Token Revocation Checking

By default, tokens are validated offline (signature + claims only). You can enable revocation checking to catch tokens that have been revoked before they expire.

No Revocation (Default)

await authplane_auth(
    issuer="https://auth.company.com",
    base_url="https://mcp.company.com",
    # revocation_checker is None by default
)

RFC 7662 Introspection

Calls the authorization server's introspection endpoint to check if a token is still active:

from authplane import ASCredentials, IntrospectionRevocation

await authplane_auth(
    issuer="https://auth.company.com",
    base_url="https://mcp.company.com",
    revocation_checker=IntrospectionRevocation(),
    as_credentials=ASCredentials(
        client_id="my_resource_server",
        client_secret="secret",
    ),
)
  • The introspection endpoint is automatically discovered from AS metadata.

  • If the endpoint returns active=false, the token is rejected with TokenRevokedError.

  • An introspection error lets the token through: if the introspection endpoint is unavailable, the token is accepted (offline validation still applies). This is the default; pass fail_closed=True to refuse instead (see below).

  • as_credentials is required. The resource server's client must be confidential (it has a secret) and must be either the client the token was issued to or a runtime-client of the Resource named in the token's aud. Register it once per resource:

    authserver admin resource runtime-client add --client-id <rs-client-id> --slug <resource-slug>

    A public client cannot introspect at all. Since authserver 0.1.2 an unauthenticated introspection call, or one from a client that is neither the issuer nor a runtime-client, is answered with {"active": false} — not an error — so every token is rejected as revoked. The SDK warns at startup when IntrospectionRevocation is configured without as_credentials, and once per resource the first time active=false comes back for a token that passed local verification.

Failure Policy: Fail-Open vs Fail-Closed

fail_closed controls what happens when the revocation check itself fails — the introspection endpoint is unreachable, returns an error, or a custom checker raises:

await authplane_auth(
    issuer="https://auth.company.com",
    base_url="https://mcp.company.com",
    revocation_checker=IntrospectionRevocation(),
    as_credentials=ASCredentials(
        client_id="my_resource_server",
        client_secret="secret",
    ),
    fail_closed=True,
)
  • False (default) accepts the token and logs a warning. Signature and claims validation still apply, so this only skips the revocation freshness check — it never admits an otherwise-invalid token.
  • True rejects the token with TokenRevokedError. Choose this for servers exposing mutation-capable or otherwise high-impact tools, where serving a revoked-but-unverifiable token is worse than downtime.

Trade-offs to understand before enabling fail_closed=True:

  • Availability: an authorization server or introspection outage makes every request fail with 401 until the outage resolves. Once the client's circuit breaker opens, checks fail fast and all tokens are rejected until the cooldown elapses.
  • Credentials: without valid as_credentials, or with a client that is not a runtime-client of the resource, authserver ≥ 0.1.2 answers active=false rather than an error — so every token is rejected under both policies, and fail_closed does not change that. Verify the credentials and the runtime-client registration as part of deployment, not just at rollout.
  • Metadata: an AS whose metadata document does not advertise introspection_endpoint fails every introspection attempt. Under the default that check is skipped and every request logs a Revocation check failed (fail-open) warning — for a missing endpoint that is every request, permanently, since the condition never clears; under fail_closed=True every token is rejected — and unlike an outage this never self-recovers, because the missing endpoint is a permanent property of the AS configuration. Confirm the endpoint is present in AS metadata before enabling.
  • fail_closed has no effect when revocation_checker is None — the flag is only consulted when a revocation check actually runs. The SDK logs a warning when it detects this misconfiguration.
  • The SDK also logs at INFO when a revocation checker is configured fail-open, so the posture in effect appears in startup output rather than only in the docs. See the core SDK user guide for why that one is not a warning.

Custom Revocation Checker

Implement your own revocation logic with an async callable:

from authplane import VerifiedClaims


async def check_blocklist(claims: VerifiedClaims, raw_token: str) -> bool:
    """Return True to reject the token (it is revoked)."""
    return await redis_client.sismember("revoked_tokens", claims.jti)


await authplane_auth(
    issuer="https://auth.company.com",
    base_url="https://mcp.company.com",
    revocation_checker=check_blocklist,
)

Token Exchange (RFC 8693)

Exchange an inbound token for a narrowly-scoped downstream token to call other services on behalf of the caller. The call goes to the authorization server's token_endpoint (discovered via RFC 8414 metadata), reuses the client's SSRF settings, and attaches DPoP proofs when a DPoPProvider was configured.

from authplane import ASCredentials
from authplane.oauth import TokenExchangeOptions
from authplane_fastmcp import authplane_auth

result = await authplane_auth(
    issuer="https://auth.company.com",
    base_url="https://mcp.company.com",
    scopes=["tools/add"],
    as_credentials=ASCredentials(
        client_id="https://mcp.company.com/mcp",
        client_secret="s3cret",
    ),
)

downstream = await result.client.exchange(
    TokenExchangeOptions(
        subject_token=inbound_token,
        scope="tools/add",  # narrow to the minimum
        resources=("https://downstream.example",),  # RFC 8707 audience binding
    )
)

# downstream.access_token — present to the downstream service
# downstream.expires_in    — lifetime in seconds
# downstream.token_type    — "Bearer" or "DPoP"

TokenExchangeOptions fields:

Field Type Purpose
subject_token str (required) Token being exchanged (typically the inbound caller's token).
subject_token_type str RFC 8693 token-type URI; defaults to urn:ietf:params:oauth:token-type:access_token.
actor_token / actor_token_type str Optional actor (delegation) token.
scope str Space-separated scopes to request on the downstream token.
resources tuple[str, ...] Target resource identifiers (RFC 8707). Binds the downstream token's audience.
audiences tuple[str, ...] Explicit audiences when not using resources.

Operator step — allowlist the exchanging client. authserver 0.2.0 only honours a cross-client exchange when the exchanging client is allowlisted on the target Resource. For each MCP server that exchanges for a downstream resource it does not itself act as, add its client id to that Resource's exchange policy:

PATCH /admin/resources/{id}
{"policy": {"exchange": {"allowed_client_ids": ["<exchanging-client-id>"]}}}

A client exchanging a token that was issued to itself, a fronted exchange, and a Broker resource need nothing.

client.exchange() raises:

  • AccessDeniedError (access_denied, HTTP 403) — the exchanging client is not allowlisted on the target Resource. This is an operator-side fix (the PATCH above); re-prompting the user will not clear it, which is why it is a distinct class from ConsentRequiredError.
  • InvalidTargetError (invalid_target, HTTP 400, RFC 8707 §2.2) — the resource string does not match a granted resource byte for byte; a trailing slash is enough.
  • ConsentRequiredError — the AS requires interactive user consent before issuance.
  • InvalidGrantError on a rejected grant, CircuitOpenError when the AS circuit is open, and other AuthplaneError subclasses for transport/protocol failures. AccessDeniedError and InvalidTargetError never trip the circuit breaker.

See Error Handling and URL Elicitation for Consent below.

URL Elicitation for Consent

When a token exchange needs interactive user consent at the AS (for example, first-time authorization against a third-party service), the AS returns consent_required with a consent_url. MCP surfaces this through the URL elicitation flow (JSON-RPC error -32042). The authplane-mcp adapter wires it up end-to-end.

fastmcp 3.2 does not propagate McpError from tool handlers (its tool dispatch wraps everything except FastMCPError as {"isError": true}), so -32042 never reaches the wire. The wrapped client.exchange() raises UrlElicitationRequiredError (the MCP-shaped form of the consent error) — catch it in the tool body and render the consent URL into the response yourself:

from authplane import ConsentRequiredError
from authplane.oauth import TokenExchangeOptions
from mcp.shared.exceptions import UrlElicitationRequiredError


@mcp.tool(auth=require_scopes("tools/call_downstream"))
async def call_downstream(payload: str) -> str:
    try:
        downstream = await auth_result.client.exchange(
            TokenExchangeOptions(subject_token=..., scope="downstream/write")
        )
    except UrlElicitationRequiredError as error:
        urls = [e.url for e in error.elicitations] if error.elicitations else []
        return f"Consent required: {urls[0] if urls else '<no url>'}"
    except ConsentRequiredError as error:
        # The wrapper only translates to UrlElicitationRequiredError when the
        # AS supplied a consent_url. Without one, the bare error reaches us —
        # surface its formatted description (no URL to render).
        return f"Consent required: {error.describe()}"
    return await downstream_api_call(downstream.access_token, payload)

The client returned by authplane_auth(...) already wraps exchange() to raise UrlElicitationRequiredError for qualifying consent errors. Once fastmcp's tool path propagates McpError, this try/except simply stops triggering — no SDK changes needed. to_url_elicitation_required_error is exported for the same reason.

Development Mode

For local development, enable dev_mode to relax SSRF restrictions and allow HTTP/localhost:

await authplane_auth(
    issuer="http://localhost:9000",
    base_url="http://localhost:8080",
    scopes=["tools/query"],
    dev_mode=True,
)

Development mode allows:

  • HTTP (non-TLS) connections
  • Localhost addresses (127.0.0.0/8)
  • Private network addresses (10.x, 172.16-31.x, 192.168.x)

Cloud metadata addresses (169.254.x) are always blocked, even in dev mode.

You can also enable dev mode via environment variable:

export AUTHPLANE_DEV_MODE=true
python myserver.py

SSRF Protection

The adapter provides SSRF controls for JWKS and metadata fetching via FetchSettings.

For most use cases, dev_mode=True is sufficient for local development. Use FetchSettings when you need fine-grained control:

from authplane import FetchSettings

settings = FetchSettings(
    ssrf_protection=True,
    allow_http=False,
    allow_localhost=True,
    allow_private_networks=True,
    timeout=10.0,
)

await authplane_auth(
    issuer="https://auth.internal.corp",
    base_url="https://api.prod.com",
    fetch_settings=settings,
)

When fetch_settings is provided, dev_mode is ignored for both metadata and JWKS fetches.

Protection Details

Check Default Description
HTTPS required Yes Blocks HTTP unless explicitly allowed
Localhost blocked Yes Blocks 127.0.0.0/8
Private networks blocked Yes Blocks 10.x, 172.16-31.x, 192.168.x
Cloud metadata blocked Always Blocks 169.254.x (cannot be disabled)
DNS pinning Yes Resolves DNS once, validates the IP
Redirect blocking Yes Prevents open redirect attacks
Size limit 64KB Maximum JWKS response size
Timeout 10s HTTP request timeout

Resource Cleanup

authplane_auth() returns an AuthplaneAuthResult that holds background JWKS / metadata refresh tasks and an HTTP connection pool. Call aclose() on shutdown:

import asyncio


async def main() -> None:
    result = await authplane_auth(...)
    try:
        mcp = FastMCP("My Server", **result)
        await mcp.run_async(transport="http", port=8080)
    finally:
        await result.aclose()


asyncio.run(main())

result.aclose() closes the underlying AuthplaneClient, cancels its background tasks, and releases connections. Skipping it surfaces as leaked tasks, open sockets, and ResourceWarning in tests.

Error Handling

Verification path

AuthplaneTokenVerifier.verify_token catches every AuthplaneError raised by AuthplaneResource.verify() (missing/expired/invalid/revoked token, DPoP failure, etc.) and returns None. FastMCP turns that into a uniform 401 Unauthorized on the wire — the wire does not differentiate by error type, but the verifier emits a logging.DEBUG event authplane.token_verification_failed (logger authplane_fastmcp.verifier) with structured error_class and error fields so operators can distinguish expired tokens from JWKS outages from DPoP replays in logs.

Scope enforcement

Scope checks happen after token validation succeeds and are a separate enforcement layer — see Scope Enforcement above for @mcp.tool(auth=require_scopes(...)). Inside a handler, claims.require_scope("…") raises InsufficientScopeError. The error carries required_scopes so the SDK can emit RFC 6750's scope= challenge automatically.

Building a WWW-Authenticate challenge in custom middleware

When you handle an AuthplaneError outside the verifier — typically because you are wrapping the adapter in your own middleware or calling AuthplaneResource.verify() directly — use response_headers_for(error, …) to map the error to (status, {"WWW-Authenticate": challenge}) in one call. It forwards realm, resource_metadata_url, and scope into the underlying www_authenticate() helper, which sanitizes every interpolated value against header injection.

error_description is a fixed sentence chosen by the error code, never the exception's message — the challenge goes to a caller who has not authenticated, and the SDK's messages name the kid, claim, or expected audience that failed. The message stays on the exception and is logged at DEBUG on the authplane.errors logger; verbose_description=True puts it back on the wire for development only.

A resource running inbound_dpop in optional mode accepts both Bearer and DPoP and should advertise both (RFC 9449 §7.1). response_headers_for() cannot express that — a dict holds one value per header name — so pair http_status() with www_authenticate_challenges(error, schemes=("Bearer", "DPoP"), algs=…) and emit one WWW-Authenticate header value per element. See the core SDK user guide for the full example.

from authplane import AuthplaneError, response_headers_for

try:
    claims = await resource.verify(token, dpop_request=ctx)
except AuthplaneError as error:
    status, headers = response_headers_for(
        error,
        realm="api.example.com",
        resource_metadata_url=resource.resource_metadata_url(),
    )
    return Response(status_code=status, headers=headers)

Catching SDK errors directly

If you call AuthplaneResource.verify() yourself (for example, in custom middleware or non-MCP code), the relevant exceptions to catch are the ones verify() actually raises:

from authplane import (
    AuthplaneError,
    DPoPError,
    InvalidClaimsError,
    InvalidSignatureError,
    TokenExpiredError,
    TokenRevokedError,
)

try:
    claims = await verifier.verify(token)
except TokenRevokedError:
    log.warning("Revoked token used")
    raise
except (TokenExpiredError, InvalidSignatureError, InvalidClaimsError):
    raise  # 401-class verification failures
except DPoPError:
    raise  # RFC 9449 binding/proof failures
except AuthplaneError:
    raise  # everything else from the verifier

InsufficientScopeError is not raised by verify(); it comes from claims.require_scope("…") after a successful verification.

API Reference

authplane_auth()

async def authplane_auth(
    issuer: str,
    base_url: str,
    scopes: list[str] | None = None,
    *,
    mcp_path: str = "/mcp",
    as_credentials: ASCredentials | None = None,
    dpop: DPoPProvider | None = None,
    allowed_algorithms: list[str] | None = None,
    jwks_refresh_seconds: int | None = None,
    metadata_refresh_seconds: int | None = None,
    cache_ttl_buffer_seconds: float | None = None,
    default_ttl_seconds: float | None = None,
    circuit_breaker_threshold: int | None = None,
    circuit_breaker_cooldown_seconds: float | None = None,
    clock_skew_seconds: int | None = None,
    dev_mode: bool | None = None,
    fetch_settings: FetchSettings | None = None,
    inbound_dpop: InboundDPoPOptions | None = None,
    revocation_checker: IntrospectionRevocation | RevocationChecker | None = None,
    fail_closed: bool = False,
    resource_metadata_url: str | None = None,
) -> AuthplaneAuthResult

Async factory that performs metadata discovery, fetches JWKS, and returns an AuthplaneAuthResult ready to unpack into FastMCP().

Raises:

  • ValueError — invalid configuration (e.g., HMAC algorithm)
  • JWKSFetchError — metadata discovery or JWKS fetch failed

AuthplaneAuthResult

Returned by authplane_auth(). Supports ** unpacking into FastMCP() — the mapping view yields only auth. token_verifier and client are exposed as plain attributes for advanced use cases. Call await result.aclose() on shutdown to release background tasks and HTTP connections.

Attribute Type Description
auth RemoteAuthProvider (a VerbatimPRMRemoteAuthProvider in practice) Auth provider for FastMCP. authplane_auth() always constructs the subclass — see below — but the attribute is typed as the base class, so a checker will not offer subclass members without a narrowing check
token_verifier AuthplaneTokenVerifier Token verifier (for advanced / manual setup)
client AuthplaneClient Underlying SDK client (use client.exchange() for RFC 8693)

VerbatimPRMRemoteAuthProvider

RemoteAuthProvider subclass that serves the Protected Resource Metadata identifiers byte-for-byte. Upstream builds the PRM from pydantic.AnyHttpUrl fields, which append a trailing slash to an empty-path authority; the core SDK compares identifiers verbatim, so a client following the advertised value literally is rejected.

authplane_auth() returns one already configured. Construct it directly only when you build the provider yourself — a documented FastMCP pattern — since using the base class instead loses the verbatim PRM silently.

Constructor argument Description
verbatim_issuer The issuer exactly as configured, not the AnyHttpUrl form
verbatim_resource The resource identifier exactly as configured

Everything else is forwarded to RemoteAuthProvider — note base_url is the server base, which is not the same value as verbatim_resource when the server is mounted under a path.

rewrite_prm_routes_verbatim(routes, *, issuer, resource)

The rewrite itself, exported for the case where you cannot subclass. Apply it to the route list your provider returns. It wraps the single route whose path is the RFC 9728 §3.1 derivation of resource, and emits a RuntimeWarning when routes exist under the well-known prefix but none is that derivation — meaning the rewrite did nothing.

AuthplaneTokenVerifier

FastMCP TokenVerifier implementation.

Method/Property Description
verify_token(token: str) -> AccessToken | None Validate JWT, return AccessToken or None
verifier (property) Access underlying AuthplaneResource
scopes_supported (property) Scopes configured in the verifier
resource_metadata_url() -> str URL to advertise as RFC 9728 §5.1 resource_metadata — the configured override, or the derivation

Core SDK types

The adapter does not re-export core SDK types. Import them from authplane (or authplane.oauth for token-operation types):

Type Import from
ASCredentials, FetchSettings, IntrospectionRevocation, RevocationChecker, DPoP types authplane
Verification errors (AuthplaneError, InsufficientScopeError, TokenExpiredError, TokenRevokedError, ConsentRequiredError, …) authplane
TokenExchangeOptions, TokenResponse authplane.oauth

Security Properties

The adapter enforces (via the core SDK):

  • RFC 9068 compliance — validates all 9 required JWT claims (iss, aud, sub, client_id, exp, nbf, iat, jti, typ)
  • Type header enforcement — only accepts typ: "at+jwt"
  • Asymmetric algorithms only — HMAC and none are rejected
  • SSRF protection — DNS pinning, IP blocklists, protocol allowlists, redirect blocking
  • Background JWKS refresh — refreshes at 80% of TTL to avoid request-time latency
  • Stale cache fallback — uses cached JWKS if a refresh fails, maintaining availability