idu-service-auth 0.1.0

Async Keycloak client_credentials token client for service-to-service authentication.

idu-service-auth

idu-service-auth is a small framework-agnostic async library for Machine-to-Machine authentication against Keycloak. It obtains OAuth2 client_credentials access tokens, caches the latest token, refreshes it before expiry, and can optionally run refresh on a background task.

The package is intended for async Python services such as FastAPI applications, FastMCP servers, background workers, and agents. It does not include framework adapters in the first version; applications own configuration loading and lifecycle wiring.

Features

• Async HTTP implementation based on aiohttp.
• Keycloak token endpoint support for the client_credentials grant.
• HTTP Basic authentication for client_id and client_secret.
• Lazy refresh by default, with optional background refresh.
• Concurrency-safe refresh through a single internal asyncio.Lock.
• Safe fallback to the previous token when refresh fails but the token is still valid.
• Explicit exception hierarchy for request, response, and lifecycle errors.
• Typed public API with a py.typed marker.

Installation

For local development:

poetry install --with dev

For runtime usage from this repository:

poetry install --only main

The project is packaged through pyproject.toml and poetry-core; a separate setup.py compatibility shim is included only for legacy tooling that still invokes setup.py directly.

Usage

from idu_service_auth import KeycloakTokenClient, KeycloakTokenConfig


config = KeycloakTokenConfig(
    auth_server_url="https://keycloak.example.com",
    realm="my-realm",
    client_id="service-a",
    client_secret="secret",
)

async with KeycloakTokenClient(config) as auth:
    token = await auth.get_access_token()
    headers = await auth.get_authorization_headers()

headers is ready to pass to an outgoing request:

{"Authorization": "Bearer <access-token>"}

More concrete framework examples are available in:

• examples/fastapi_app.py: FastAPI lifespan, shared aiohttp.ClientSession, dependency wiring, and an internal service call.
• examples/fastmcp_server.py: FastMCP lifespan context, shared runtime resources, and a tool that calls an internal API.

Configuration

KeycloakTokenConfig accepts the following parameters:

• auth_server_url: base Keycloak URL, for example https://keycloak.example.com or https://keycloak.example.com/auth.
• realm: Keycloak realm name.
• client_id: confidential service client ID.
• client_secret: confidential service client secret.
• scope: optional OAuth2 scope string.
• extra_token_params: optional additional form parameters for the token endpoint.
• request_timeout_seconds: total request timeout, default 10.
• refresh_before_expiry_seconds: preferred refresh margin, default 30.
• background_refresh: starts background refresh automatically inside the async context manager when True.

The token URL is built as:

{auth_server_url}/realms/{realm}/protocol/openid-connect/token

The client normalizes repeated trailing slashes and preserves deployments that use /auth in the base URL.

Refresh Behavior

By default, the client refreshes lazily. A call to get_access_token() returns the cached token while it is valid and requests a new token only when the cached token is missing or close to expiry.

The effective refresh margin is:

min(refresh_before_expiry_seconds, 20% of token TTL)

This avoids refresh loops for short-lived tokens.

For long-running services, background refresh can be enabled:

config = KeycloakTokenConfig(
    auth_server_url="https://keycloak.example.com",
    realm="my-realm",
    client_id="service-a",
    client_secret="secret",
    background_refresh=True,
)

async with KeycloakTokenClient(config) as auth:
    ...

If background refresh fails, the last valid token remains available. Retries use bounded backoff. If the token has already expired and Keycloak is still unavailable, the next lazy token request raises an authentication error.

External aiohttp Session

Applications may pass their own aiohttp.ClientSession:

import aiohttp


async with aiohttp.ClientSession() as session:
    async with KeycloakTokenClient(config, session=session) as auth:
        token = await auth.get_access_token()

When an external session is provided, the client does not close it. Internally created sessions are closed by aclose() and by the async context manager.

Errors

All library errors inherit from KeycloakAuthError.

• TokenRequestError: the HTTP request could not be completed.
• TokenResponseError: Keycloak returned an error response or invalid token payload.
• TokenClientClosedError: the client or its session is closed.

Development Commands

make install-dev
make format
make format-check
make lint
make test
make check
make build

Equivalent Poetry commands:

poetry run black idu_service_auth tests
poetry run black idu_service_auth tests examples
poetry run isort idu_service_auth tests examples
poetry run ruff check idu_service_auth tests examples
poetry run pytest
poetry build