httpx-oauth2-flows

Typed, asynchronous OAuth 2 token flows and authentication for HTTPX, using OpenID Connect discovery.

The package can acquire tokens directly or act as an asynchronous httpx.Auth implementation that acquires, caches, refreshes, and attaches access tokens to outgoing requests.

[!IMPORTANT] This package is asynchronous only. It acquires OAuth 2 access tokens but is not a complete OpenID Connect relying-party implementation: ID tokens are not exposed or validated.

Supported Features

Implicit, Resource Owner Password Credentials, Device Authorization, SAML Bearer, Token Exchange, and CIBA are not implemented.

Requirements

• Python 3.12 or newer
• httpx.AsyncClient
• An authorization server exposing OpenID Connect discovery at {auth_server}/.well-known/openid-configuration

Authorization Code also requires:

• A system browser
• Permission to bind an ephemeral port on 127.0.0.1
• A provider that accepts a dynamic loopback redirect URI
• An OpenID UserInfo endpoint

For the default callback path, register a redirect URI in this form with the provider:

http://127.0.0.1:<dynamic-port>/authorize-callback

Native application providers commonly allow any port for a registered loopback redirect. Providers that require one exact port are not compatible with the current Authorization Code implementation.

Installation

Install from PyPI with pip:

python -m pip install httpx-oauth2-flows

Or with uv:

uv add httpx-oauth2-flows

Input Normalization

Public Config and Flow constructors accept convenient input values and normalize them immediately. The model attributes remain the canonical Url and Scope types, so application and library code never needs to handle input unions after construction.

UrlLike accepts:

• str
• Url or httpx.URL
• pydantic.HttpUrl or pydantic.AnyHttpUrl

Every URL is validated as HTTP or HTTPS and stored in canonical form as Url. Relative URLs, other schemes, bytes, paths, and arbitrary stringifiable objects are rejected.

ScopeLike accepts:

• Scope
• An OAuth scope expression such as 'openid profile'
• list[str], tuple[str, ...], set[str], or frozenset[str]

A string is split on ASCII spaces. Collection elements are individual scope tokens. Tokens are validated using the RFC 6749 scope-token grammar, duplicates are removed, and transmission order is sorted. Empty values, ambiguous whitespace, mappings, bytes, generators, and arbitrary iterables are rejected. Use None to omit the scope parameter.

from httpx_oauth2_flows import (
    ClientCredentialsClientSecretPostAuth,
    ClientCredentialsConfig,
    Scope,
    Url,
)

config = ClientCredentialsConfig(
    auth_server='https://identity.example.com',
    client_id='reporting-service',
    auth=ClientCredentialsClientSecretPostAuth(
        client_secret='read-from-a-secret-manager'
    ),
    scope='reports.write reports.read',
)

assert isinstance(config.auth_server, Url)
assert isinstance(config.scope, Scope)

UrlLike and ScopeLike are closed type aliases, not structural protocols. This prevents accidental conversion of objects that merely implement __str__ or __iter__.

Quick Start

Use Auth with an AuthorizationCodeFlow to authenticate requests from a desktop or command-line application:

import asyncio

import httpx

from httpx_oauth2_flows import Auth, AuthorizationCodeFlow


async def main() -> None:
    oauth = Auth(
        AuthorizationCodeFlow(
            auth_server='https://identity.example.com',
            client_id='desktop-client',
            scope='openid profile email reports.read',
        ),
        service_name='reports-cli:identity.example.com:desktop-client',
    )

    async with httpx.AsyncClient(
        base_url='https://api.example.com',
        auth=oauth,
        timeout=10,
    ) as client:
        response = await client.get('/reports/current')
        response.raise_for_status()
        print(response.json())


asyncio.run(main())

On the first request, the flow:

1. Starts a temporary callback server on 127.0.0.1.
2. Discovers the provider's authorization and token endpoints.
3. Opens the authorization request in the system browser.
4. Validates the callback state and exchanges the code using S256 PKCE.
5. Fetches UserInfo and displays a localized completion page.
6. Stores the token in memory and, when available, the system keyring.
7. Adds the token to the API request's Authorization header.

Use a unique service_name for each application, authorization server, and client ID. The keyring lookup identity does not otherwise distinguish clients or scopes on the same authorization server.

Auth cannot be used with synchronous httpx.Client.

Provider Discovery

Every flow resolves its endpoints from:

https://identity.example.com/.well-known/openid-configuration

The discovery document must contain issuer. Additional requirements depend on the flow:

If grant_types_supported is present, the selected grant must be listed. If code_challenge_methods_supported is present, Authorization Code requires S256. Client Credentials uses token_endpoint_auth_methods_supported to validate or automatically select client authentication.

OAuth Authorization Server Metadata at /.well-known/oauth-authorization-server and manually configured endpoints are not currently supported.

Authorization Code

Authorization Code is intended for public/native clients. It does not send a client secret or other client authentication during code exchange.

The default configuration is:

The package does not add the openid scope automatically. Request every scope required by the provider, including openid when its UserInfo endpoint requires it.

Direct Execution

Use authorization_code_flow when the application needs the token response instead of automatic HTTPX authentication:

import asyncio

import httpx

from httpx_oauth2_flows import AuthorizationCodeConfig, authorization_code_flow


async def main() -> None:
    config = AuthorizationCodeConfig(
        auth_server='https://identity.example.com',
        client_id='desktop-client',
        scope='openid profile reports.read',
    )

    async with httpx.AsyncClient(timeout=10) as client:
        token = await authorization_code_flow(config, client)
        print(token.access_token)


asyncio.run(main())

Callback Customization

The built-in completion page provides Bulgarian, Czech, German, English, Spanish, French, Hungarian, Indonesian, Italian, Japanese, Korean, Macedonian, Dutch, Polish, Portuguese, Romanian, Russian, Swedish, Turkish, and Chinese text.

Override or add locales with CallbackResponseTexts:

from httpx_oauth2_flows import AuthorizationCodeFlow, CallbackResponseTexts

flow = AuthorizationCodeFlow(
    auth_server='https://identity.example.com',
    client_id='desktop-client',
    callback_response_localized_texts={
        'en': CallbackResponseTexts(
            title='Sign-in complete',
            body='Welcome, $username. You may return to the application.',
            body_male=None,
            body_female=None,
            button_text='Close tab',
            footer_text='This window can be closed safely.',
        )
    },
    callback_response_default_locale='en',
)

Custom body text may use $username. A custom HTML template may use $title, $body, $button_text, $footer_text, and $lang.

Only use trusted templates and trusted text. Template substitutions and UserInfo values are inserted as HTML and are not escaped by the current implementation.

Client Credentials

Client Credentials supports these token endpoint authentication methods:

If token_endpoint_auth_methods_supported is absent, the package assumes its supported methods are available.

Client Secret

Read client secrets from a secret manager or environment variable rather than source code:

import asyncio
import os

import httpx

from httpx_oauth2_flows import (
    ClientCredentialsClientSecretPostAuth,
    ClientCredentialsConfig,
    client_credentials_flow,
)


async def main() -> None:
    config = ClientCredentialsConfig(
        auth_server='https://identity.example.com',
        client_id='reporting-service',
        auth=ClientCredentialsClientSecretPostAuth(
            client_secret=os.environ['OAUTH_CLIENT_SECRET']
        ),
        scope='reports.read',
    )

    async with httpx.AsyncClient(timeout=10) as client:
        token = await client_credentials_flow(config, client)
        response = await client.get(
            'https://api.example.com/reports/current',
            headers={'Authorization': f'{token.token_type} {token.access_token}'},
        )
        response.raise_for_status()
        print(response.json())


asyncio.run(main())

To let discovery choose between Basic and Post, replace the authentication configuration with:

from httpx_oauth2_flows import ClientCredentialsClientSecretAutoAuth

