RFC: Auth utility for JWT verification and OAuth2 client credentials

[bfreiberg]

Is this related to an existing feature request or issue?

No response

Which Powertools for AWS Lambda (Python) utility does this relate to?

Other

Summary

A small utility with two objects and one job each:

aws_lambda_powertools.utilities.auth
├── JWTVerifier     # inbound: JWT access token → verified claims
└── OAuth2Client    # outbound: client credentials → cached bearer token for a configured API

JWTVerifier combines JWT verification with an explicit claim-validation policy, bounded JWKS caching, coordinated key refresh and Lambda integrations. It plugs into Event Handler as middleware and into API Gateway as a Lambda authorizer.

OAuth2Client handles token acquisition, caching and reacquisition for a Lambda function calling an OAuth2-protected API.

Both use provider-neutral interfaces with documented compatibility requirements. Cognito has a factory for resource-bound access tokens.

Protected HTTP MCP servers running on Lambda are a first-class consumer. They act as OAuth resource servers, and the MCP Python SDK accepts a TokenVerifier adapter. JWTVerifier.verify() supplies local verification when the authorization server issues compatible JWT access tokens. MCP authorization is optional, and access tokens need not be JWTs; opaque-token introspection is future work.[^mcp-auth]

All aws_lambda_powertools.utilities.auth APIs and the [auth] extra below are proposed, not currently available APIs.

Use case

1. Validating tokens from a non-Cognito IdP

Applications that validate third-party bearer tokens inside Lambda need to connect a JWT library to their request handling and authorization policy. This includes custom Lambda authorizers for REST APIs and functions that perform their own verification behind an ALB or Function URL.

PyJWT already provides JWKS retrieval and caching through PyJWKClient. A basic implementation using existing APIs looks like this:[^pyjwt-usage]

import jwt

jwk_client = jwt.PyJWKClient(
    "https://my-tenant.auth0.com/.well-known/jwks.json",
    cache_jwk_set=True,
    lifespan=300,
    cache_keys=False,
    timeout=3,
)

def verify_access_token(token: str) -> dict:
    signing_key = jwk_client.get_signing_key_from_jwt(token)
    return jwt.decode(
        token,
        signing_key.key,
        algorithms=["RS256"],
        audience="https://api.example.com",
        issuer="https://my-tenant.auth0.com/",
        options={"require": ["iss", "aud", "exp"]},
    )

The remaining work includes provider-specific token-purpose checks, refresh coordination and rate limiting, consistent errors, scope enforcement, middleware and authorizer responses. Those integrations and verification defaults are the value of this proposal.

2. MCP servers on Lambda

For a protected HTTP MCP server using JWT access tokens, the integration needs to:

1. Verify the signature and require valid iss, aud and exp claims.
2. Check that the audience includes the configured MCP resource identifier.
3. Enforce the permissions required by each operation.
4. Use separately obtained authorization appropriate to each downstream API, without forwarding the incoming bearer token.[^mcp-auth]

The SDK supplies transport and authorization metadata handling. A small adapter converts verified claims into its AccessToken model. The proposed adapter uses local JWT verification; the referenced tutorial demonstrates introspection, which is a different verification approach.[^mcp-tutorial]

3. Calling an OAuth2-protected API

A Lambda calling an OAuth2-protected API needs to select the target resource, retrieve its client secret, exchange credentials for a token, and cache that token until it needs replacing. The cache must keep tokens for different APIs separate even when those APIs use the same scope names.

The proposed client centralizes this behavior while allowing secret retrieval through Parameters and HTTP calls through the application's existing client.

Proposal

JWTVerifier

from aws_lambda_powertools.utilities.auth import JWTVerifier

verifier = JWTVerifier(
    issuer="https://my-tenant.auth0.com/",
    audience="https://api.example.com",
    algorithms=["RS256"],
)

claims = verifier.verify(token)   # dict; raises InvalidTokenError or JWKSFetchError

Parameter	Required	Notes
issuer	yes	Must be https://. Compared exactly against iss.
audience	yes	Nonempty str or list[str] of resource identifiers. At least one must exactly match an entry in the token's string or string-array aud.
algorithms	yes	Explicit allowlist. none and HMAC are always rejected for JWKS-based verification.
jwks_uri	no	Explicit HTTPS JWKS endpoint; skips OIDC discovery. Mutually exclusive with jwks.
jwks	no	Static JWKS dict. Verification and prefetch make no network calls.
clock_skew_seconds	no	Default 60; nonnegative tolerance for time validation. Set to 0 for no leeway.
required_claims	no	Additional required claims, e.g. ["sub", "scope"]. Cannot remove the baseline iss, aud, exp.
jwks_max_age_seconds	no	Default 300; positive maximum age for a remotely fetched key set.
unknown_kid_cooldown_seconds	no	Default 300; minimum interval between refresh attempts triggered by unknown key IDs. Separate from maximum age.
timeout_seconds	no	Default 3; finite time budget for discovery/key retrieval during one verification or prefetch operation.

Verification and discovery

Every verification requires iss, aud and exp to be present and valid. Expiration validation remains enabled, and nbf is validated when present. required_claims is additive: supplying ["sub"] requires all four claims. Requiring a claim's presence does not by itself validate its meaning.[^pyjwt-api]

The algorithm allowlist comes from trusted configuration. Keys must be compatible with the allowed signing algorithm. The generic verifier provides these checks; applications must also apply their issuer's access-token profile. The Cognito factory below includes its provider-specific token-purpose checks.

Without static keys or an explicit endpoint, use OIDC discovery for the configured issuer, require the metadata issuer to match exactly, and obtain its HTTPS jwks_uri. An issuer without discovery must use an explicit jwks_uri or a provider factory. Do not guess a fallback endpoint after a discovery failure or fetch URLs supplied by the token.

JWKS caching and rotation

Cache remote JWKS in memory for the execution environment, with the following contract:

• Check key-set age before using any key, including an already-known kid. Once the maximum age is reached, verification requires a successful refresh.
• A successful refresh atomically replaces the set and invalidates removed or changed keys in every derived key cache. A per-key cache must never extend trust beyond its parent key set.
• If refresh fails, do not use keys past their maximum age. Raise JWKSFetchError when the required keys cannot be obtained. Still-fresh keys may continue to serve matching tokens until their age limit.
• An unknown kid can trigger one refresh if its cooldown permits; otherwise reject it immediately. Recheck once after a permitted refresh and reject if the key remains unknown. An empty-cache or age-triggered fetch also satisfies that attempt, so one verification cannot trigger a second miss-driven fetch.
• The unknown-key cooldown never authorizes stale-key use or prevents an age-required refresh. Coordinate concurrent refreshes so callers share one in-flight fetch per configured issuer and key source; bound retries and use backoff after failures.
• prefetch() populates the same cache during initialization. It does not eliminate later refreshes or reset key age without successful retrieval.

The proposed 300-second values are policy defaults, not protocol requirements. They bound trust in keys removed by an issuer and limit requests caused by random key IDs. They also introduce an availability tradeoff: a legitimate new key can be rejected during the miss cooldown, and an issuer outage can prevent verification when cached keys expire.

For static jwks, there is no remote refresh or age-based revalidation. The operator must replace the configured keys and recreate affected verifier instances when rotating or removing a key. This mode deliberately transfers key-lifecycle responsibility to the application.

Factories

# Cognito access-token profile: app client and API audience are separate.
verifier = JWTVerifier.cognito(
    user_pool_id="us-east-1_abc123",
    client_id="my-app-client",
    audience="https://api.example.com",
)

# Multiple explicitly configured issuers.
verifier = JWTVerifier.any_of(
    verifier,
    JWTVerifier(
        issuer="https://login.microsoftonline.com/<tenant>/v2.0",
        audience="api://orders",
        algorithms=["RS256"],
    ),
)

cognito() fixes the algorithm to RS256, derives the pool's issuer and JWKS endpoint, and requires token_use == "access", matching client_id, and a matching resource aud, in addition to the baseline claims. This factory does not offer an ID-token mode.

Cognito ID tokens use the app client as their audience; access tokens identify the app client in client_id and include resource aud only when resource binding is requested. An ID token or an access token without resource aud is rejected by this profile.[^cognito-access]

To obtain a compatible user access token, configure the app client's allowed scopes and use Cognito's authorization-code flow with resource=https://api.example.com in the authorization request, then exchange the code. The resulting access token must contain that URL in aud; configuring custom scopes alone does not provide this binding. For an MCP client, use its configured MCP resource URL. The verifier consumes the resulting token; implementing the browser flow remains outside this utility.[^cognito-resource]

any_of() uses unverified iss only to select from its configured verifiers, then performs full verification with that issuer's keys and policy. Unknown issuers are rejected without discovery, and duplicate issuer configurations are rejected as ambiguous. Keys are never tried across providers.

Event Handler middleware

from aws_lambda_powertools.event_handler import APIGatewayHttpResolver
from aws_lambda_powertools.utilities.auth import JWTVerifier

verifier = JWTVerifier(
    issuer="https://my-tenant.auth0.com/",
    audience="https://api.example.com",
    algorithms=["RS256"],
    required_claims=["sub"],
)
app = APIGatewayHttpResolver()

@app.get("/orders", middlewares=[verifier.require(scopes=["orders:read"])])
def list_orders():
    claims = app.context["claims"]
    return {"user": claims["sub"]}

@app.get("/health")   # no middleware = public
def health():
    return {"status": "ok"}

Behavior:

• Reads Authorization: Bearer <token> (case-insensitive scheme)
• Missing token → 401 with WWW-Authenticate: Bearer; invalid or expired token → 401 with WWW-Authenticate: Bearer error="invalid_token". Both return {"message": "Unauthorized"}.
• Valid token but missing scope → 403 with WWW-Authenticate: Bearer error="insufficient_scope", scope="orders:read"
• On success, claims are stored in app.context["claims"] (cleared per invocation by Event Handler)
• Failure to obtain required verification keys → a generic 503; never execute the protected handler.
• Never leaks kid, JWKS URL or the underlying exception in the response

scopes=[...] requires every listed scope. Read the first present claim in the order scope, scp, scopes, accepting a space-separated string or a list of strings. A malformed selected claim is invalid; do not merge claims or fall back to another claim after a format error. Missing scopes grant no permissions.

Customization lives on require(), not on the constructor, so different routes can behave differently:

from aws_lambda_powertools.event_handler import Response

@app.get("/admin", middlewares=[
    verifier.require(
        scopes=["admin"],
        authorize=lambda claims: claims.get("org_id") == "acme",   # extra check → 403 if False
        on_error=lambda err: Response(
            status_code=err.status_code,
            content_type="application/json",
            body={"error": "access_denied"},
            headers=err.headers,
        ),
    )
])
def admin(): ...

The proposed error callback receives the mapped HTTP status and challenge headers so customization can preserve the 401/403/503 distinction. An error callback cannot turn failed verification into execution of the protected handler.

Lambda Authorizer

Connect verification to the existing authorizer Data Classes and response builders:

from aws_lambda_powertools.utilities.auth import JWTVerifier
from aws_lambda_powertools.utilities.data_classes import event_source
from aws_lambda_powertools.utilities.data_classes.api_gateway_authorizer_event import APIGatewayAuthorizerRequestEvent

verifier = JWTVerifier(
    issuer="https://my-tenant.auth0.com/",
    audience="https://api.example.com",
    algorithms=["RS256"],
    required_claims=["sub"],
)

@event_source(data_class=APIGatewayAuthorizerRequestEvent)
def handler(event: APIGatewayAuthorizerRequestEvent, context):
    return verifier.authorize(
        event,
        scopes=["orders:read"],
        response_format="iam",
        context_claims=["sub"],
    )

authorize() supports REST TOKEN and REQUEST events and HTTP API authorizer events. It returns a serialized response dictionary, using existing Data Classes where applicable:

Response format	Behavior
response_format="iam" (default)	Allow or Deny policy restricted to the current method/route ARN. A successful result requires a nonempty string sub as its principal.
response_format="simple"	HTTP API payload version 2.0 only: {"isAuthorized": true/false}. Deployment must enable simple responses.

