Python
TL;DR — Three packages.
authplane-mcptargets the official MCP Python SDK (from mcp.server.fastmcp import FastMCP).authplane-fastmcptargets PrefectHQ FastMCP (from fastmcp import FastMCP) — different framework, same class name, different adapter.authplaneis the framework-agnostic core. All three require Python 3.11+ and share the same underlying JWT validation, DPoP, token-exchange, and elicitation semantics.
Install
# Official MCP Python SDK
pip install authplane-mcp
# PrefectHQ FastMCP
pip install authplane-fastmcp
# Core primitives only (custom integrations)
pip install authplane
Compatibility:
mcp >=1.28 renamed the elicitation field from elicitationId (camelCase) to elicitation_id (snake_case), which breaks authplane-mcp at the wire layer — a fix is planned, not yet cut. Track python-sdk issues if you’re on 1.28+.
authplane-mcp — Official MCP Python SDK adapter
Quickstart
import asyncio
from authplane import InboundDPoPOptions
from authplane_mcp import authplane_mcp_auth, install_request_context, require_scope
from mcp.server.fastmcp import FastMCP
async def main() -> None:
auth = await authplane_mcp_auth(
issuer="https://auth.example.com",
resource="https://mcp.example.com/mcp",
scopes=["tools/query"],
enforce_scopes_on_all_requests=True,
inbound_dpop=InboundDPoPOptions(required=True),
)
mcp = FastMCP("my-server", port=8080, json_response=True, **auth)
install_request_context(mcp) # required for inbound DPoP
@mcp.tool()
async def query(sql: str) -> str:
require_scope("tools/query")
return f"Ran: {sql}"
try:
await mcp.run_streamable_http_async()
finally:
await auth.aclose()
asyncio.run(main())
How the return value plugs into FastMCP: authplane_mcp_auth() returns an AuthplaneAuthResult. It implements the mapping protocol — **auth yields exactly the two keys FastMCP() expects (token_verifier, auth). .client is exposed as a plain attribute for client.exchange() calls from your tool code. .aclose() releases the JWKS refresh task + httpx pool.
authplane_mcp_auth() options
Same shape as authplane-fastmcp and authplaneMcpAuth in TS — see full reference for every option with defaults. The most-used:
Scope enforcement
Two layers, use both:
Request-layer (enforce_scopes_on_all_requests=True) — the MCP SDK’s RequireAuthMiddleware rejects any request whose token doesn’t carry every scope in scopes=[...]. Also required for PRM scopes_supported to be populated (OAuth-discovery clients like Claude Code request scopes based on this).
Per-tool (require_scope("scope") inside each handler) — granular. Remains correct even when request-layer enforcement is on (no-op in that case).
from authplane_mcp import require_scope
@mcp.tool()
async def query(sql: str) -> str:
require_scope("tools/query") # raises PermissionError if scope missing
return f"Ran: {sql}"
Inbound DPoP
To enforce DPoP proofs on inbound requests you need three things:
inbound_dpop=InboundDPoPOptions(required=True)onauthplane_mcp_auth().install_request_context(mcp)after buildingFastMCP.- The AS must issue DPoP-bound tokens (
AUTHPLANE_DPOP_ENABLED=true).
Why install_request_context? The official MCP Python SDK’s FastMCP doesn’t expose a middleware seam. The adapter wraps mcp.streamable_http_app and installs AuthplaneRequestContextMiddleware that publishes the active Starlette Request on a ContextVar before the verifier runs — the verifier reads it to build a DPoPRequestContext and forward it to AuthplaneResource.verify(). Without this call, DPoP-bound requests fail closed with 401 WWW-Authenticate: DPoP error="invalid_dpop_proof" — the misconfiguration surfaces immediately rather than as a silent bypass.
The call is idempotent — a second call on the same FastMCP instance is a no-op.
Token exchange (RFC 8693)
Vend upstream tokens or delegate on behalf of the user from inside a tool handler:
from authplane.oauth import TokenExchangeOptions
from mcp.server.auth.middleware.auth_context import get_access_token
@mcp.tool()
async def read_calendar() -> dict[str, str]:
require_scope("tools/read_calendar")
inbound = get_access_token()
downstream = await auth.client.exchange(TokenExchangeOptions(
subject_token=inbound.token,
resources=("https://www.googleapis.com/calendar/v3",),
scope="https://www.googleapis.com/auth/calendar",
))
return {"token_type": downstream.token_type, "expires_in": str(downstream.expires_in)}
The result includes access_token, token_type ("Bearer" or "DPoP" if the subject was DPoP-bound), expires_in, and scope.
URL elicitation for consent
If the exchange target requires the user to complete an upstream OAuth Connect flow (they haven’t yet linked Google Calendar), the AS returns error=consent_required with a consent_url. The adapter wraps client.exchange to translate this into UrlElicitationRequiredError — MCP JSON-RPC error code -32042. The MCP SDK re-raises it from tool handlers, the client sees the elicitation, and the user visits the URL. No try/except needed in your tool code.
Cleanup
auth.aclose() cancels the JWKS refresh task and closes the httpx pool. Always wrap mcp.run_streamable_http_async() in try/finally.
authplane-fastmcp — PrefectHQ FastMCP adapter
Different framework, similar-shaped API. Uses from fastmcp import FastMCP (PrefectHQ), not the official MCP Python SDK.
Quickstart
import asyncio
from fastmcp import FastMCP
from authplane_fastmcp import authplane_auth
async def main() -> None:
auth = await authplane_auth(
issuer="https://auth.example.com",
base_url="https://mcp.example.com",
scopes=["tools/query"],
)
mcp = FastMCP("my-server", **auth)
@mcp.tool()
def query(sql: str) -> str:
"""Execute a query."""
return f"Ran: {sql}"
try:
await mcp.run_async(transport="http", port=8080)
finally:
await auth.aclose()
asyncio.run(main())
The audience URL is auto-derived from base_url + mcp_path (mcp_path defaults to /mcp). You can also pass an explicit resource= if your setup diverges.
authplane_auth() options
Same shape as authplane_mcp_auth() with two differences:
base_url+mcp_pathinstead ofresource(or passresourceexplicitly).- No
enforce_scopes_on_all_requests— PrefectHQ FastMCP filters tools from the catalog per-scope instead (see below). - Same
inbound_dpop,revocation_checker,as_credentials,dpop,dev_mode,allowed_algorithms, cache tunables.
Scope enforcement — different from authplane-mcp
PrefectHQ FastMCP uses require_scopes from fastmcp.server.auth as a decorator argument on @mcp.tool:
from fastmcp.server.auth import require_scopes
@mcp.tool(auth=require_scopes("tools/query"))
def query(sql: str) -> str:
"""Requires tools/query scope."""
return f"Ran: {sql}"
@mcp.tool(auth=require_scopes("tools/admin", "tools/delete"))
def delete_all() -> str:
"""Requires BOTH scopes."""
return clear_database()
Enforcement happens by filtering tools out of the catalog, not by returning 403 on call. A caller missing tools/query sees tools/list without query present, and a tools/call for query returns HTTP 200 with {"isError": true, "content": [{"text": "Unknown tool: 'query'"}]} — not a 403. UX layers expecting 403 to prompt re-auth won’t see one; key off isError + tool-not-found content text instead.
Inbound DPoP
Setting inbound_dpop=InboundDPoPOptions(required=True) on authplane_auth() fully enforces DPoP proof validation on every incoming request — the verifier checks the proof, binding, and replay cache before your tool handler runs. No additional middleware is needed.
No extra wiring is needed: the fastmcp verifier reads the raw request via fastmcp.server.dependencies.get_http_request() when it builds the DPoPRequestContext. (install_request_context(mcp) belongs to authplane-mcp, not this package.)
The AS must issue DPoP-bound tokens (AUTHPLANE_DPOP_ENABLED=true) for enforcement to have anything to check.
Token exchange, elicitation, cleanup
Same shape as authplane-mcp above — auth.client.exchange(TokenExchangeOptions(...)) from inside a tool, elicitation auto-translation to -32042, await auth.aclose() in finally.
authplane — Core primitives
Framework-agnostic package. Use it when:
- You’re writing a custom transport or framework integration.
- You want to verify tokens outside a request context (batch job, background worker).
- You need programmatic PRM generation without the adapter’s Starlette wiring.
- You want direct control over the httpx client / DPoP outbound / caching.
Building an AuthplaneClient
from authplane import AuthplaneClient, ASCredentials
client = await AuthplaneClient.create(
issuer="https://auth.example.com",
auth=ASCredentials(client_id="my-tool", client_secret="…"),
dev_mode=False,
jwks_refresh_seconds=300,
)
# Later, at shutdown:
await client.aclose()
Building a resource + verifying a token
verifier = client.resource(
resource="https://mcp.example.com/mcp",
scopes=["tools/query"],
)
claims = await verifier.verify(access_token, dpop_request=None)
# claims.sub, claims.scopes, claims.audience, claims.dpop_proof, claims.act, claims.agent_id, ...
AuthplaneResource.verify() returns VerifiedClaims with all standard JWT claims plus AuthPlane extensions (agent_id, act chain, dpop_proof for DPoP-bound tokens).
Token exchange from the core client
from authplane.oauth import TokenExchangeOptions
response = await client.exchange(TokenExchangeOptions(
subject_token=inbound_jwt,
resources=("https://api.example.com/downstream",),
scope="tools/write",
))
# response.access_token, response.token_type, response.expires_in
Inbound DPoP
Build a DPoPRequestContext from your framework’s request object and pass it to verify(dpop_request=ctx). The DPoPRequestContext wraps method + URL + DPoP header value(s). See core user-guide §8 Inbound DPoP Summary for the shape.
Error handling
Every failure mode is a typed exception in authplane.errors:
TokenMissingError— noAuthorizationheader, or unparseableTokenExpiredError/InvalidSignatureError/InvalidClaimsError— expiry, signature, or claim validation failedInsufficientScopeError— token lacks required scope(s)TokenRevokedError— introspection returnedactive: falseDPoPBindingMismatchError/DPoPReplayDetectedError— DPoP failuresConsentRequiredError— token exchange needs upstream Connect (containsconsent_url)
http_status and www_authenticate are module-level functions in authplane.errors that take an error instance and produce the correct RFC 6750 / 9449 challenge for your response:
from authplane.errors import http_status, www_authenticate
try:
claims = await verifier.verify(access_token, dpop_request=None)
except Exception as err:
status = http_status(err)
challenge = www_authenticate(err)
# return an HTTP response with `status` and a `WWW-Authenticate: <challenge>` header
Related
- SDKs overview — pick between adapters
- Quickstart —
authplane-mcpend-to-end in 10 minutes - Your first MCP server — line-by-line walkthrough
- Guides: Enable DPoP end-to-end — configuring inbound DPoP correctly
- Guides: Wire up the Token Vault — using
client.exchange()for upstream vending - Concepts: DPoP — why proof-of-possession, when to enable
- Full user-guides in the python-sdk repo: