Registration

RFC 7591 Dynamic Client Registration. build_register_handler assembles a Starlette endpoint that parses a registration request, applies host-supplied policy, and delegates persistence to a pluggable ClientRegistry.

RFC 7591 Dynamic Client Registration components.

build_register_handler

build_register_handler

build_register_handler(
    registry: ClientRegistry,
    *,
    default_scope: str,
    rate_limiter: SlidingWindowRateLimiter | None = None,
    auth_validator: RegistrationAuthValidator | None = None,
    get_client_ip: ClientIpResolver | None = None,
    default_redirect_uris: Sequence[str] = (),
    redirect_uri_rewriters: Iterable[
        RedirectUriRewriter
    ] = (),
    redirect_uri_validator: RedirectUriValidator
    | None = _default_redirect_uri_valid,
    client_name_factory: ClientNameFactory | None = None,
    post_register_hooks: Iterable[PostRegisterHook] = (),
) -> Callable[[Request], Awaitable[Response]]

Return a Starlette handler implementing RFC 7591 registration.

Parameters:

Name	Type	Description	Default
registry	ClientRegistry	Persistence backend that issues and stores clients.	required
default_scope	str	Scope granted when the request omits one. Also included in the registration response.	required
rate_limiter	SlidingWindowRateLimiter | None	Optional per-IP limiter applied before parsing. Skipped when None (tests / trusted networks).	None
auth_validator	RegistrationAuthValidator | None	Optional async callable gating the endpoint (RFC 7591 §3.1 initial access token). Invoked before the rate-limit check; a falsy return yields 401 and the request is not processed. When None the endpoint is open (rate limiting is not authentication) — production deployments SHOULD configure this.	None
get_client_ip	ClientIpResolver | None	Resolves the rate-limit key from the request. Defaults to the direct TCP peer (request.client.host), which is the proxy IP behind a reverse proxy / LB / ingress. Pass a callable that consults X-Forwarded-For only for explicitly trusted proxy CIDRs (or run ProxyHeadersMiddleware) to key on the real client.	None
default_redirect_uris	Sequence[str]	Fallback redirect_uris if the request sends none. Empty by default — most deployments should set this or reject requests that omit URIs.	()
redirect_uri_rewriters	Iterable[RedirectUriRewriter]	Ordered list of callables applied to the redirect_uris list. Each receives and returns the list, allowing additions (e.g. debug-variant expansion) or normalization.	()
redirect_uri_validator	RedirectUriValidator | None	Predicate applied to each redirect_uri after rewriting; any URI returning falsy is rejected with invalid_redirect_uri (400). Defaults to an https-only policy (http allowed for loopback hosts) per OAuth 2.1 §9.7. Pass None to disable validation, or a custom predicate to override the policy (e.g. to allow native-app custom schemes).	_default_redirect_uri_valid
client_name_factory	ClientNameFactory | None	Override the client name. Receives the parsed request; returns the name to assign. When None the request's client_name (or a registry-supplied default) is used.	None
post_register_hooks	Iterable[PostRegisterHook]	Async callables invoked with the newly issued RegisteredClient after persistence (e.g. to populate an in-process cache).	()

Source code in mcp_authflow/registration/handler.py

def build_register_handler(
    registry: ClientRegistry,
    *,
    default_scope: str,
    rate_limiter: SlidingWindowRateLimiter | None = None,
    auth_validator: RegistrationAuthValidator | None = None,
    get_client_ip: ClientIpResolver | None = None,
    default_redirect_uris: Sequence[str] = (),
    redirect_uri_rewriters: Iterable[RedirectUriRewriter] = (),
    redirect_uri_validator: RedirectUriValidator | None = _default_redirect_uri_valid,
    client_name_factory: ClientNameFactory | None = None,
    post_register_hooks: Iterable[PostRegisterHook] = (),
) -> Callable[[Request], Awaitable[Response]]:
    """Return a Starlette handler implementing RFC 7591 registration.

    Args:
        registry: Persistence backend that issues and stores clients.
        default_scope: Scope granted when the request omits one. Also
            included in the registration response.
        rate_limiter: Optional per-IP limiter applied before parsing.
            Skipped when ``None`` (tests / trusted networks).
        auth_validator: Optional async callable gating the endpoint
            (RFC 7591 §3.1 initial access token). Invoked before the
            rate-limit check; a falsy return yields ``401`` and the
            request is not processed. When ``None`` the endpoint is open
            (rate limiting is not authentication) — production
            deployments SHOULD configure this.
        get_client_ip: Resolves the rate-limit key from the request.
            Defaults to the direct TCP peer (``request.client.host``),
            which is the proxy IP behind a reverse proxy / LB / ingress.
            Pass a callable that consults ``X-Forwarded-For`` only for
            explicitly trusted proxy CIDRs (or run
            ``ProxyHeadersMiddleware``) to key on the real client.
        default_redirect_uris: Fallback ``redirect_uris`` if the request
            sends none. Empty by default — most deployments should set
            this or reject requests that omit URIs.
        redirect_uri_rewriters: Ordered list of callables applied to the
            ``redirect_uris`` list. Each receives and returns the list,
            allowing additions (e.g. debug-variant expansion) or
            normalization.
        redirect_uri_validator: Predicate applied to each redirect_uri
            after rewriting; any URI returning falsy is rejected with
            ``invalid_redirect_uri`` (400). Defaults to an https-only
            policy (http allowed for loopback hosts) per OAuth 2.1 §9.7.
            Pass ``None`` to disable validation, or a custom predicate to
            override the policy (e.g. to allow native-app custom schemes).
        client_name_factory: Override the client name. Receives the
            parsed request; returns the name to assign. When ``None``
            the request's ``client_name`` (or a registry-supplied
            default) is used.
        post_register_hooks: Async callables invoked with the newly
            issued ``RegisteredClient`` after persistence (e.g. to
            populate an in-process cache).
    """
    rewriters = list(redirect_uri_rewriters)
    hooks = list(post_register_hooks)
    defaults = list(default_redirect_uris)
    client_ip_of = get_client_ip or _default_client_ip

    async def register_handler(request: Request) -> Response:
        if auth_validator is not None and not await auth_validator(request):
            logger.warning("DCR request: authorization rejected")
            return invalid_client("Registration not authorized")

        if rate_limiter is not None:
            caller_ip = client_ip_of(request)
            if not await rate_limiter.is_allowed(caller_ip):
                retry_after = await rate_limiter.get_retry_after(caller_ip)
                return rate_limit_exceeded("Too many registration requests", retry_after)

        try:
            body = await request.body()
            payload: dict[str, Any] = json.loads(body) if body else {}
        except json.JSONDecodeError as e:
            logger.warning("DCR request: invalid JSON (%s)", e)
            return invalid_request("Invalid JSON")

        if not isinstance(payload, dict):
            return invalid_request("Request body must be a JSON object")

        logger.info(
            "DCR request: client_name=%r grant_types=%r redirect_uris=%r",
            payload.get("client_name"),
            payload.get("grant_types"),
            payload.get("redirect_uris"),
        )

        redirect_uris = list(payload.get("redirect_uris") or [])
        for rewriter in rewriters:
            redirect_uris = rewriter(redirect_uris)
        if not redirect_uris:
            redirect_uris = list(defaults)

        if redirect_uri_validator is not None:
            for uri in redirect_uris:
                if not redirect_uri_validator(uri):
                    logger.warning("DCR request: rejected redirect_uri %r", uri)
                    return invalid_redirect_uri(f"Invalid redirect_uri: {uri}")

        grant_types, auth_method = _derive_grant_types_and_auth_method(
            payload.get("grant_types") or []
        )
        response_types = list(payload.get("response_types") or ["code"])
        scope = payload.get("scope") or default_scope

        parsed = ClientRegistrationRequest(
            client_name=payload.get("client_name"),
            redirect_uris=redirect_uris,
            grant_types=grant_types,
            response_types=response_types,
            token_endpoint_auth_method=auth_method,
            scope=scope,
            extra={
                k: v
                for k, v in payload.items()
                if k
                not in {
                    "client_name",
                    "redirect_uris",
                    "grant_types",
                    "response_types",
                    "token_endpoint_auth_method",
                    "scope",
                }
            },
        )
        if client_name_factory is not None:
            parsed = dataclasses.replace(parsed, client_name=client_name_factory(parsed))

        try:
            client = await registry.create_client(parsed)
        except Exception:
            logger.exception("DCR: registry.create_client failed")
            return server_error("Failed to register client")

        for hook in hooks:
            try:
                await hook(client)
            except Exception:
                logger.exception(
                    "DCR: post_register hook failed for client %s",
                    client.client_id,
                )

        logger.info("DCR: registered client %s", client.client_id)
        return JSONResponse(_build_response_body(client), status_code=201)

    return register_handler

ClientRegistry

ClientRegistry

Bases: ABC

Persistence interface for dynamically registered OAuth clients.

Implementations decide where clients live (in-memory, database, delegated to an upstream identity service). The handler factory in :mod:mcp_authflow.registration.handler is storage-agnostic and drives this interface.

create_client abstractmethod async

create_client(
    request: ClientRegistrationRequest,
) -> RegisteredClient

Issue credentials for a new client and persist it.

Implementations are responsible for generating client_id and, for confidential clients, client_secret.

Source code in mcp_authflow/registration/base.py

@abstractmethod
async def create_client(self, request: ClientRegistrationRequest) -> RegisteredClient:
    """Issue credentials for a new client and persist it.

    Implementations are responsible for generating ``client_id`` and,
    for confidential clients, ``client_secret``.
    """
    ...

get_client abstractmethod async

get_client(client_id: str) -> RegisteredClient | None

Look up a previously registered client by id.

Source code in mcp_authflow/registration/base.py

@abstractmethod
async def get_client(self, client_id: str) -> RegisteredClient | None:
    """Look up a previously registered client by id."""
    ...

ClientRegistrationRequest

ClientRegistrationRequest dataclass

ClientRegistrationRequest(
    client_name: str | None,
    redirect_uris: list[str],
    grant_types: list[str],
    response_types: list[str],
    token_endpoint_auth_method: str,
    scope: str | None,
    extra: dict[str, object] = dict(),
)

Parsed and normalized RFC 7591 registration request.

Only the fields used by the handler are modeled; additional metadata received in the request body is preserved in extra for adapters that want to forward it to a backend.

is_public_client property

is_public_client: bool

True if the requested auth method indicates a public client.

RegisteredClient

RegisteredClient dataclass

RegisteredClient(
    client_id: str,
    client_secret: str | None,
    client_name: str | None,
    redirect_uris: list[str],
    grant_types: list[str],
    response_types: list[str],
    token_endpoint_auth_method: str,
    scope: str | None,
    client_id_issued_at: int,
    client_secret_expires_at: int = 0,
)

An OAuth client that has been issued credentials.

Returned by ClientRegistry.create_client. client_secret is None for public clients (token_endpoint_auth_method == "none").

MemoryClientRegistry

MemoryClientRegistry

MemoryClientRegistry()

Bases: ClientRegistry

Process-local client registry. Not persistent across restarts.

Source code in mcp_authflow/registration/memory.py

def __init__(self) -> None:
    self._clients: dict[str, RegisteredClient] = {}