`symfonic.core.oauth_provider`¶

oauth_provider ¶

AnthropicOAuthProvider — DEV/TEST USE ONLY.

ToS caveat¶

Replaying a Claude subscription OAuth token to the Anthropic API from a third-party program is subscription-auth reuse, which may be throttled or disallowed under your plan's Terms of Service. Anthropic ships the sanctioned claude -p CLI for subscription-backed use. Treat this provider as the same risk class as :class:~symfonic.core.codex_provider.CodexOAuthProvider — a local dev/test convenience only. For production, prefer :class:~symfonic.core.providers.AnthropicProvider with an ANTHROPIC_API_KEY.

Uses Claude Code's OAuth access token to call Anthropic's messages API against the user's Claude Pro/Max subscription instead of an API key.

When to use¶

Local development without provisioning a separate ANTHROPIC_API_KEY
Running the scaffolded test suite without API credits
Experimenting with the full pipeline (autonomous learning, metacognition) against a real LLM during development

When NOT to use¶

❌ Production multi-tenant SaaS — subscription tokens don't bill per-tenant, rate limits are different from API-key plans, and token refresh requires an interactive OAuth flow that breaks headless deployments.
❌ CI that runs across many developers — each dev's token bills against their own subscription.
❌ Cost-tracking tests — subscription usage doesn't map cleanly to the per-token pricing in symfonic.core.observability.pricing.

This provider explicitly refuses to work in environments that look like production (SYMFONIC_ENV=production or ENVIRONMENT=production) to prevent accidental use under multi-tenant billing.

Credential sources¶

macOS keychain: security find-generic-password -s "Claude Code-credentials" -w
Linux/Windows: ~/.claude/.credentials.json

Both return a JSON document shaped like::

{
  "claudeAiOauth": {
    "accessToken": "sk-ant-oat01-…",
    "refreshToken": "sk-ant-ort01-…",
    "expiresAt": 1776462035348,
    "scopes": ["user:inference", …],
    "subscriptionType": "max" | "pro" | …
  }
}

AnthropicOAuthProvider ¶

AnthropicOAuthProvider(
    *,
    allow_production: bool = False,
    access_token: str | None = None,
    refresh_token: str | None = None,
    credentials_resolver: Callable[[], Mapping[str, str]]
    | None = None,
)

DEV/TEST provider that uses Claude Code's OAuth token.

Instantiate via AnthropicOAuthProvider() on a developer machine that has run claude login. Pass the instance to SymfonicAgent(model_provider=...).

Parameters:

