symfonic.core.grok_oauth_provider

grok_oauth_provider

GrokOAuthProvider — DEV/TEST USE ONLY.

Calls xAI's subscription chat proxy using a grok.com subscription OAuth token instead of a metered XAI_API_KEY.

Credential sources (two, mutually exclusive)

1. Static token: supply GROK_OAUTH_ACCESS_TOKEN (or access_token=). The provider sends it verbatim and NEVER refreshes — the caller owns the token's lifecycle.
2. grok CLI store (auto-refresh): when no static token is given and a grok credential file exists, the provider auto-loads it via :class:~symfonic.core.grok_credentials.GrokCredentialStore and refreshes the short-lived (~6h) access token transparently — including mid-session, on every outgoing request. Use from_grok_cli() to request this path explicitly.

Side effect of the store path

Auto-refresh MUTATES the local grok credential file: the OAuth refresh_token rotates on every refresh, so the new value is written back (atomically, under a file lock, owner-only) or the developer's grok login would break. This is a deliberate, necessary side effect of the seamless-refresh posture and applies ONLY to the store path — the static-token path touches no files.

When to use

• Local development against Grok models without provisioning a separate metered XAI_API_KEY.
• Experimenting with the pipeline against a real LLM during development.

When NOT to use

• ❌ Production / multi-tenant SaaS — a subscription token doesn't bill per-tenant and its rate limits differ from API-key plans.
• ❌ Cost-tracking — subscription usage doesn't map to per-token pricing.

This provider explicitly refuses to work in production-like environments (SYMFONIC_ENV/ENVIRONMENT/ENV/NODE_ENV set to production/prod) to prevent accidental subscription billing under multi-tenant load, and allow_production=True is the sole reviewed escape hatch. This mirrors :class:~symfonic.core.codex_provider.CodexOAuthProvider and :class:~symfonic.core.kimi_oauth_provider.KimiOAuthProvider.

ToS caveat

Replaying a grok.com subscription token to xAI's chat proxy from a third-party program is subscription-auth reuse. xAI ships its OAuth login (auth.x.ai) for its own grok CLI — we deliberately do NOT run that handshake. The adopter runs grok login themselves; this provider only reuses the resulting local session (or an explicitly exported token) for dev/test, in the same risk class as the Kimi and Codex OAuth providers.

Credentials (environment variables)

• GROK_OAUTH_ACCESS_TOKEN (required for the static path) subscription access token, obtained by running grok login yourself.
• GROK_OAUTH_BASE_URL (optional) override the endpoint; defaults to the subscription chat proxy https://cli-chat-proxy.grok.com/v1 — the only endpoint a subscription OAuth token authenticates against (the metered api.x.ai platform endpoint rejects it with 401).
• GROK_OAUTH_CREDENTIALS_FILE (optional) override the credential file path for the store path (default: $GROK_HOME/auth.json, else ~/.grok/auth.json).

You obtain the token by signing in with grok yourself, e.g.::

grok login   # then just construct the provider; the store auto-loads

GrokCredentialError

Bases: RuntimeError

Raised when a Grok subscription token cannot be resolved or refreshed.

GrokCredentialStore

GrokCredentialStore(credentials_path: Path | None = None, *, token_url: str = _DEFAULT_TOKEN_URL, expiry_skew_seconds: int = 300)

Yields an always-valid grok CLI access token, refreshing as needed.

Parameters:

Name	Type	Description	Default
credentials_path	Path | None	Override the credential file. Defaults to the GROK_OAUTH_CREDENTIALS_FILE env var, then $GROK_HOME, then ~/.grok/auth.json.	None
token_url	str	OAuth token endpoint. Defaults to auth.x.ai.	_DEFAULT_TOKEN_URL
expiry_skew_seconds	int	Refresh this many seconds BEFORE the cached token's expires_at so an in-flight request never carries a just-expired token. Default 300s, matching the CLI's own GROK_AUTH_EARLY_INVALIDATION_SECS (token life is ~6h).	300

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

def __init__(
    self,
    credentials_path: Path | None = None,
    *,
    token_url: str = _DEFAULT_TOKEN_URL,
    expiry_skew_seconds: int = 300,
) -> None:
    self._path = (
        Path(credentials_path).expanduser()
        if credentials_path is not None
        else default_credentials_path()
    )
    self._lock_path = self._path.with_name(self._path.name + ".lock")
    self._token_url = token_url
    self._skew = expiry_skew_seconds

access_token

access_token() -> str

Return a valid access token, refreshing if within the skew window.

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

def access_token(self) -> str:
    """Return a valid access token, refreshing if within the skew window."""
    data, entry = self._load()
    if self._is_fresh(entry):
        token = _entry_key(entry)
        assert token is not None  # _load guarantees a keyed entry
        return token
    return self._refresh()

peek_access_token

peek_access_token() -> str | None

Return the cached token without refreshing; None if unavailable.

For logging / fingerprinting at construction time — must never raise or trigger a network refresh.

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

def peek_access_token(self) -> str | None:
    """Return the cached token without refreshing; ``None`` if unavailable.

    For logging / fingerprinting at construction time — must never raise or
    trigger a network refresh.
    """
    try:
        _, entry = self._load()
    except GrokCredentialError:
        return None
    return _entry_key(entry)

GrokOAuthProvider

GrokOAuthProvider(*, access_token: str | None = None, base_url: str | None = None, allow_production: bool = False, _store: GrokCredentialStore | None = None)

DEV/TEST provider that calls xAI via a subscription token.

Mirrors the OpenAI-compatible ChatOpenAI wiring shared by the other OAuth providers (same defaults, sampling parity, capability flags); the ONLY differences are auth (a grok.com subscription token rather than XAI_API_KEY), the chat-proxy endpoint, and the production guard.

Parameters:

Name	Type	Description	Default
access_token	str | None	Grok subscription access token. Defaults to the GROK_OAUTH_ACCESS_TOKEN env var.	None
base_url	str | None	xAI backend base URL. Defaults to GROK_OAUTH_BASE_URL then the subscription chat proxy.	None
allow_production	bool	Skip the production-env guard. Default False. Set True only if you have an explicit, ToS-reviewed reason.	False

Raises:

Type	Description
GrokCredentialError	No static token (arg/env) AND no grok CLI credential file to auto-load.
GrokProductionMisuseError	A production env var is set and allow_production=True was not passed. Fires regardless of credential source (env var, explicit access_token=, OR the auto-refreshing grok CLI store).

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

def __init__(
    self,
    *,
    access_token: str | None = None,
    base_url: str | None = None,
    allow_production: bool = False,
    _store: GrokCredentialStore | None = None,
) -> None:
    # Credential source resolution (mutually exclusive):
    #   1. explicit static token (arg or env) -> no refresh, caller owns it
    #   2. a GrokCredentialStore (from_grok_cli, or auto-detected) -> refresh
    # An explicit static token ALWAYS wins over auto-detection so
    # env-vars-only behavior is preserved. A store passed in via
    # ``_store`` (from ``from_grok_cli``) is an explicit opt-in and wins
    # over any ambient env token.
    store = _store
    static_token = None if store is not None else (
        access_token or os.environ.get(_ACCESS_TOKEN_ENV)
    )
    if store is None and static_token is None:
        cred_path = default_credentials_path()
        if cred_path.exists():
            store = GrokCredentialStore(cred_path)
    if store is None and static_token is None:
        msg = (
            f"No Grok credentials found. Either set ${_ACCESS_TOKEN_ENV} "
            "(export a token from `grok` login), pass access_token=..., "
            "or run `grok` login so the credential file exists "
            "(auto-loaded + auto-refreshed)."
        )
        raise GrokCredentialError(msg)

    # Refuse production regardless of credential source. Grok has no
    # multi-tenant resolver seam, so allow_production=True is the SOLE
    # escape. Fires for the static-token AND the store-backed path. Guards
    # accidental subscription billing under multi-tenant prod load.
    if _is_production_env() and not allow_production:
        msg = (
            "GrokOAuthProvider is for dev/test only. Detected a "
            "production environment (SYMFONIC_ENV/ENVIRONMENT/ENV/NODE_ENV). "
            "Use an xAI API-key provider with a metered XAI_API_KEY for "
            "production. Pass allow_production=True to override "
            "(not recommended)."
        )
        raise GrokProductionMisuseError(msg)

    self._store = store
    self._static_token = static_token
    self._base_url = (
        base_url or os.environ.get(_BASE_URL_ENV) or _DEFAULT_BASE_URL
    )

    logger.info(
        "GrokOAuthProvider initialized: source=%s token=%s base_url=%s",
        "grok-cli-store" if store is not None else "static-token",
        self.token_fingerprint,
        self._base_url,
    )

token_fingerprint property

token_fingerprint: str

Short, non-leaking identifier for the active token.

from_grok_cli classmethod

from_grok_cli(*, credentials_path: Any = None, base_url: str | None = None, allow_production: bool = False) -> GrokOAuthProvider

Build a provider backed by the grok CLI credential store.

Auto-loads ~/.grok/auth.json (or GROK_HOME / credentials_path / GROK_OAUTH_CREDENTIALS_FILE) and refreshes the short-lived access token transparently on every request — so a developer runs grok login ONCE and never exports or refreshes a token by hand. The production guard still fires on this path.

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

@classmethod
def from_grok_cli(
    cls,
    *,
    credentials_path: Any = None,
    base_url: str | None = None,
    allow_production: bool = False,
) -> GrokOAuthProvider:
    """Build a provider backed by the grok CLI credential store.

    Auto-loads ``~/.grok/auth.json`` (or ``GROK_HOME`` /
    ``credentials_path`` / ``GROK_OAUTH_CREDENTIALS_FILE``) and refreshes
    the short-lived access token transparently on every request — so a
    developer runs ``grok`` login ONCE and never exports or refreshes a
    token by hand. The production guard still fires on this path.
    """
    store = GrokCredentialStore(credentials_path)
    return cls(
        _store=store,
        base_url=base_url,
        allow_production=allow_production,
    )

get_chat_model

get_chat_model(config: ModelConfig) -> BaseChatModel

Return a ChatOpenAI pointed at the xAI chat proxy.

Static-token path: the token is passed as api_key — LangChain sends it as Authorization: Bearer <token> — verbatim, no refresh.

Store-backed path: api_key is a placeholder and a per-request httpx event hook rewrites Authorization with a FRESHLY-valid token on every outgoing request (sync + async), so a token that expires mid-session refreshes transparently without rebuilding the model.

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

def get_chat_model(self, config: ModelConfig) -> BaseChatModel:
    """Return a ``ChatOpenAI`` pointed at the xAI chat proxy.

    Static-token path: the token is passed as ``api_key`` — LangChain sends
    it as ``Authorization: Bearer <token>`` — verbatim, no refresh.

    Store-backed path: ``api_key`` is a placeholder and a per-request httpx
    event hook rewrites ``Authorization`` with a FRESHLY-valid token on
    every outgoing request (sync + async), so a token that expires
    mid-session refreshes transparently without rebuilding the model.
    """
    try:
        from langchain_openai import ChatOpenAI
    except ImportError as exc:
        raise ImportError(
            "langchain-openai is required for GrokOAuthProvider. "
            "Install it via: pip install symfonic-core[grok-oauth]",
        ) from exc

    headers = dict(config.extra_headers)
    # The subscription proxy gates chat completions on a CLI identity
    # header (HTTP 426 without one); declare the installed CLI's version.
    headers.setdefault("x-grok-client-version", _resolve_cli_version())
    kwargs: dict[str, Any] = {
        "model": config.model_name or _DEFAULT_MODEL,
        "temperature": config.temperature,
        "max_tokens": config.max_tokens,
        "base_url": self._base_url,
        "api_key": (
            _STORE_MANAGED_API_KEY
            if self._store is not None
            else self._static_token
        ),
        "default_headers": headers,
        # Emit a usage chunk at the end of every SSE stream so
        # input_tokens / output_tokens are non-zero on streamed calls.
        "stream_usage": True,
    }
    # Grok rides the OpenAI-compatible ChatOpenAI class, inheriting
    # ``timeout`` and ``http_async_client`` (mirrors KimiOAuthProvider).
    # This also threads any adopter-supplied ModelConfig.http_client; the
    # store-backed hook below composes onto (never clobbers) it.
    _thread_timeout_and_http_client(
        kwargs=kwargs,
        config=config,
        langchain_cls=ChatOpenAI,
        **_binding("grok-oauth"),
        provider_family="grok",
    )
    # Thread top_p (ChatOpenAI accepts it); top_k warns+drops (the OpenAI
    # chat API has no top_k parameter).
    _apply_sampling_kwargs(kwargs, config, ChatOpenAI, "grok")
    self._install_http_hooks(kwargs)
    return ChatOpenAI(**kwargs)

GrokProductionMisuseError

Bases: RuntimeError

Raised when GrokOAuthProvider is used in a production-like env.