The event identifies its payload format but does not reveal whether the deployment enabled simple responses; select the response format explicitly. context_claims defaults to an empty list and copies only selected scalar claims on success. Do not copy the entire JWT payload or token into Gateway context.[^gateway-output]

Missing/invalid tokens and insufficient permissions return a denial in the selected format. An infrastructure failure such as JWKSFetchError raises a sanitized error and never produces an allow. Gateway determines the resulting HTTP response; these policy/Boolean responses do not provide Event Handler's custom status or challenge headers.

API Gateway authorizer-result caching

Initial deployment examples disable authorizer-result caching for both REST and HTTP APIs. This cache is separate from the in-process JWKS cache: a Gateway cache hit bypasses the authorizer function and its verification and scope checks.[^gateway-http]

For example, these properties belong inside an HTTP API OpenAPI x-amazon-apigateway-authorizer definition; this is a configuration fragment:

authorizerResultTtlInSeconds: 0
identitySource: "$request.header.Authorization"

If caching is enabled later, its cache key and policy must cover every input used to authorize the request:

• For HTTP APIs with route-specific decisions, include $context.routeKey alongside the bearer token in the identity sources. Otherwise a cached simple-response allow for GET /orders can also authorize DELETE /orders/{id} without checking its write scope.
• Route separation does not cover decisions based on a particular object, tenant or other request attribute. Those require additional cache-key design or disabled caching.
• IAM policies restricted to one method can cause implicit denials on other methods when reused. Broad wildcard policies can overgrant. Do not broaden the policy solely to improve cache reuse.
• A cached allow can outlive the token. A token accepted one second before its validation deadline can leave a usable cached decision for nearly the full Gateway TTL. Returning exp in context does not set a per-token TTL. Keep caching disabled when every request must respect the token's expiry and configured clock tolerance, unless the application independently revalidates.[^gateway-http]

For HTTP API route separation, the OpenAPI identity-source setting would be:

identitySource: "$request.header.Authorization, $context.routeKey"

This separates routes only; it does not solve the expiration or request-attribute concerns above.

authorize() cannot change these deployment settings by returning a different dictionary. Documentation must show the corresponding REST and HTTP API configuration separately.

OAuth2Client

import httpx

from aws_lambda_powertools.utilities import parameters
from aws_lambda_powertools.utilities.auth import OAuth2Client

orders_api = OAuth2Client(
    token_url="https://auth.example.com/oauth/token",
    client_id="orders-service",
    client_secret=lambda: parameters.get_secret("orders/oauth-secret", max_age=300),
    scopes=["orders:read", "orders:write"],
    audience="https://api.example.com",   # Provider-specific token request parameter
    timeout_seconds=3,
)

# Option 1: headers only — use with any HTTP client
headers = orders_api.auth_headers()   # {"Authorization": "Bearer eyJ..."}
httpx.get("https://api.example.com/orders", headers=headers, timeout=5.0)

# Option 2: let it send
response = orders_api.request("GET", "https://api.example.com/orders", timeout=5.0)
response.json()

Resource selection

Expose two optional, mutually exclusive constructor parameters:

Parameter	Token request	Use
audience: str	audience=<value>	Provider-specific selection, such as an Auth0 API identifier.
resource: str	resource=<value>	A single resource indicator for endpoints supporting RFC 8707.

These parameters are not aliases and are not sent interchangeably. Auth0 requires an API audience for this flow; other providers may use resource or derive the target from scope/configuration conventions. If both parameters are omitted, documentation must explain the provider's resource-selection convention. Scopes alone do not universally identify an API.[^auth0-credentials][^resource-indicators]

Each instance has immutable token-request configuration and a private cache, bound to its endpoint, client, scopes, resource selection and authentication method. No per-call resource override or shared token cache is proposed. Use separate clients for separate resources. The eventual HTTP request URL does not change the token's audience.

Token exchange and cache behavior

• Support the client_credentials grant only. Refresh-ahead means another client-credentials exchange, not use of an OAuth refresh token.[^oauth]
• Cache the token in the execution environment and reacquire it on demand when fewer than 30 seconds of its advertised lifetime remain. For tokens with a lifetime of 30 seconds or less, return the newly obtained token without caching it; do not loop trying to refresh it. Never return a token whose advertised lifetime has already elapsed, including during the exchange.
• Require a nonempty access token, bearer token type and a positive expires_in before caching. If expires_in is omitted, return the token without caching; reject malformed values. Track lifetime conservatively from request start with a monotonic clock.
• client_secret accepts str | Callable[[], str]. Invoke the callable for each exchange attempt and retain no separate cached secret. A Parameters provider may have its own cache: in the example, a changed secret can remain unseen for up to max_age=300. An existing access token remains reusable until its own refresh boundary.
• Phase 2 supports client_secret_basic only. Form-encode the client identifier and secret before constructing the HTTP Basic credentials, as RFC 6749 specifies; do not send them in the request body. Providers or clients requiring client_secret_post, private-key JWT or mTLS authentication are outside this first client implementation.[^oauth]
• Require HTTPS and finite network timeouts. timeout_seconds defaults to 3 for the complete token acquisition operation, including retries. Allow at most two retries with backoff for transient exchange failures within that budget; do not retry invalid-client or invalid-scope responses.
• Coordinate concurrent cache misses or reacquisitions so they share one exchange. Do not expose tokens, secrets or authorization headers through exceptions or __repr__.
• auth_headers() returns a header dictionary for an application-owned HTTP client. request() is a synchronous convenience method returning an HTTP response with .json(); it requires HTTPS, does not follow redirects or automatically retry downstream requests, and rejects a competing Authorization header. Its timeout governs the downstream HTTP operation separately from token acquisition. Pass only trusted destination URLs.

MCP examples

MCP does not appear in the API surface. These are documentation examples showing what verify() and OAuth2Client enable.

Plugging into the MCP Python SDK

The adapter below follows the MCPServer API used by the referenced tutorial, which declares mcp>=2.0.0rc1. It maps a Keycloak-style access token with azp, sub and a space-separated scope; other providers need their own claim mapping.[^mcp-tutorial]

import asyncio

from pydantic import AnyHttpUrl
from mcp.server import MCPServer
from mcp.server.auth.provider import AccessToken, TokenVerifier
from mcp.server.auth.settings import AuthSettings
from aws_lambda_powertools.utilities.auth import JWTVerifier
from aws_lambda_powertools.utilities.auth.exceptions import InvalidTokenError, JWKSFetchError

RESOURCE_URL = "https://mcp.example.com"
ISSUER_URL = "https://keycloak.example.com/realms/mcp"

verifier = JWTVerifier(
    issuer=ISSUER_URL,
    audience=RESOURCE_URL,
    algorithms=["RS256"],
    required_claims=["azp", "sub", "scope"],
    timeout_seconds=3,
)


class PowertoolsTokenVerifier(TokenVerifier):
    async def verify_token(self, token: str) -> AccessToken | None:
        try:
            claims = await asyncio.to_thread(verifier.verify, token)
        except (InvalidTokenError, JWKSFetchError):
            # This adapter rejects authentication if verification cannot complete.
            return None
        if not all(isinstance(claims[name], str) for name in ("azp", "sub", "scope")):
            return None
        if not claims["azp"] or not claims["sub"]:
            return None
        return AccessToken(
            token=token,
            client_id=claims["azp"],
            subject=claims["sub"],
            scopes=claims["scope"].split(),
            expires_at=claims["exp"],
            resource=RESOURCE_URL,
        )


mcp = MCPServer(
    name="orders",
    token_verifier=PowertoolsTokenVerifier(),
    auth=AuthSettings(
        issuer_url=AnyHttpUrl(ISSUER_URL),
        resource_server_url=AnyHttpUrl(RESOURCE_URL),
        required_scopes=["mcp:tools"],
    ),
)

The SDK publishes Protected Resource Metadata and handles its authentication challenges. Powertools verifies the JWT signature, required claims and key freshness. This adapter returns None for both invalid tokens and unavailable keys, so both fail authentication; applications needing a distinct availability response must integrate that mapping at the SDK transport boundary.[^mcp-tutorial]

asyncio.to_thread() keeps synchronous discovery and JWKS I/O off the event loop. Prefetch only addresses initialization: unknown keys, expired key sets and failed-fetch retries can still cause network I/O later. Phase 1 therefore requires thread-safe caches, coordinated refreshes and finite timeouts. Cancelling the awaiting task does not terminate an already-running network operation.[^asyncio]

Scope per tool

Tools can enforce additional permissions using the scopes mapped into the SDK access token:

from mcp.server.auth.middleware.auth_context import get_access_token

def require_scope(scope: str):
    token = get_access_token()
    if token is None or scope not in token.scopes:
        raise PermissionError(f"scope '{scope}' required")

@mcp.tool()
def list_orders(customer_id: str) -> list[dict]:
    require_scope("orders:read")
    ...

@mcp.tool()
def cancel_order(order_id: str) -> dict:
    require_scope("orders:write")
    ...

This illustrates a tool-level authorization check. Raising PermissionError does not itself implement an HTTP 403 challenge or an MCP scope-upgrade flow; production examples must use the SDK's supported error handling for the targeted version.

Corporate IdP plus Cognito during migration

An MCP server used by both employees (Entra ID) and an internal app (Cognito) while the org consolidates identity:

verifier = JWTVerifier.any_of(
    JWTVerifier(
        issuer="https://login.microsoftonline.com/<tenant>/v2.0",
        audience="https://mcp.example.com",
        algorithms=["RS256"],
    ),
    JWTVerifier.cognito(
        user_pool_id="us-east-1_abc123",
        client_id="mcp-internal",
        audience="https://mcp.example.com",
    ),
)

Both issuers must issue access tokens for the configured resource audience; the Cognito client must request resource binding. This shows verifier composition only. The SDK adapter must map each issuer's verified client and scope claims, and the deployment must advertise the appropriate authorization servers. The Keycloak-specific adapter above cannot be reused unchanged.

Calling a downstream API without token passthrough

The MCP spec forbids forwarding the incoming bearer token to another API. This example obtains a separate Inventory token with the MCP server's client credentials. A separately obtained delegated token is another valid design; token exchange is outside this RFC.[^mcp-auth]

from urllib.parse import quote

from aws_lambda_powertools import Logger
from aws_lambda_powertools.utilities import parameters
from aws_lambda_powertools.utilities.auth import OAuth2Client

logger = Logger()

inventory_api = OAuth2Client(
    token_url="https://auth.example.com/oauth/token",
    client_id="orders-mcp",
    client_secret=lambda: parameters.get_secret("orders-mcp/secret"),
    scopes=["inventory:read"],
    audience="https://inventory.example.com",
)

@mcp.tool()
async def check_stock(sku: str) -> dict:
    require_scope("inventory:read")
    caller = get_access_token()
    response = await asyncio.to_thread(
        inventory_api.request,
        "GET",
        f"https://inventory.example.com/stock/{quote(sku, safe='')}",
        timeout=5.0,
    )
    logger.info("stock lookup", extra={"subject": caller.subject, "sku": sku})
    return response.json()

API Gateway in front of an MCP server

An API Gateway Lambda authorizer can reject requests before invoking the MCP Lambda. Reuse verifier configuration in each function; separate execution environments have separate caches.

For a REST REQUEST authorizer, using a verifier configured for the MCP audience and requiring sub:

@event_source(data_class=APIGatewayAuthorizerRequestEvent)
def authorizer(event: APIGatewayAuthorizerRequestEvent, context):
    return verifier.authorize(
        event,
        scopes=["mcp:tools"],
        response_format="iam",
        context_claims=["sub"],
    )

Use authorizer-result TTL zero as described above. The full MCP deployment must also expose its metadata endpoint and preserve the authentication discovery/challenge behavior clients need. A Lambda authorizer denial alone does not provide that MCP integration.

Pinned keys for a private MCP server

An internal MCP server whose IdP is not reachable from the Lambda VPC, or where the team wants no network calls at verification time:

import json