auth = ClientCredentialsClientSecretAutoAuth(client_secret='read-from-a-secret-manager')

ClientCredentialsClientSecretBasicAuth accepts header_name for providers that require the Basic value in a nonstandard header.

Private Key JWT

Private Key JWT signs a client assertion with iss and sub set to the client ID and aud set to the discovered token endpoint:

import os
from pathlib import Path

from httpx_oauth2_flows import (
    ClientCredentialsConfig,
    ClientCredentialsPrivateKeyJwtAuth,
    SignerFilePrivateKey,
)

config = ClientCredentialsConfig(
    auth_server='https://identity.example.com',
    client_id='reporting-service',
    scope='reports.read',
    auth=ClientCredentialsPrivateKeyJwtAuth(
        private_key=SignerFilePrivateKey(
            path=Path('/run/secrets/oauth-signing-key.pem'),
            password=os.environ['OAUTH_KEY_PASSWORD'].encode(),
        ),
        key_id='reporting-service-2026',
    ),
)

Mutual TLS

Configure an X.509 certificate and, when separate, its private key:

from pathlib import Path

from httpx_oauth2_flows import ClientCredentialsConfig, ClientCredentialsTlsClientAuth

config = ClientCredentialsConfig(
    auth_server='https://identity.example.com',
    client_id='reporting-service',
    auth=ClientCredentialsTlsClientAuth(
        certfile=Path('/run/secrets/client.crt'),
        keyfile=Path('/run/secrets/client.key'),
        password='read-from-a-secret-manager',
    ),
)

The certificate is used for the token request, not discovery. The token request uses a dedicated HTTPX client and therefore does not inherit the supplied client's transport, proxy, CA, headers, cookies, or timeout configuration.

JWT Bearer

The JWT Bearer grant exchanges a signed assertion for an access token:

import asyncio
import os
from pathlib import Path

import httpx

from httpx_oauth2_flows import (
    JwtBearerConfig,
    SignerFilePrivateKey,
    jwt_bearer_execute,
)


async def main() -> None:
    config = JwtBearerConfig(
        auth_server='https://identity.example.com',
        client_id='automation-client',
        user_id='service-account@example.com',
        scope='reports.read',
        private_key=SignerFilePrivateKey(
            path=Path('/run/secrets/assertion-key.pem'),
            password=os.environ['ASSERTION_KEY_PASSWORD'].encode(),
        ),
        key_id='automation-client-2026',
    )

    async with httpx.AsyncClient(timeout=10) as client:
        token = await jwt_bearer_execute(config, client)
        print(token.access_token)


asyncio.run(main())

The assertion contains iss, sub, aud, jti, iat, and exp. user_id controls sub and defaults to the client ID. The audience is the discovered issuer, which the provider must accept. Assertions are valid for five minutes by default; change this with jwt_duration.

Signing Keys

SignerInlinePrivateKey accepts an unencrypted PEM string. SignerFilePrivateKey loads a PEM file using a required byte-string password.

Default algorithms are selected from the key type:

Override the default with algorithm only when the key and provider support that algorithm. Set key_id to add a kid header to the assertion.

Token Responses

Direct flow functions and Flow.execute() return SuccessfulTokenResponse:

Unknown response fields, including id_token, are ignored.

HTTPX Auth, Caching, and Refresh

Auth accepts AuthorizationCodeFlow, ClientCredentialsFlow, or JwtBearerFlow and performs the following work before an API request:

1. Reuses a sufficiently long-lived in-memory token when possible.
2. For Authorization Code, attempts to load a persisted token from the system keyring.
3. Attempts an unauthenticated refresh-token grant for eligible Authorization Code tokens.
4. Executes the configured flow if no valid token remains.
5. Sets Authorization: {token_type} {access_token} on the request.

Tokens are considered valid only when their lifetime exceeds the combined HTTPX connect, read, write, and pool timeout for the request.

If expires_in is missing, Auth treats the access token as a JWT and reads its exp claim without verifying its signature. An opaque token without expires_in, or a JWT without exp, cannot be cached by Auth.

Keyring behavior:

• Tokens are persisted when a keyring backend is available.
• Persisted tokens are reloaded only for Authorization Code.
• The default service name is httpx_oauth2_flows.
• The keyring username is derived from the canonical authorization-server URL. Legacy root URLs without a trailing slash are loaded and migrated automatically.
• clear_stored_token() removes persisted state but does not clear the current in-memory token.
• A 401 Unauthorized or 403 Forbidden response removes persisted state but is not retried.

Concurrent requests are not synchronized around token acquisition and may start duplicate flows.

Error Handling

OAuth-specific errors share the public OAuth2FlowError base class:

import httpx

from httpx_oauth2_flows import OAuth2FlowError, TokenError, TokenSchemaError

try:
    response = await client.get('/reports/current')
    response.raise_for_status()
except TokenError as error:
    print(error.error, error.error_description, error.error_uri)
except TokenSchemaError as error:
    print('The token endpoint returned an invalid response:', error)
except OAuth2FlowError as error:
    print('The OAuth flow failed:', error)
except httpx.HTTPError as error:
    print('The HTTP request failed:', error)

Public OAuth exceptions are:

Transport errors remain httpx.HTTPError subclasses. Invalid provider documents and configuration may also raise pydantic.ValidationError, RuntimeError, ValueError, or TypeError.

Exceptions for malformed remote responses can include response headers and bodies. Avoid logging them where logs may be accessible to untrusted users.

Security Considerations

• Use trusted HTTPS authorization servers and API endpoints. HTTPS is not enforced by the package.
• Never hardcode client secrets, private keys, or key passwords in source code.
• Treat access tokens, refresh tokens, authorization codes, and exception details as secrets.
• Use a unique Auth.service_name for each application, provider, and client ID.
• Review the security properties of the operating system's configured keyring backend.
• Keep the system clock synchronized when creating JWT assertions.
• Validate access tokens at the resource server; this package does not validate them.
• Do not use the returned UserInfo or discarded ID token as proof of identity.
• Only use trusted callback HTML and localized text because substitutions are not HTML-escaped.

Known Limitations

• Only asynchronous HTTPX authentication is implemented.
• All flows require OpenID Connect discovery; endpoint overrides are unavailable.
• Authorization Code supports public clients with S256 PKCE only.
• Authorization Code always fetches UserInfo after token exchange.
• The callback port is dynamic and cannot be configured.
• Provider-specific authorization parameters such as audience, resource, prompt, and login_hint cannot be added.
• Refresh-token requests do not support client secrets, Private Key JWT, or mutual TLS.
• A refreshed response that omits refresh_token does not retain the previous refresh token.
• Mutual TLS token requests do not reuse the supplied HTTPX client configuration.
• ID tokens are discarded and not validated.
• Access tokens are not inspected or validated, except for the unverified exp fallback used by Auth when expires_in is missing.
• Requests are not automatically retried after a 401 or 403 response.

Public API

The supported package-level imports are grouped below.

Flow, ClientCredentialsAuth, SignerPrivateKey, CallbackResponseLocalizedTexts, ScopeLike, and UrlLike are type aliases, not constructors.

Use Url and Scope when explicit canonical values are useful. Scope accepts OAuth scope expressions and the same closed collection types, removes duplicates, and sends scopes in sorted order:

from httpx_oauth2_flows import Scope

scope = Scope('openid profile', {'reports.read', 'reports.write'})

Configuration classes are frozen Pydantic models. The corresponding Flow classes add an execute(http_client) method and can be passed to Auth. The standalone functions accept the matching Config class.

Development

Clone the repository and install its locked dependencies:

git clone https://gitlab.com/daude_f/httpx-oauth2-flows.git
cd httpx-oauth2-flows
uv sync

Run the verification suite:

uv run pytest
uv run ruff check
uv run ruff format --check
uv run pyright
uv build

License

Licensed under the Apache License 2.0.