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:httpx.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()

OBOTokenAuth

OBOTokenAuth(
    *,
    provider,
    region="us-east-1",
    scopes=None,
    workload_name=None,
    workload_token=None,
    custom_parameters=None,
    force_authentication=False,
    header="Authorization",
    scheme="Bearer",
)

Bases: _BearerAuth

AgentCore On-Behalf-Of token exchange for a downstream resource.

Two exchanges happen, and both run inside AgentCore Identity:

1. The inbound end-user token is exchanged for a workload access token (GetWorkloadAccessTokenForJWT), which is what binds this agent's workload identity to that user. Set workload_name to perform it here; leave it unset when the runtime already hands the workload token in on the request, in which case it is read from the principal.
2. The workload access token is exchanged for a downstream access token (GetResourceOauth2Token) against the named credential provider.

Downstream tokens are cached per workload token until shortly before expiry.

Provider differences belong in configuration, not in this class: custom_parameters is passed through to the exchange, which is how an Entra ID provider gets its requested_token_use=on_behalf_of, and header / scheme decide how the result is presented downstream.

Parameters:

Name	Type	Description	Default
provider	str	Name of the AgentCore OAuth2 credential provider (the downstream resource) to exchange for.	required
region	str	AWS region of the AgentCore control plane.	'us-east-1'
scopes	list[str] | None	OAuth2 scopes to request. The API requires the field, so an empty list is sent when none are given.	None
workload_name	str | None	Workload identity to mint a workload access token for from the inbound user token. Unset means the inbound request is expected to carry the workload token already.	None
workload_token	str | None	Explicit workload token (bypasses the principal).	None
custom_parameters	dict[str, str] | None	Extra provider-specific parameters forwarded to the token exchange.	None
force_authentication	bool	Skip AgentCore's cached token for this identity.	False
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,
    *,
    provider: str,
    region: str = "us-east-1",
    scopes: list[str] | None = None,
    workload_name: str | None = None,
    workload_token: str | None = None,
    custom_parameters: dict[str, str] | None = None,
    force_authentication: bool = False,
    header: str | Sequence[str] = "Authorization",
    scheme: str = "Bearer",
) -> None:
    self._provider = provider
    self._region = region
    self._scopes = scopes or []
    self._workload_name = workload_name
    self._explicit_workload = workload_token
    self._custom_parameters = custom_parameters or {}
    self._force = force_authentication
    self.header_names = _header_names(header)
    self.scheme = scheme
    self._client: Any = None
    self._cache: dict[str, tuple[str, float]] = {}
    self._workload_cache: dict[str, str] = {}
    self._lock = threading.Lock()

RelayTokenAuth

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

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'

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

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

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 httpx.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: httpx.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 ``httpx.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"] = httpx.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, obo, m2m.	required
params	dict[str, Any] | None	Constructor keyword arguments for the strategy.	None

Returns:

Type	Description
Auth	A configured :class:httpx.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) -> httpx.Auth:
    """Build an outbound MCP auth strategy from a name + params.

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

    Returns:
        A configured :class:`httpx.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 {}))