from aws_lambda_powertools.utilities import parameters
from aws_lambda_powertools.utilities.auth import JWTVerifier

verifier = JWTVerifier(
    issuer="https://idp.internal",
    audience="https://mcp.internal",
    algorithms=["ES256"],
    jwks=json.loads(parameters.get_parameter("/mcp/jwks", max_age=3600)),
)

The parameter is read once when this verifier is constructed. Parameters' max_age does not refresh the verifier's static key snapshot. Updating the parameter requires recreating affected verifiers or recycling their execution environments.

Errors

from aws_lambda_powertools.utilities.auth.exceptions import (
    AuthError,               # base
    InvalidTokenError,       # base for token-verification failures
    TokenExpiredError,
    InvalidSignatureError,
    InvalidClaimsError,      # wrong/missing claim, invalid time claim, wrong token purpose
    JWKSFetchError,          # AuthError, separate from InvalidTokenError
    TokenExchangeError,      # OAuth2Client: token endpoint returned an error (redacted)
)

Exceptions and utility logs never expose tokens, secrets, authorization headers or token-endpoint response bodies, including through chained exceptions. Use stable reason codes for diagnostics; redact credentials and sensitive query values from any logged URL. Application responses do not expose key IDs, endpoint URLs or underlying library exceptions.

Testing support

from aws_lambda_powertools.utilities.auth.testing import mock_claims

def test_list_orders():
    with mock_claims(verifier, {"sub": "user-123", "scope": "orders:read"}):
        response = app.resolve(event, context)
    assert response["statusCode"] == 200

Patches verify() to return the given claims with no network call. This deliberately bypasses cryptographic and claim validation for application tests; it does not replace tests of verification, key rotation or authorization policy.

Phasing

Phase 1 — JWTVerifier: verify(), require(), authorize(), the strict Cognito access-token factory, any_of(), prefetch(), static jwks, bounded and coordinated key caching, testing helper, and inbound MCP documentation using the thread adapter. Include deployment examples with Gateway result caching disabled.

Phase 2 — OAuth2Client: auth_headers(), request(), typed resource selection, separate per-instance token caches, callable secret and reacquisition before expiry. Add the downstream MCP example in this phase.

Future, not committed — Token introspection (RFC 7662) for opaque access tokens, delegated token exchange (RFC 8693), additional client-authentication methods, and native async verification/token acquisition. Native async support needs its own HTTP-client lifecycle and concurrency design.

Acceptance criteria

• Reject missing baseline claims, expired tokens beyond configured leeway, invalid nbf, wrong issuer/audience/signature/algorithm, and unknown issuers without network discovery. Additional required claims preserve the baseline.
• Cognito accepts only access tokens matching both client and resource identifiers. Reject ID tokens, absent resource aud, and wrong token_use or client_id.
• Exercise known and unknown keys across cache expiry, successful removal/replacement, refresh failure and concurrent requests. After a successful refresh, no derived cache can retain removed keys; no cache may serve expired key sets.
• Verify all supported authorizer event/response combinations, scalar context handling, and denial behavior. Deployment documentation covers TTL zero, route-aware caching limits, IAM policy reuse and cached decisions outliving tokens.
• Exercise the async adapter during an actual key fetch and refresh: the event loop remains responsive, refreshes are coordinated, and network work has finite timeouts.
• For the outbound client, test audience and resource request encoding, mutual exclusion, isolation between resources, short/missing token lifetimes, concurrent reacquisition, secret rotation and retry limits.
• Verify that error messages, logs and object representations do not expose tokens or credentials.

Out of scope

• MCP server or resolver. SDK transport, metadata publication and protocol-level authorization responses remain with the MCP SDK and deployment.
• SigV4 signing. Separate from JWT verification and OAuth token acquisition.
• Interactive OAuth grants. This utility consumes inbound access tokens and performs outbound client-credentials exchanges; it does not implement browser redirects, login sessions or authorization-code exchange.
• Individual-token revocation checks. Local signature and claim verification does not check revocation. JWKS removal affects trust in a signing key, not one individual token. Use suitable token lifetimes or a separate revocation/introspection mechanism when required.
• Symmetric (HMAC) JWTs. This proposal supports asymmetric verification.
• Disabling resource-audience checks. No verify_aud=False, Cognito ID-token profile, or Cognito access-token profile without aud is included. Supporting the latter later would require a separately named profile with a different resource-binding guarantee.

