Skip to content

API Reference

Public API of litestar-keycloak. Import from the top-level package:

from litestar_keycloak import (
    KeycloakPlugin,
    KeycloakConfig,
    TokenLocation,
    KeycloakUser,
    TokenPayload,
    CurrentUser,
    CurrentTokenPayload,
    CurrentRawToken,
    require_roles,
    require_client_roles,
    require_scopes,
    MatchStrategy,
)

Plugin and config

litestar_keycloak.plugin.KeycloakPlugin

Bases: InitPluginProtocol

Litestar plugin that integrates Keycloak authentication.

Usage::

from litestar import Litestar
from litestar_keycloak import KeycloakPlugin, KeycloakConfig

app = Litestar(
    route_handlers=[...],
    plugins=[KeycloakPlugin(
        KeycloakConfig(
            server_url="https://keycloak.example.com",
            realm="my-realm",
            client_id="my-app",
        )
    )],
)
Source code in src/litestar_keycloak/plugin.py 
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
class KeycloakPlugin(InitPluginProtocol):
    """Litestar plugin that integrates Keycloak authentication.

    Usage::

        from litestar import Litestar
        from litestar_keycloak import KeycloakPlugin, KeycloakConfig

        app = Litestar(
            route_handlers=[...],
            plugins=[KeycloakPlugin(
                KeycloakConfig(
                    server_url="https://keycloak.example.com",
                    realm="my-realm",
                    client_id="my-app",
                )
            )],
        )
    """

    __slots__ = ("_config", "_http", "_jwks_cache", "_verifier")

    def __init__(self, config: KeycloakConfig) -> None:
        self._config = config
        self._http = KeycloakHttpClient(config.http_timeout)
        self._jwks_cache = JWKSCache(
            jwks_url=config.jwks_url,
            ttl=config.jwks_cache_ttl,
            http=self._http,
        )
        self._verifier = TokenVerifier(config, self._jwks_cache)

    def on_app_init(self, app_config: AppConfig) -> AppConfig:
        """Hook called by Litestar during application startup.

        Registers:
        - Authentication middleware (bearer token extraction + validation)
        - Exception handlers (``KeycloakError`` hierarchy -> HTTP responses)
        - DI providers (``current_user``, ``token_payload``, ``raw_token``)
        - OIDC routes (when ``include_routes`` is ``True``)
        - Lifespan handlers to warm the JWKS cache on startup (best-effort) and
          close the shared HTTP client on shutdown
        """
        # -- auth middleware -----------------------------------------------
        # Appended (not inserted at 0) so it runs *after* any app-level session
        # middleware — required for callback_response_mode="redirect", where the
        # token is read from the session the session middleware populates.
        middleware_cls = create_auth_middleware(self._config, self._verifier)
        app_config.middleware.append(
            DefineMiddleware(
                middleware_cls,
                exclude=self._config.exclude_auth_patterns,
                exclude_from_auth_key=self._config.exclude_opt_key,
            )
        )

        # -- exception handlers --------------------------------------------
        app_config.exception_handlers = cast(
            "Any",
            {
                **exception_handlers,
                **app_config.exception_handlers,
            },
        )

        # -- DI providers --------------------------------------------------
        app_config.dependencies = {
            **build_dependencies(),
            **app_config.dependencies,
        }

        # -- OIDC routes ---------------------------------------------------
        if self._config.include_routes:
            controller = build_auth_controller(self._config, self._http)
            app_config.route_handlers.append(controller)

        # -- lifespan: warm JWKS on startup, close HTTP client on shutdown --
        app_config.on_startup.append(self._on_startup)
        app_config.on_shutdown.append(self._on_shutdown)

        return app_config

    async def _on_startup(self) -> None:
        """Warm the JWKS cache so the first request doesn't pay fetch latency.

        Best-effort: if Keycloak is unreachable at boot the warm-up is skipped
        rather than aborting application startup — the cache refetches lazily on
        the first request (which returns 502 until Keycloak is reachable).
        """
        try:
            await self._jwks_cache.warm()
        except JWKSFetchError as exc:
            logger.warning(
                "JWKS warm-up skipped (Keycloak unreachable at startup): %s. "
                "Keys will be fetched on the first request.",
                exc,
            )

    async def _on_shutdown(self) -> None:
        """Close the shared HTTP client so its connections are released."""
        await self._http.close()

__slots__ = ('_config', '_http', '_jwks_cache', '_verifier') class-attribute instance-attribute

_config = config instance-attribute

_http = KeycloakHttpClient(config.http_timeout) instance-attribute

_jwks_cache = JWKSCache(jwks_url=(config.jwks_url), ttl=(config.jwks_cache_ttl), http=(self._http)) instance-attribute

_verifier = TokenVerifier(config, self._jwks_cache) instance-attribute

__init__(config)

Source code in src/litestar_keycloak/plugin.py 
57
58
59
60
61
62
63
64
65
def __init__(self, config: KeycloakConfig) -> None:
    self._config = config
    self._http = KeycloakHttpClient(config.http_timeout)
    self._jwks_cache = JWKSCache(
        jwks_url=config.jwks_url,
        ttl=config.jwks_cache_ttl,
        http=self._http,
    )
    self._verifier = TokenVerifier(config, self._jwks_cache)

_on_shutdown() async

Close the shared HTTP client so its connections are released.

Source code in src/litestar_keycloak/plugin.py 
133
134
135
async def _on_shutdown(self) -> None:
    """Close the shared HTTP client so its connections are released."""
    await self._http.close()

_on_startup() async

Warm the JWKS cache so the first request doesn't pay fetch latency.

Best-effort: if Keycloak is unreachable at boot the warm-up is skipped rather than aborting application startup — the cache refetches lazily on the first request (which returns 502 until Keycloak is reachable).

Source code in src/litestar_keycloak/plugin.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
async def _on_startup(self) -> None:
    """Warm the JWKS cache so the first request doesn't pay fetch latency.

    Best-effort: if Keycloak is unreachable at boot the warm-up is skipped
    rather than aborting application startup — the cache refetches lazily on
    the first request (which returns 502 until Keycloak is reachable).
    """
    try:
        await self._jwks_cache.warm()
    except JWKSFetchError as exc:
        logger.warning(
            "JWKS warm-up skipped (Keycloak unreachable at startup): %s. "
            "Keys will be fetched on the first request.",
            exc,
        )

on_app_init(app_config)

Hook called by Litestar during application startup.

Registers: - Authentication middleware (bearer token extraction + validation) - Exception handlers (KeycloakError hierarchy -> HTTP responses) - DI providers (current_user, token_payload, raw_token) - OIDC routes (when include_routes is True) - Lifespan handlers to warm the JWKS cache on startup (best-effort) and close the shared HTTP client on shutdown

Source code in src/litestar_keycloak/plugin.py 
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
def on_app_init(self, app_config: AppConfig) -> AppConfig:
    """Hook called by Litestar during application startup.

    Registers:
    - Authentication middleware (bearer token extraction + validation)
    - Exception handlers (``KeycloakError`` hierarchy -> HTTP responses)
    - DI providers (``current_user``, ``token_payload``, ``raw_token``)
    - OIDC routes (when ``include_routes`` is ``True``)
    - Lifespan handlers to warm the JWKS cache on startup (best-effort) and
      close the shared HTTP client on shutdown
    """
    # -- auth middleware -----------------------------------------------
    # Appended (not inserted at 0) so it runs *after* any app-level session
    # middleware — required for callback_response_mode="redirect", where the
    # token is read from the session the session middleware populates.
    middleware_cls = create_auth_middleware(self._config, self._verifier)
    app_config.middleware.append(
        DefineMiddleware(
            middleware_cls,
            exclude=self._config.exclude_auth_patterns,
            exclude_from_auth_key=self._config.exclude_opt_key,
        )
    )

    # -- exception handlers --------------------------------------------
    app_config.exception_handlers = cast(
        "Any",
        {
            **exception_handlers,
            **app_config.exception_handlers,
        },
    )

    # -- DI providers --------------------------------------------------
    app_config.dependencies = {
        **build_dependencies(),
        **app_config.dependencies,
    }

    # -- OIDC routes ---------------------------------------------------
    if self._config.include_routes:
        controller = build_auth_controller(self._config, self._http)
        app_config.route_handlers.append(controller)

    # -- lifespan: warm JWKS on startup, close HTTP client on shutdown --
    app_config.on_startup.append(self._on_startup)
    app_config.on_shutdown.append(self._on_shutdown)

    return app_config

litestar_keycloak.config.KeycloakConfig dataclass

Immutable configuration for the Keycloak plugin.

Only server_url, realm, and client_id are required. Everything else has sensible defaults suitable for a typical confidential-client setup with RS256-signed JWTs.

Example::

KeycloakConfig(
    server_url="https://keycloak.example.com",
    realm="my-realm",
    client_id="my-app",
    client_secret="s3cret",
)
Source code in src/litestar_keycloak/config.py 
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
@dataclass(frozen=True, slots=True)
class KeycloakConfig:
    """Immutable configuration for the Keycloak plugin.

    Only ``server_url``, ``realm``, and ``client_id`` are required.
    Everything else has sensible defaults suitable for a typical
    confidential-client setup with RS256-signed JWTs.

    Example::

        KeycloakConfig(
            server_url="https://keycloak.example.com",
            realm="my-realm",
            client_id="my-app",
            client_secret="s3cret",
        )
    """

    # -- connection --------------------------------------------------------
    server_url: str
    """Base Keycloak URL without trailing slash (e.g. ``https://kc.example.com``)."""

    realm: str
    """Keycloak realm name."""

    client_id: str
    """OIDC client identifier registered in the realm."""

    client_secret: str | None = None
    """Client secret for confidential clients.  ``None`` for public clients."""

    # -- token -------------------------------------------------------------
    token_location: TokenLocation = TokenLocation.HEADER
    """Where to read the bearer token from: ``Authorization`` header or a cookie."""

    cookie_name: str = "access_token"
    """Cookie name the middleware reads when ``token_location`` is ``COOKIE``.
    The plugin never *sets* this cookie — how the token gets there is the
    application's responsibility (see ``callback_response_mode``)."""

    cookie_secure: bool = True
    """``Secure`` attribute on the OAuth ``state`` cookie the routes set.  Set
    ``False`` for plain-HTTP local development."""

    cookie_samesite: Literal["lax", "strict", "none"] = "lax"
    """``SameSite`` attribute on the OAuth ``state`` cookie the routes set."""

    algorithms: tuple[str, ...] = ("RS256",)
    """Accepted JWT signing algorithms."""

    scopes: tuple[str, ...] = ("openid",)
    """Scopes requested during the authorization code flow."""

    # -- JWKS --------------------------------------------------------------
    jwks_cache_ttl: int = 3600
    """How long (seconds) to cache the JWKS before re-fetching."""

    # -- routes ------------------------------------------------------------
    include_routes: bool = False
    """Mount ``/auth/login``, ``/callback``, ``/logout``, ``/refresh`` routes."""

    auth_prefix: str = "/auth"
    """URL prefix for the optional OIDC route group."""

    redirect_uri: str | None = None
    """OAuth2 redirect URI for the authorization code flow.
    Required when ``include_routes`` is ``True``."""

    callback_response_mode: Literal["json", "redirect"] = "json"
    """What ``/auth/callback`` does after exchanging the code:

    - ``"json"`` (default): return the token response as JSON — the SPA/BFF
      stores the access token and sends it as ``Authorization: Bearer``.
    - ``"redirect"``: store the tokens in the **server-side session** and redirect
      to ``post_login_redirect_uri`` — for server-rendered apps.  Requires
      Litestar session middleware (e.g. ``ServerSideSessionConfig``); the auth
      middleware then reads the access token from the session."""

    post_login_redirect_uri: str = "/"
    """Where ``redirect``-mode login redirects after storing tokens in the session."""

    post_logout_redirect_uri: str | None = None
    """Where ``redirect``-mode logout redirects.  Returns JSON status when ``None``."""

    # -- advanced ----------------------------------------------------------
    audience: str | None = None
    """Expected ``aud`` claim.  Defaults to ``client_id`` when ``None``."""

    expected_issuer: str | None = None
    """Expected ``iss`` claim.  Defaults to ``realm_url`` when ``None``.  Set this
    when Keycloak's frontend/hostname URL differs from ``server_url`` (e.g. behind
    a reverse proxy), so token issuer validation matches the value Keycloak signs."""

    optional_audiences: frozenset[str] = field(default_factory=frozenset)
    """Additional audiences to accept (e.g. service client IDs)."""

    expected_token_type: str | None = "Bearer"
    """Required payload ``typ`` claim.  ``"Bearer"`` (default) rejects Keycloak
    ID and Refresh tokens presented as access tokens.  Set to ``None`` to disable
    the check (e.g. non-Keycloak OIDC providers that omit ``typ``)."""

    http_timeout: int = 10
    """Timeout in seconds for outgoing HTTP calls to Keycloak."""

    excluded_paths: frozenset[str] = field(default_factory=frozenset)
    """Exact request paths that skip authentication entirely (e.g. ``/health``).
    Matched literally — for prefixes or subtrees use ``exclude_patterns``."""

    exclude_patterns: tuple[str, ...] = ()
    """Regex patterns whose matching request paths skip authentication.

    Unlike ``excluded_paths`` (exact match), these cover prefixes and subtrees,
    e.g. ``("^/public/", "^/docs")``.  Patterns are matched unanchored against
    the path, so anchor with ``^`` to match from the start."""

    exclude_opt_key: str = "exclude_from_auth"
    """Route-handler ``opt`` key for per-handler opt-out, e.g.
    ``@get("/open", exclude_from_auth=True)``."""

    # -- derived properties ------------------------------------------------

    @property
    def realm_url(self) -> str:
        """``{server_url}/realms/{realm}`` — base for all OIDC endpoints."""
        return f"{self.server_url.rstrip('/')}/realms/{self.realm}"

    @property
    def issuer(self) -> str:
        """Expected ``iss`` claim value.

        Defaults to ``realm_url``.  Set ``expected_issuer`` when Keycloak issues
        tokens under a different base (its configured frontend/hostname URL),
        which is common behind a reverse proxy where the public URL differs from
        the ``server_url`` the backend uses to reach Keycloak.
        """
        if self.expected_issuer is not None:
            return self.expected_issuer
        return self.realm_url

    @property
    def jwks_url(self) -> str:
        """JSON Web Key Set endpoint URL."""
        return f"{self.realm_url}/protocol/openid-connect/certs"

    @property
    def authorization_url(self) -> str:
        """Authorization endpoint for the code flow."""
        return f"{self.realm_url}/protocol/openid-connect/auth"

    @property
    def token_url(self) -> str:
        """Token endpoint for code exchange and refresh."""
        return f"{self.realm_url}/protocol/openid-connect/token"

    @property
    def logout_url(self) -> str:
        """End-session endpoint."""
        return f"{self.realm_url}/protocol/openid-connect/logout"

    @property
    def effective_audience(self) -> str:
        """Primary audience for JWT validation (explicit or client_id)."""
        return self.audience if self.audience is not None else self.client_id

    @property
    def accepted_audiences(self) -> frozenset[str]:
        """All audiences accepted for JWT validation (primary + optional_audiences)."""
        return frozenset({self.effective_audience}) | self.optional_audiences

    @property
    def effective_excluded_paths(self) -> frozenset[str]:
        """Paths that skip auth: excluded_paths plus auth routes when include_routes."""
        if not self.include_routes:
            return self.excluded_paths
        auth_paths = {
            f"{self.auth_prefix}/login",
            f"{self.auth_prefix}/callback",
            f"{self.auth_prefix}/logout",
            f"{self.auth_prefix}/refresh",
        }
        return self.excluded_paths | auth_paths

    @property
    def exclude_auth_patterns(self) -> list[str] | None:
        """Regex patterns for request paths that bypass the auth middleware.

        Exact ``excluded_paths`` (and the auth routes when ``include_routes``) are
        anchored to ``^…$``; ``exclude_patterns`` are used verbatim.  Returns
        ``None`` when nothing is excluded — an empty pattern list would compile to
        an empty regex that greedily matches every path.
        """
        patterns = [
            f"^{re.escape(path)}$" for path in sorted(self.effective_excluded_paths)
        ]
        patterns.extend(self.exclude_patterns)
        return patterns or None

    def __post_init__(self) -> None:
        if self.include_routes and self.redirect_uri is None:
            raise ValueError("redirect_uri is required when include_routes is True")
        if self.jwks_cache_ttl < 0:
            raise ValueError("jwks_cache_ttl must be non-negative")
        if self.http_timeout <= 0:
            raise ValueError("http_timeout must be positive")

accepted_audiences property

All audiences accepted for JWT validation (primary + optional_audiences).

algorithms = ('RS256',) class-attribute instance-attribute

Accepted JWT signing algorithms.

audience = None class-attribute instance-attribute

Expected aud claim. Defaults to client_id when None.

auth_prefix = '/auth' class-attribute instance-attribute

URL prefix for the optional OIDC route group.

authorization_url property

Authorization endpoint for the code flow.

callback_response_mode = 'json' class-attribute instance-attribute

What /auth/callback does after exchanging the code:

  • "json" (default): return the token response as JSON — the SPA/BFF stores the access token and sends it as Authorization: Bearer.
  • "redirect": store the tokens in the server-side session and redirect to post_login_redirect_uri — for server-rendered apps. Requires Litestar session middleware (e.g. ServerSideSessionConfig); the auth middleware then reads the access token from the session.

client_id instance-attribute

OIDC client identifier registered in the realm.

client_secret = None class-attribute instance-attribute

Client secret for confidential clients. None for public clients.

cookie_name = 'access_token' class-attribute instance-attribute

Cookie name the middleware reads when token_location is COOKIE. The plugin never sets this cookie — how the token gets there is the application's responsibility (see callback_response_mode).

cookie_samesite = 'lax' class-attribute instance-attribute

SameSite attribute on the OAuth state cookie the routes set.

cookie_secure = True class-attribute instance-attribute

Secure attribute on the OAuth state cookie the routes set. Set False for plain-HTTP local development.

effective_audience property

Primary audience for JWT validation (explicit or client_id).

effective_excluded_paths property

Paths that skip auth: excluded_paths plus auth routes when include_routes.

exclude_auth_patterns property

Regex patterns for request paths that bypass the auth middleware.

Exact excluded_paths (and the auth routes when include_routes) are anchored to ^…$; exclude_patterns are used verbatim. Returns None when nothing is excluded — an empty pattern list would compile to an empty regex that greedily matches every path.

exclude_opt_key = 'exclude_from_auth' class-attribute instance-attribute

Route-handler opt key for per-handler opt-out, e.g. @get("/open", exclude_from_auth=True).

exclude_patterns = () class-attribute instance-attribute

Regex patterns whose matching request paths skip authentication.

Unlike excluded_paths (exact match), these cover prefixes and subtrees, e.g. ("^/public/", "^/docs"). Patterns are matched unanchored against the path, so anchor with ^ to match from the start.

excluded_paths = field(default_factory=frozenset) class-attribute instance-attribute

Exact request paths that skip authentication entirely (e.g. /health). Matched literally — for prefixes or subtrees use exclude_patterns.

expected_issuer = None class-attribute instance-attribute

Expected iss claim. Defaults to realm_url when None. Set this when Keycloak's frontend/hostname URL differs from server_url (e.g. behind a reverse proxy), so token issuer validation matches the value Keycloak signs.

expected_token_type = 'Bearer' class-attribute instance-attribute

Required payload typ claim. "Bearer" (default) rejects Keycloak ID and Refresh tokens presented as access tokens. Set to None to disable the check (e.g. non-Keycloak OIDC providers that omit typ).

http_timeout = 10 class-attribute instance-attribute

Timeout in seconds for outgoing HTTP calls to Keycloak.

include_routes = False class-attribute instance-attribute

Mount /auth/login, /callback, /logout, /refresh routes.

issuer property

Expected iss claim value.

Defaults to realm_url. Set expected_issuer when Keycloak issues tokens under a different base (its configured frontend/hostname URL), which is common behind a reverse proxy where the public URL differs from the server_url the backend uses to reach Keycloak.

