kaboo_workflows.auth

Outbound MCP auth strategies: relay, on-behalf-of, machine-to-machine and static tokens. These are what an auth: block on an MCP client resolves to — see Chapter 9 for the YAML form.

kaboo_workflows.auth

Authentication helpers for kaboo-workflows.

Inbound identity is carried by :class:~kaboo_workflows._context.Principal (read via :func:~kaboo_workflows._context.get_auth_context). Outbound MCP auth is expressed with the :class:httpx2.Auth strategies below.

M2MClientCredentialsAuth

M2MClientCredentialsAuth(
    *,
    token_url,
    client_id,
    client_secret,
    scope=None,
    audience=None,
    extra=None,
    header="Authorization",
    scheme="Bearer",
)

Bases: _BearerAuth

OAuth2 client-credentials (machine-to-machine) token for MCP calls.

Fetches a token from token_url using the client-credentials grant and caches it until shortly before expiry. Independent of the inbound caller — represents the workflow service's own machine identity, so it is safe on a long-lived shared MCP client.

Parameters:

Name	Type	Description	Default
token_url	str	OAuth2 token endpoint.	required
client_id	str	Client identifier.	required
client_secret	str	Client secret.	required
scope	str | None	Optional space-delimited scopes.	None
audience	str | None	Optional audience parameter (e.g. Auth0).	None
extra	dict[str, str] | None	Extra form fields to include in the token request.	None
header	str | Sequence[str]	Header to set, or several (default Authorization).	'Authorization'
scheme	str	Auth scheme prefix (default Bearer).	'Bearer'

Source code in src/kaboo_workflows/auth/strategies.py

def __init__(
    self,
    *,
    token_url: str,
    client_id: str,
    client_secret: str,
    scope: str | None = None,
    audience: str | None = None,
    extra: dict[str, str] | None = None,
    header: str | Sequence[str] = "Authorization",
    scheme: str = "Bearer",
) -> None:
    self._token_url = token_url
    self._client_id = client_id
    self._client_secret = client_secret
    self._scope = scope
    self._audience = audience
    self._extra = extra or {}
    self.header_names = _header_names(header)
    self.scheme = scheme
    self._cache: tuple[str, float] | None = None
    self._lock = threading.Lock()

RelayTokenAuth

RelayTokenAuth(
    *,
    token=None,
    header="Authorization",
    scheme="Bearer",
    context_headers=None,
)

Bases: _BearerAuth

Forward the inbound caller token to the MCP unchanged.

Reads the current :class:~kaboo_workflows._context.Principal token unless an explicit token is given (useful when binding a per-request client with a captured token).

Parameters:

Name	Type	Description	Default
token	str | None	Explicit token to forward. When None, the inbound principal's token is used at call time.	None
header	str | Sequence[str]	Header to set, or several (default Authorization). A managed gateway that authenticates on Authorization and forwards its own credential needs both it and the header the target reads.	'Authorization'
scheme	str	Auth scheme prefix (default Bearer; "" for a raw value).	'Bearer'
context_headers	Mapping[str, str] | None	Headers to stamp with the current run's context, each mapped to run_id, thread_id or forwarded_props.<key>. A header whose value is missing for this run is left off.	None

Source code in src/kaboo_workflows/auth/strategies.py

def __init__(
    self,
    *,
    token: str | None = None,
    header: str | Sequence[str] = "Authorization",
    scheme: str = "Bearer",
    context_headers: Mapping[str, str] | None = None,
) -> None:
    self._explicit = token
    self.header_names = _header_names(header)
    self.scheme = scheme
    self.context_headers = dict(context_headers or {})
    for source in self.context_headers.values():
        _check_context_source(source)

StaticTokenAuth

StaticTokenAuth(
    *, token, header="Authorization", scheme="Bearer"
)

Bases: _BearerAuth

Attach a fixed token (e.g. a long-lived API key) to every MCP request.

Parameters:

Name	Type	Description	Default
token	str	The credential to send.	required
header	str | Sequence[str]	Header to set, or several (default Authorization).	'Authorization'
scheme	str	Auth scheme prefix (default Bearer; "" for a raw value).	'Bearer'

Source code in src/kaboo_workflows/auth/strategies.py

def __init__(
    self,
    *,
    token: str,
    header: str | Sequence[str] = "Authorization",
    scheme: str = "Bearer",
) -> None:
    self._value = token
    self.header_names = _header_names(header)
    self.scheme = scheme

apply_auth_to_transport_options

apply_auth_to_transport_options(
    options, auth, *, transport
)

Return transport_options with auth wired in for the given transport.

SSE accepts an auth kwarg directly. Streamable-http has no auth kwarg, so the auth rides a dedicated httpx2.AsyncClient (folding in any headers and timeout the caller set, with the same semantics as :func:~kaboo_workflows.mcp.transports.streamable_http_transport). A user-provided http_client / auth is left untouched.

Source code in src/kaboo_workflows/auth/strategies.py

def apply_auth_to_transport_options(
    options: dict[str, Any] | None,
    auth: httpx2.Auth,
    *,
    transport: str,
) -> dict[str, Any]:
    """Return ``transport_options`` with *auth* wired in for the given transport.

    SSE accepts an ``auth`` kwarg directly. Streamable-http has no ``auth``
    kwarg, so the auth rides a dedicated ``httpx2.AsyncClient`` (folding in any
    ``headers`` and ``timeout`` the caller set, with the same semantics as
    :func:`~kaboo_workflows.mcp.transports.streamable_http_transport`). A
    user-provided ``http_client`` / ``auth`` is left untouched.
    """
    opts = dict(options or {})
    if transport == "sse":
        opts.setdefault("auth", auth)
        return opts
    if "http_client" not in opts:
        from ..mcp.transports import _httpx_timeout

        headers = opts.pop("headers", None)
        timeout = opts.pop("timeout", None)
        kwargs: dict[str, Any] = {"auth": auth, "headers": headers or None}
        if timeout is not None:
            kwargs["timeout"] = _httpx_timeout(timeout)
        opts["http_client"] = httpx2.AsyncClient(**kwargs)
    return opts

build_auth

build_auth(strategy, params=None)

Build an outbound MCP auth strategy from a name + params.

Parameters:

Name	Type	Description	Default
strategy	str	One of relay, static, m2m.	required
params	dict[str, Any] | None	Constructor keyword arguments for the strategy.	None

Returns:

Type	Description
Auth	A configured :class:httpx2.Auth instance.

Raises:

Type	Description
ValueError	If strategy is not a known strategy name.

Source code in src/kaboo_workflows/auth/strategies.py

def build_auth(strategy: str, params: dict[str, Any] | None = None) -> httpx2.Auth:
    """Build an outbound MCP auth strategy from a name + params.

    Args:
        strategy: One of ``relay``, ``static``, ``m2m``.
        params: Constructor keyword arguments for the strategy.

    Returns:
        A configured :class:`httpx2.Auth` instance.

    Raises:
        ValueError: If ``strategy`` is not a known strategy name.
    """
    try:
        cls = _STRATEGIES[strategy]
    except KeyError:
        raise ValueError(
            f"Unknown MCP auth strategy '{strategy}'. Available: {', '.join(sorted(_STRATEGIES))}."
        ) from None
    return cls(**(params or {}))