Name	Type	Description	Default
`allow_production`	`bool`	Skip the production-env guard. Default False. Set True only if you have an explicit, ToS-reviewed reason.	`False`
`access_token`	`str \| None`	SECONDARY single-tenant convenience (v8.5.0, issue #4). A static OAuth access token bound at construction. Documented footgun: if this provider instance is cached and shared across tenants, the static token is reused by every request — re-introducing the cross-tenant leak. Use `credentials_resolver` for multi-tenant deployments.	`None`
`refresh_token`	`str \| None`	Optional static refresh token companion to `access_token` (currently informational — the client-swap authenticates with the access token).	`None`
`credentials_resolver`	`Callable[[], Mapping[str, str]] \| None`	PRIMARY multi-tenant path (v8.5.0). A zero-arg callable invoked inside :meth:`get_chat_model` on every call, returning a mapping with at least `{"access_token": "sk-ant-oat01-..."}` (and optionally `"refresh_token"`). The resolved token is threaded into the fresh client-swap and never stored on the instance — safe under a cached/shared provider.	`None`

Credential precedence (inside :meth:get_chat_model): credentials_resolver() > static access_token > eager Keychain / credentials.json load (the v8.4.1 fallback).

Raises:

Type	Description
`OAuthCredentialError`	Credentials missing or malformed on the eager-load fallback path.
`ProductionMisuseError`	Detected a production env var and `allow_production=False` — fires only on the fallback path (no resolver and no static `access_token`); an explicit resolver/token is the deliberate, reviewed opt-in.

Source code in src/symfonic/core/oauth_provider.py

def __init__(
    self,
    *,
    allow_production: bool = False,
    access_token: str | None = None,
    refresh_token: str | None = None,
    credentials_resolver: Callable[[], Mapping[str, str]] | None = None,
) -> None:
    # One-time ToS reminder at construction. Logged (not a
    # DeprecationWarning): this provider is a supported dev/test
    # convenience, but subscription-auth reuse carries a real ToS risk
    # the caller should see. Mirrors the Codex provider's posture.
    logger.warning(_TOS_CAVEAT_MSG)

    self._static_access_token = access_token
    self._static_refresh_token = refresh_token
    self._credentials_resolver = credentials_resolver

    # When an explicit credential seam is supplied (resolver OR static
    # access_token), skip the eager Keychain load AND the prod-env
    # guard: the explicit injection is the deliberate, reviewed opt-in
    # for headless/multi-tenant deployments.  The guard exists to stop
    # accidental subscription billing in prod via the implicit
    # Keychain path — so it stays active on the fallback path only.
    if credentials_resolver is not None or access_token is not None:
        self._oauth = None
        self._credential_source = (
            "resolver" if credentials_resolver is not None else "static"
        )
        return

    # Fallback path — byte-identical to v8.4.1.
    if _is_production_env() and not allow_production:
        msg = (
            "AnthropicOAuthProvider is for dev/test only. Detected a "
            "production environment (SYMFONIC_ENV/ENVIRONMENT/ENV/NODE_ENV). "
            "Use AnthropicProvider with ANTHROPIC_API_KEY for production. "
            "Pass allow_production=True to override (not recommended)."
        )
        raise ProductionMisuseError(msg)

    self._oauth, self._credential_source = _load_credentials()
    _warn_if_expiring(self._oauth)
    logger.info(
        "AnthropicOAuthProvider initialized: "
        "subscription=%s tier=%s token=%s source=%s scopes=%s",
        self._oauth.get("subscriptionType", "unknown"),
        self._oauth.get("rateLimitTier", "unknown"),
        _token_fingerprint(self._oauth.get("accessToken", "")),
        self._credential_source,
        self._oauth.get("scopes", []),
    )

credential_source `property` ¶

credential_source: str

Return where the credential was loaded from.

"keychain" on macOS (Keychain item "Claude Code-credentials"), an absolute filesystem path on the fallback path, or "resolver" / "static" when an explicit seam is wired.

subscription_type `property` ¶

subscription_type: str

Claude subscription tier: "max", "pro", etc.

token_fingerprint `property` ¶

token_fingerprint: str

Return a short identifier for the active token.

Useful for logging / banners so you can tell whether two processes are sharing the same Claude Code session. Format: sk-ant-oat01-…xxxxxxxx (prefix + last 8 chars).

On the explicit-seam paths the token is request-scoped and not held on the instance, so we fingerprint the static access_token when present (resolver-supplied tokens are never retained, hence <empty>).

get_chat_model ¶

get_chat_model(config: ModelConfig) -> BaseChatModel

Return a ChatAnthropic using the OAuth token.

Strategy: construct ChatAnthropic with a placeholder api_key, then swap its internal _client / _async_client for anthropic SDK clients built with auth_token=... (which selects Bearer auth at the SDK level) + the required anthropic-beta: oauth-2025-04-20 default header.

Subclassing is avoided so this provider keeps behaving like a drop-in replacement for AnthropicProvider.

Source code in src/symfonic/core/oauth_provider.py

def get_chat_model(self, config: ModelConfig) -> BaseChatModel:
    """Return a ChatAnthropic using the OAuth token.

    Strategy: construct ChatAnthropic with a placeholder api_key,
    then swap its internal ``_client`` / ``_async_client`` for
    anthropic SDK clients built with ``auth_token=...`` (which
    selects Bearer auth at the SDK level) + the required
    ``anthropic-beta: oauth-2025-04-20`` default header.

    Subclassing is avoided so this provider keeps behaving like a
    drop-in replacement for ``AnthropicProvider``.
    """
    try:
        import anthropic
        from langchain_anthropic import ChatAnthropic
    except ImportError as exc:
        raise ImportError(
            "langchain-anthropic is required for AnthropicOAuthProvider. "
            "Install with: pip install symfonic-core[anthropic]",
        ) from exc

    # Resolve the per-call token (resolver > static > eager Keychain).
    # Local-only — never stored on ``self``; goes out of scope when
    # this method returns.
    token = self._resolve_access_token()

    kwargs: dict[str, Any] = {
        "model": config.model_name,
        "temperature": config.temperature,
        "max_tokens": config.max_tokens,
        "max_retries": config.max_retries,
        # Placeholder: SDK rejects empty api_key; we swap the
        # clients below before any request goes out.
        "anthropic_api_key": "sk-ant-oauth-placeholder",
    }
    if config.extra_headers:
        kwargs["default_headers"] = dict(config.extra_headers)
    if config.thinking:
        kwargs["model_kwargs"] = {"thinking": config.thinking}

    chat = ChatAnthropic(**kwargs)

    # Swap the underlying SDK clients to use auth_token (Bearer).
    # We also inject the oauth-beta header so the server dispatches
    # to the subscription-billed path. Each SDK client carries an httpx
    # client with a log-only 401/403 response hook so an expired/invalid
    # OAuth token surfaces ONE actionable WARNING at request time (this
    # complements the proactive construction-time `_warn_if_expiring`).
    import httpx

    from ._oauth_errors import (
        make_auth_failure_log_hook,
        make_auth_failure_log_hook_async,
    )

    oauth_default_headers = {"anthropic-beta": _OAUTH_BETA_HEADER}
    remediation = (
        "Your Claude Code OAuth token is expired or invalid. Run "
        "`claude login`, then retry."
    )
    resp_sync = make_auth_failure_log_hook(
        logger, provider="Anthropic", remediation=remediation
    )
    resp_async = make_auth_failure_log_hook_async(
        logger, provider="Anthropic", remediation=remediation
    )
    chat._client = anthropic.Anthropic(
        auth_token=token,
        default_headers=oauth_default_headers,
        http_client=httpx.Client(event_hooks={"response": [resp_sync]}),
    )
    chat._async_client = anthropic.AsyncAnthropic(
        auth_token=token,
        default_headers=oauth_default_headers,
        http_client=httpx.AsyncClient(event_hooks={"response": [resp_async]}),
    )
    return chat

OAuthCredentialError ¶

Bases: RuntimeError

Raised when Claude Code credentials cannot be loaded.

ProductionMisuseError ¶