jwks_cache_ttl = 3600 class-attribute instance-attribute

How long (seconds) to cache the JWKS before re-fetching.

jwks_url property

JSON Web Key Set endpoint URL.

logout_url property

End-session endpoint.

optional_audiences = field(default_factory=frozenset) class-attribute instance-attribute

Additional audiences to accept (e.g. service client IDs).

post_login_redirect_uri = '/' class-attribute instance-attribute

Where redirect-mode login redirects after storing tokens in the session.

post_logout_redirect_uri = None class-attribute instance-attribute

Where redirect-mode logout redirects. Returns JSON status when None.

realm instance-attribute

Keycloak realm name.

realm_url property

{server_url}/realms/{realm} — base for all OIDC endpoints.

redirect_uri = None class-attribute instance-attribute

OAuth2 redirect URI for the authorization code flow. Required when include_routes is True.

scopes = ('openid',) class-attribute instance-attribute

Scopes requested during the authorization code flow.

server_url instance-attribute

Base Keycloak URL without trailing slash (e.g. https://kc.example.com).

token_location = TokenLocation.HEADER class-attribute instance-attribute

Where to read the bearer token from: Authorization header or a cookie.

token_url property

Token endpoint for code exchange and refresh.

__init__(server_url, realm, client_id, client_secret=None, token_location=TokenLocation.HEADER, cookie_name='access_token', cookie_secure=True, cookie_samesite='lax', algorithms=('RS256',), scopes=('openid',), jwks_cache_ttl=3600, include_routes=False, auth_prefix='/auth', redirect_uri=None, callback_response_mode='json', post_login_redirect_uri='/', post_logout_redirect_uri=None, audience=None, expected_issuer=None, optional_audiences=frozenset(), expected_token_type='Bearer', http_timeout=10, excluded_paths=frozenset(), exclude_patterns=(), exclude_opt_key='exclude_from_auth')

__post_init__()

Source code in src/litestar_keycloak/config.py 
222
223
224
225
226
227
228
def __post_init__(self) -> None:
    if self.include_routes and self.redirect_uri is None:
        raise ValueError("redirect_uri is required when include_routes is True")
    if self.jwks_cache_ttl < 0:
        raise ValueError("jwks_cache_ttl must be non-negative")
    if self.http_timeout <= 0:
        raise ValueError("http_timeout must be positive")

litestar_keycloak.config.TokenLocation

Bases: Enum

Where the plugin looks for the access token on incoming requests.

Source code in src/litestar_keycloak/config.py 
18
19
20
21
22
class TokenLocation(enum.Enum):
    """Where the plugin looks for the access token on incoming requests."""

    HEADER = "header"
    COOKIE = "cookie"

COOKIE = 'cookie' class-attribute instance-attribute

HEADER = 'header' class-attribute instance-attribute

Models

litestar_keycloak.models.KeycloakUser dataclass

High-level identity object injected into route handlers.

Wraps TokenPayload and exposes a flattened, ergonomic API so handlers don't need to know about raw JWT claim structures::

@get("/me")
async def me(current_user: CurrentUser) -> dict:
    return {"name": current_user.name, "roles": current_user.realm_roles}
Source code in src/litestar_keycloak/models.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
@dataclass(frozen=True, slots=True)
class KeycloakUser:
    """High-level identity object injected into route handlers.

    Wraps ``TokenPayload`` and exposes a flattened, ergonomic API so
    handlers don't need to know about raw JWT claim structures::

        @get("/me")
        async def me(current_user: CurrentUser) -> dict:
            return {"name": current_user.name, "roles": current_user.realm_roles}
    """

    sub: str
    """Keycloak user ID."""

    preferred_username: str | None = None
    email: str | None = None
    email_verified: bool = False
    given_name: str | None = None
    family_name: str | None = None
    name: str | None = None

    realm_roles: frozenset[str] = field(default_factory=frozenset)
    """All realm-level roles."""

    client_roles: dict[str, frozenset[str]] = field(default_factory=dict)
    """Client roles keyed by client ID."""

    scopes: frozenset[str] = field(default_factory=frozenset)
    """Token scopes as a set."""

    raw: TokenPayload | None = None
    """Full token payload for advanced use cases."""

    def has_role(self, role: str) -> bool:
        """Check if the user holds a realm role."""
        return role in self.realm_roles

    def has_client_role(self, client_id: str, role: str) -> bool:
        """Check if the user holds a role for a specific client."""
        return role in self.client_roles.get(client_id, frozenset())

    def has_scope(self, scope: str) -> bool:
        """Check if the token carries a specific scope."""
        return scope in self.scopes

    @classmethod
    def from_token(cls, payload: TokenPayload) -> KeycloakUser:
        """Construct from a validated ``TokenPayload``."""
        client_roles = {
            client_id: frozenset(access.get("roles", []))
            for client_id, access in payload.resource_access.items()
        }
        return cls(
            sub=payload.sub,
            preferred_username=payload.preferred_username,
            email=payload.email,
            email_verified=payload.email_verified,
            given_name=payload.given_name,
            family_name=payload.family_name,
            name=payload.name,
            realm_roles=payload.realm_roles,
            client_roles=client_roles,
            scopes=payload.scopes,
            raw=payload,
        )

client_roles = field(default_factory=dict) class-attribute instance-attribute

Client roles keyed by client ID.

email = None class-attribute instance-attribute

email_verified = False class-attribute instance-attribute

family_name = None class-attribute instance-attribute

given_name = None class-attribute instance-attribute

name = None class-attribute instance-attribute

preferred_username = None class-attribute instance-attribute

raw = None class-attribute instance-attribute

Full token payload for advanced use cases.

realm_roles = field(default_factory=frozenset) class-attribute instance-attribute

All realm-level roles.

scopes = field(default_factory=frozenset) class-attribute instance-attribute

Token scopes as a set.

sub instance-attribute

Keycloak user ID.

__init__(sub, preferred_username=None, email=None, email_verified=False, given_name=None, family_name=None, name=None, realm_roles=frozenset(), client_roles=dict(), scopes=frozenset(), raw=None)

from_token(payload) classmethod

Construct from a validated TokenPayload.

Source code in src/litestar_keycloak/models.py 
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
@classmethod
def from_token(cls, payload: TokenPayload) -> KeycloakUser:
    """Construct from a validated ``TokenPayload``."""
    client_roles = {
        client_id: frozenset(access.get("roles", []))
        for client_id, access in payload.resource_access.items()
    }
    return cls(
        sub=payload.sub,
        preferred_username=payload.preferred_username,
        email=payload.email,
        email_verified=payload.email_verified,
        given_name=payload.given_name,
        family_name=payload.family_name,
        name=payload.name,
        realm_roles=payload.realm_roles,
        client_roles=client_roles,
        scopes=payload.scopes,
        raw=payload,
    )

has_client_role(client_id, role)

Check if the user holds a role for a specific client.

Source code in src/litestar_keycloak/models.py 
161
162
163
def has_client_role(self, client_id: str, role: str) -> bool:
    """Check if the user holds a role for a specific client."""
    return role in self.client_roles.get(client_id, frozenset())

has_role(role)

Check if the user holds a realm role.

Source code in src/litestar_keycloak/models.py 
157
158
159
def has_role(self, role: str) -> bool:
    """Check if the user holds a realm role."""
    return role in self.realm_roles

has_scope(scope)

Check if the token carries a specific scope.

Source code in src/litestar_keycloak/models.py 
165
166
167
def has_scope(self, scope: str) -> bool:
    """Check if the token carries a specific scope."""
    return scope in self.scopes

litestar_keycloak.models.TokenPayload dataclass

Decoded and validated JWT claims.

Mirrors the standard OIDC claims plus Keycloak-specific role structures. Constructed by the token module after successful validation — fields are guaranteed to satisfy all configured checks (issuer, audience, expiry, algorithm) by the time this object exists.

Source code in src/litestar_keycloak/models.py 
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
@dataclass(frozen=True, slots=True)
class TokenPayload:
    """Decoded and validated JWT claims.

    Mirrors the standard OIDC claims plus Keycloak-specific role
    structures.  Constructed by the ``token`` module after successful
    validation — fields are guaranteed to satisfy all configured checks
    (issuer, audience, expiry, algorithm) by the time this object exists.
    """

    # -- standard OIDC claims ---------------------------------------------
    sub: str
    """Subject identifier (Keycloak user ID, typically a UUID)."""

    iss: str
    """Issuer URL (``{server_url}/realms/{realm}``)."""

    aud: str | list[str]
    """Audience — single client ID or a list when multiple audiences are present."""

    exp: int
    """Expiration timestamp (Unix epoch seconds)."""

    iat: int
    """Issued-at timestamp (Unix epoch seconds)."""

    # -- optional standard claims ------------------------------------------
    azp: str | None = None
    """Authorized party — the client that requested the token."""

    scope: str = ""
    """Space-delimited scope string (e.g. ``"openid profile email"``)."""

    jti: str | None = None
    """Unique token identifier."""

    typ: str | None = None
    """Token type (typically ``"Bearer"``)."""

    # -- Keycloak-specific claims ------------------------------------------
    preferred_username: str | None = None
    email: str | None = None
    email_verified: bool = False
    given_name: str | None = None
    family_name: str | None = None
    name: str | None = None

    realm_access: dict[str, Any] = field(default_factory=dict)
    """Raw ``realm_access`` claim (e.g. ``{"roles": ["admin", "user"]}``)."""

    resource_access: dict[str, Any] = field(default_factory=dict)
    """Raw ``resource_access`` claim keyed by client ID."""

    # -- extra claims catchall ---------------------------------------------
    extra: dict[str, Any] = field(default_factory=dict)
    """Any remaining claims not explicitly modeled above."""

    # -- convenience -------------------------------------------------------

    @property
    def realm_roles(self) -> frozenset[str]:
        """Realm-level roles extracted from ``realm_access.roles``."""
        return frozenset(self.realm_access.get("roles", []))

    def client_roles(self, client_id: str) -> frozenset[str]:
        """Roles for a specific client from ``resource_access``."""
        access = self.resource_access.get(client_id, {})
        return frozenset(access.get("roles", []))

    @property
    def scopes(self) -> frozenset[str]:
        """Scope string split into a set for O(1) membership tests."""
        return frozenset(self.scope.split()) if self.scope else frozenset()

    @property
    def expires_at(self) -> datetime:
        """``exp`` as a timezone-aware UTC datetime."""
        return datetime.fromtimestamp(self.exp, tz=UTC)

    @property
    def issued_at(self) -> datetime:
        """``iat`` as a timezone-aware UTC datetime."""
        return datetime.fromtimestamp(self.iat, tz=UTC)

    @classmethod
    def from_claims(cls, claims: dict[str, Any]) -> TokenPayload:
        """Build a ``TokenPayload`` from a raw decoded JWT dict.

        Known fields are mapped to explicit attributes; everything else
        lands in ``extra`` so no claim is silently dropped.
        Keycloak may omit ``aud``; fall back to ``azp`` when building.
        """
        # Normalize aud (Keycloak sometimes omits it; use azp as fallback)
        normalized = dict(claims)
        if not normalized.get("aud"):
            normalized["aud"] = normalized.get("azp") or ""
        # Some OIDC providers omit "sub" from access tokens;
        # falling back to preferred_username
        if "sub" not in normalized or normalized.get("sub") is None:
            normalized["sub"] = normalized.get("preferred_username") or ""
        known_fields = {f.name for f in cls.__dataclass_fields__.values()} - {"extra"}
        known = {k: v for k, v in normalized.items() if k in known_fields}
        extra = {k: v for k, v in normalized.items() if k not in known_fields}
        return cls(**known, extra=extra)

aud instance-attribute

Audience — single client ID or a list when multiple audiences are present.

azp = None class-attribute instance-attribute

Authorized party — the client that requested the token.

email = None class-attribute instance-attribute

email_verified = False class-attribute instance-attribute

exp instance-attribute

Expiration timestamp (Unix epoch seconds).

expires_at property

exp as a timezone-aware UTC datetime.

extra = field(default_factory=dict) class-attribute instance-attribute

Any remaining claims not explicitly modeled above.

family_name = None class-attribute instance-attribute

given_name = None class-attribute instance-attribute

iat instance-attribute

Issued-at timestamp (Unix epoch seconds).

iss instance-attribute

Issuer URL ({server_url}/realms/{realm}).

issued_at property

iat as a timezone-aware UTC datetime.

jti = None class-attribute instance-attribute

Unique token identifier.

name = None class-attribute instance-attribute

preferred_username = None class-attribute instance-attribute

realm_access = field(default_factory=dict) class-attribute instance-attribute

Raw realm_access claim (e.g. {"roles": ["admin", "user"]}).

realm_roles property

Realm-level roles extracted from realm_access.roles.

resource_access = field(default_factory=dict) class-attribute instance-attribute

Raw resource_access claim keyed by client ID.

scope = '' class-attribute instance-attribute

Space-delimited scope string (e.g. "openid profile email").

scopes property

Scope string split into a set for O(1) membership tests.

sub instance-attribute

Subject identifier (Keycloak user ID, typically a UUID).

typ = None class-attribute instance-attribute

Token type (typically "Bearer").

__init__(sub, iss, aud, exp, iat, azp=None, scope='', jti=None, typ=None, preferred_username=None, email=None, email_verified=False, given_name=None, family_name=None, name=None, realm_access=dict(), resource_access=dict(), extra=dict())

client_roles(client_id)

Roles for a specific client from resource_access.

Source code in src/litestar_keycloak/models.py 
81
82
83
84
def client_roles(self, client_id: str) -> frozenset[str]:
    """Roles for a specific client from ``resource_access``."""
    access = self.resource_access.get(client_id, {})
    return frozenset(access.get("roles", []))

from_claims(claims) classmethod

Build a TokenPayload from a raw decoded JWT dict.

Known fields are mapped to explicit attributes; everything else lands in extra so no claim is silently dropped. Keycloak may omit aud; fall back to azp when building.

Source code in src/litestar_keycloak/models.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
@classmethod
def from_claims(cls, claims: dict[str, Any]) -> TokenPayload:
    """Build a ``TokenPayload`` from a raw decoded JWT dict.

    Known fields are mapped to explicit attributes; everything else
    lands in ``extra`` so no claim is silently dropped.
    Keycloak may omit ``aud``; fall back to ``azp`` when building.
    """
    # Normalize aud (Keycloak sometimes omits it; use azp as fallback)
    normalized = dict(claims)
    if not normalized.get("aud"):
        normalized["aud"] = normalized.get("azp") or ""
    # Some OIDC providers omit "sub" from access tokens;
    # falling back to preferred_username
    if "sub" not in normalized or normalized.get("sub") is None:
        normalized["sub"] = normalized.get("preferred_username") or ""
    known_fields = {f.name for f in cls.__dataclass_fields__.values()} - {"extra"}
    known = {k: v for k, v in normalized.items() if k in known_fields}
    extra = {k: v for k, v in normalized.items() if k not in known_fields}
    return cls(**known, extra=extra)

Handler annotations

Annotate handler parameters with these; the parameter names must stay current_user, token_payload and raw_token, since injection is name-based.

Annotation Injects
CurrentUser KeycloakUser
CurrentTokenPayload TokenPayload
CurrentRawToken str

Guards

litestar_keycloak.guards.MatchStrategy

Bases: Enum

How multiple required values are evaluated.

Source code in src/litestar_keycloak/guards.py 
25
26
27
28
29
30
31
32
class MatchStrategy(enum.Enum):
    """How multiple required values are evaluated."""

    ALL = "all"
    """User must hold **every** required role/scope (default)."""

    ANY = "any"
    """User must hold **at least one** of the required roles/scopes."""

ALL = 'all' class-attribute instance-attribute

User must hold every required role/scope (default).

ANY = 'any' class-attribute instance-attribute

User must hold at least one of the required roles/scopes.

litestar_keycloak.guards.require_roles(*roles, strategy=MatchStrategy.ALL)

Guard factory that enforces realm-level roles.

Parameters:

Name Type Description Default
*roles str

One or more role names the user must hold.

 ()
strategy MatchStrategy

ALL (default) requires every role; ANY requires at least one.

 ALL

Example::

@get("/admin", guards=[require_roles("admin")])
async def admin_panel() -> dict: ...

@get("/staff", guards=[require_roles(
    "admin", "manager", strategy=MatchStrategy.ANY)])
async def staff_area() -> dict: ...
Source code in src/litestar_keycloak/guards.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
def require_roles(
    *roles: str,
    strategy: MatchStrategy = MatchStrategy.ALL,
) -> Callable[..., None]:
    """Guard factory that enforces realm-level roles.

    Args:
        *roles: One or more role names the user must hold.
        strategy: ``ALL`` (default) requires every role;
                  ``ANY`` requires at least one.

    Example::

        @get("/admin", guards=[require_roles("admin")])
        async def admin_panel() -> dict: ...

        @get("/staff", guards=[require_roles(
            "admin", "manager", strategy=MatchStrategy.ANY)])
        async def staff_area() -> dict: ...
    """
    required = frozenset(roles)

    def guard(
        connection: ASGIConnection[Any, Any, Any, Any], _: BaseRouteHandler
    ) -> None:
        user = _resolve_user(connection)
        if strategy is MatchStrategy.ALL:
            satisfied = required <= user.realm_roles
        else:
            satisfied = bool(required & user.realm_roles)

        if not satisfied:
            raise InsufficientRoleError(required=required, actual=user.realm_roles)

    return guard

litestar_keycloak.guards.require_client_roles(client_id, *roles, strategy=MatchStrategy.ALL)

Guard factory that enforces client-level roles.

Parameters:

Name Type Description Default
client_id str

The Keycloak client whose roles are checked.

 required
*roles str

One or more role names the user must hold for client_id.

 ()
strategy MatchStrategy

ALL (default) or ANY.

 ALL

Example::

@get("/billing", guards=[require_client_roles("billing-service", "read")])
async def billing() -> dict: ...
Source code in src/litestar_keycloak/guards.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
def require_client_roles(
    client_id: str,
    *roles: str,
    strategy: MatchStrategy = MatchStrategy.ALL,
) -> Callable[..., None]:
    """Guard factory that enforces client-level roles.

    Args:
        client_id: The Keycloak client whose roles are checked.
        *roles: One or more role names the user must hold for *client_id*.
        strategy: ``ALL`` (default) or ``ANY``.

    Example::

        @get("/billing", guards=[require_client_roles("billing-service", "read")])
        async def billing() -> dict: ...
    """
    required = frozenset(roles)

    def guard(
        connection: ASGIConnection[Any, Any, Any, Any], _: BaseRouteHandler
    ) -> None:
        user = _resolve_user(connection)
        actual = user.client_roles.get(client_id, frozenset())

        if strategy is MatchStrategy.ALL:
            satisfied = required <= actual
        else:
            satisfied = bool(required & actual)

        if not satisfied:
            raise InsufficientRoleError(required=required, actual=actual)

    return guard

litestar_keycloak.guards.require_scopes(*scopes, strategy=MatchStrategy.ALL)

Guard factory that enforces token scopes.

Parameters:

Name Type Description Default
*scopes str

One or more scope strings the token must carry.

 ()
strategy MatchStrategy

ALL (default) or ANY.

 ALL

Example::

@get("/reports", guards=[require_scopes("reports:read")])
async def reports() -> dict: ...
Source code in src/litestar_keycloak/guards.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
def require_scopes(
    *scopes: str,
    strategy: MatchStrategy = MatchStrategy.ALL,
) -> Callable[..., None]:
    """Guard factory that enforces token scopes.

    Args:
        *scopes: One or more scope strings the token must carry.
        strategy: ``ALL`` (default) or ``ANY``.

    Example::

        @get("/reports", guards=[require_scopes("reports:read")])
        async def reports() -> dict: ...
    """
    required = frozenset(scopes)

    def guard(
        connection: ASGIConnection[Any, Any, Any, Any], _: BaseRouteHandler
    ) -> None:
        user = _resolve_user(connection)

        if strategy is MatchStrategy.ALL:
            satisfied = required <= user.scopes
        else:
            satisfied = bool(required & user.scopes)

        if not satisfied:
            raise InsufficientScopeError(required=required, actual=user.scopes)

    return guard