Potential challenges

• Packaging. Ship cryptographic and HTTP dependencies through the optional [auth] extra, without adding requirements to the base Powertools installation. Measure package size and cold-start impact across supported Lambda architectures; do not assume runtime-bundled transitive dependencies are part of this utility's contract.
• Discovery and refresh latency. Initial discovery can require multiple HTTPS requests. Prefetch shifts initial work to initialization; subsequent expiry and rotation still require network access. Document the timeout and key-freshness tradeoffs.
• Availability during rotation/outages. Cooldowns may delay acceptance of a newly published key. Expired caches cannot be used during an outage. Document these behaviors without assuming every provider publishes new keys sufficiently early.
• Synchronous API in async applications. Thread offloading is the Phase 1 integration. Cache synchronization, network budgets and cancellation behavior need tests; prefetch does not make verification permanently free of I/O.
• Gateway configuration. Response format and caching are deployment decisions. The helper can validate payload compatibility but cannot infer every authorizer setting.
• Provider differences. Access-token purposes, resource identifiers and scope representations differ. Factories and examples must state the profile they support. A JWKS endpoint alone does not guarantee compatibility.

Dependencies and Integrations

Component	Proposed dependencies	Packaging
JWTVerifier	Supported PyJWT and cryptography versions	Required to use JWT verification; optional to the base package through [auth].
OAuth2Client	Explicitly declared urllib3 dependency	Included in [auth]; do not depend on botocore's transitive dependency or the runtime's installed version.

Reuse PyJWT for cryptography and key parsing/retrieval where its behavior satisfies the contract. PyJWKClient already has key-set caching and refresh on key misses. Its optional individual-key cache is not time-expiring; leave it disabled or ensure any replacement is invalidated with the parent set. Evaluate library defaults against the refresh, concurrency and failure rules above.[^pyjwt-api][^pyjwt-usage]

Integrations:

• Event Handler: require() follows the existing middleware protocol; claims via app.append_context().
• Data Classes: accept the supported authorizer event classes and use response builders internally; authorize() returns a serialized dictionary.
• Parameters: callable client_secret and static jwks pair naturally with get_secret() / get_parameter().
• Logger: diagnostics use reason codes and redact credentials, tokens and sensitive endpoint data.
• MCP Python SDK: a documented adapter with a stated SDK target; no mcp import or dependency inside Powertools.

Alternative solutions

| Option | Tradeoff |
|---|---|
| Recommend PyJWT directly | Already handles JWT cryptography, JWKS retrieval, caching and key lookup. A viable option when applications supply their own profiles and integrations. This proposal adds bounded refresh behavior, consistent errors, middleware and Lambda authorizer support. |
| Recommend a broader OAuth library such as Authlib | Consider as an implementation alternative or for applications needing more grants and authentication methods. It does not remove the need to define Powertools-specific verification and integration behavior. |
| Use another JOSE library, such as `python-jose` | Evaluate supported algorithms and dependency maintenance during implementation. The Lambda integration and cache-policy requirements remain whichever library is selected. |
| API Gateway JWT authorizer | Prefer managed verification when it satisfies the application's deployment and token-profile needs. An in-function utility remains useful for other entry points and application-level permission checks. |
| Include SigV4 signing in the same utility | Adds a separate authentication mechanism and API surface; keep this proposal focused on bearer tokens. |
| Build an MCP-specific resource-server abstraction | Adds an SDK-specific abstraction where a verifier adapter is sufficient for this scope. |

Acknowledgment

• This feature request meets Powertools for AWS Lambda (Python) Tenets
• Should this be considered in other Powertools for AWS Lambda languages? i.e. Java, TypeScript, and .NET

[powertools-for-aws-oss-automation]

Thanks for opening your first issue here! We'll come back to you as soon as we can.
In the meantime, check out our GitHub Discussions: Join the conversation