symfonic.core.kimi_oauth_provider

kimi_oauth_provider

KimiOAuthProvider — DEV/TEST USE ONLY.

Calls the Moonshot/Kimi OpenAI-compatible backend using a kimi.com subscription OAuth token instead of a metered MOONSHOT_API_KEY.

Credential sources (two, mutually exclusive)

1. Static token (unchanged, backward compatible): supply KIMI_OAUTH_ACCESS_TOKEN (or access_token=). The provider sends it verbatim and NEVER refreshes — the caller owns the token's lifecycle.
2. kimi-cli store (auto-refresh): when no static token is given and a kimi-cli credential file exists, the provider auto-loads it via :class:~symfonic.core.kimi_credentials.KimiCredentialStore and refreshes the short-lived (~15 min) access token transparently — including mid-session, on every outgoing request. Use from_kimi_cli() to request this path explicitly. This DOES embed the kimi-code CLI's public device-flow client id (see kimi_credentials): a published identifier, not a secret, and refresh is impossible without it.

Side effect of the store path

Auto-refresh MUTATES the local kimi-cli credential file: the OAuth refresh_token rotates on every refresh, so the new value is written back (atomically, under a file lock) or the developer's kimi-cli 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 Kimi models without provisioning a separate metered MOONSHOT_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 the per-token pricing in :mod:symfonic.core.observability.pricing (the Kimi model ids resolve to pricing rows, but subscription billing is flat-rate).

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. Kimi is env-vars-only and dev/test-only, so the guard fires regardless of how the token is supplied and allow_production=True is the sole reviewed escape hatch. This mirrors :class:~symfonic.core.codex_provider.CodexOAuthProvider.

ToS caveat

Replaying a kimi.com subscription token to the Moonshot API from a third-party program is subscription-auth reuse. Moonshot ships a device-auth flow (auth.kimi.com) for its own kimi-cli under a fixed published client id — we deliberately do NOT embed that client identity or run the handshake. The adopter runs kimi-cli login themselves and exports the resulting token, keeping us in the same risk class as :class:~symfonic.core.codex_provider.CodexOAuthProvider — dev/test only.

Credentials (environment variables)

• KIMI_OAUTH_ACCESS_TOKEN (required) subscription access token, obtained by running kimi-cli login yourself.
• KIMI_OAUTH_BASE_URL (optional) override the endpoint; defaults to the kimi.com subscription (coding) endpoint https://api.kimi.com/coding/v1 — the only endpoint a subscription OAuth token authenticates against (the metered api.moonshot.ai platform endpoint rejects it with 401).
• KIMI_OAUTH_PLATFORM (optional) → X-Msh-Platform request header. Only set this if your kimi.com subscription token requires a platform header to be accepted. We do NOT default it to Moonshot's kimi_code_cli — supply your own value, matching the "we don't embed their client identity" principle.

You obtain the token by signing in with kimi-cli yourself and exporting the resulting value, e.g.::

export KIMI_OAUTH_ACCESS_TOKEN=...
# optional, only if your token requires it:
export KIMI_OAUTH_PLATFORM=...

KimiCredentialError

Bases: RuntimeError

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

KimiCredentialStore

KimiCredentialStore(
    credentials_path: Path | None = None,
    *,
    client_id: str = _KIMI_CODE_PUBLIC_CLIENT_ID,
    token_url: str = _DEFAULT_TOKEN_URL,
    expiry_skew_seconds: int = 120,
)

Yields an always-valid kimi-cli access token, refreshing as needed.

Parameters:

Name	Type	Description	Default
credentials_path	Path | None	Override the credential file. Defaults to the KIMI_OAUTH_CREDENTIALS_FILE env var, then ~/.kimi-code/credentials/kimi-code.json.	None
client_id	str	OAuth device-flow client id. Defaults to the kimi-code CLI's public client id.	_KIMI_CODE_PUBLIC_CLIENT_ID
token_url	str	OAuth token endpoint. Defaults to auth.kimi.com.	_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 120s (token life is ~15 min).	120

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

def __init__(
    self,
    credentials_path: Path | None = None,
    *,
    client_id: str = _KIMI_CODE_PUBLIC_CLIENT_ID,
    token_url: str = _DEFAULT_TOKEN_URL,
    expiry_skew_seconds: int = 120,
) -> 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._client_id = client_id
    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/kimi_credentials.py

def access_token(self) -> str:
    """Return a valid access token, refreshing if within the skew window."""
    creds = self._load()
    if self._is_fresh(creds):
        return str(creds["access_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/kimi_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:
        creds = self._load()
    except KimiCredentialError:
        return None
    token = creds.get("access_token")
    return str(token) if token else None

KimiOAuthProvider

KimiOAuthProvider(
    *,
    access_token: str | None = None,
    base_url: str | None = None,
    platform: str | None = None,
    allow_production: bool = False,
    _store: KimiCredentialStore | None = None,
)

DEV/TEST provider that calls Moonshot/Kimi via a subscription token.

Mirrors :class:~symfonic.core.providers.KimiProvider's OpenAI-compatible ChatOpenAI wiring (same Moonshot defaults, sampling parity, capability flags); the ONLY differences are auth (a kimi.com subscription token from an env var rather than MOONSHOT_API_KEY) and the production guard.

Parameters:

Name	Type	Description	Default
access_token	str | None	Kimi subscription access token. Defaults to the KIMI_OAUTH_ACCESS_TOKEN env var.	None
base_url	str | None	Moonshot backend base URL. Defaults to KIMI_OAUTH_BASE_URL then the international endpoint.	None
platform	str | None	Value for the X-Msh-Platform header. Defaults to KIMI_OAUTH_PLATFORM; when present it is sent so a subscription token that requires a platform header is accepted. Left unset (no header) by default — supply your own value; we do NOT embed Moonshot's kimi_code_cli client identity.	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
KimiCredentialError	No static token (arg/env) AND no kimi-cli credential file to auto-load.
KimiProductionMisuseError	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 kimi-cli store).

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

def __init__(
    self,
    *,
    access_token: str | None = None,
    base_url: str | None = None,
    platform: str | None = None,
    allow_production: bool = False,
    _store: KimiCredentialStore | None = None,
) -> None:
    # Credential source resolution (mutually exclusive):
    #   1. explicit static token (arg or env) -> no refresh, caller owns it
    #   2. a KimiCredentialStore (from_kimi_cli, or auto-detected) -> refresh
    # An explicit static token ALWAYS wins over auto-detection so today's
    # env-vars-only behavior is byte-for-byte preserved. A store passed in
    # via ``_store`` (from ``from_kimi_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 = KimiCredentialStore(cred_path)
    if store is None and static_token is None:
        msg = (
            f"No Kimi credentials found. Either set ${_ACCESS_TOKEN_ENV} "
            "(export a token from `kimi-cli` login), pass access_token=..., "
            "or run `kimi-cli` login so the credential file exists "
            "(auto-loaded + auto-refreshed)."
        )
        raise KimiCredentialError(msg)

    # Refuse production regardless of credential source. Kimi 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 = (
            "KimiOAuthProvider is for dev/test only. Detected a "
            "production environment (SYMFONIC_ENV/ENVIRONMENT/ENV/NODE_ENV). "
            "Use KimiProvider with a metered MOONSHOT_API_KEY for "
            "production. Pass allow_production=True to override "
            "(not recommended)."
        )
        raise KimiProductionMisuseError(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
    )
    self._platform = platform or os.environ.get(_PLATFORM_ENV)

    logger.info(
        "KimiOAuthProvider initialized: source=%s token=%s base_url=%s "
        "platform=%s",
        "kimi-cli-store" if store is not None else "static-token",
        self.token_fingerprint,
        self._base_url,
        self._platform or "<none>",
    )

token_fingerprint property

token_fingerprint: str

Short, non-leaking identifier for the active token.

from_kimi_cli classmethod

from_kimi_cli(
    *,
    credentials_path: Any = None,
    base_url: str | None = None,
    platform: str | None = None,
    allow_production: bool = False,
) -> KimiOAuthProvider

Build a provider backed by the kimi-cli credential store.

Auto-loads ~/.kimi-code/credentials/kimi-code.json (or credentials_path / KIMI_OAUTH_CREDENTIALS_FILE) and refreshes the short-lived access token transparently on every request — so a developer runs kimi-cli 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/kimi_oauth_provider.py

@classmethod
def from_kimi_cli(
    cls,
    *,
    credentials_path: Any = None,
    base_url: str | None = None,
    platform: str | None = None,
    allow_production: bool = False,
) -> KimiOAuthProvider:
    """Build a provider backed by the kimi-cli credential store.

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

get_chat_model

get_chat_model(config: ModelConfig) -> BaseChatModel

Return a ChatOpenAI pointed at the Moonshot/Kimi backend.

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/kimi_oauth_provider.py

def get_chat_model(self, config: ModelConfig) -> BaseChatModel:
    """Return a ``ChatOpenAI`` pointed at the Moonshot/Kimi backend.

    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 KimiOAuthProvider. "
            "Install it via: pip install symfonic-core[kimi]",
        ) from exc

    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": self._default_headers(config),
        # 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,
    }
    # Kimi rides the OpenAI-compatible ChatOpenAI class, inheriting
    # ``timeout`` and ``http_async_client`` (mirrors KimiProvider). 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,
        timeout_kwarg="timeout",
        http_client_kwarg="http_async_client",
        provider_family="kimi",
    )
    # Thread top_p (ChatOpenAI accepts it); top_k warns+drops (the OpenAI
    # chat API has no top_k parameter). Sampling parity with KimiProvider.
    _apply_sampling_kwargs(kwargs, config, ChatOpenAI, "kimi")
    self._install_http_hooks(kwargs)
    return ChatOpenAI(**kwargs)

KimiProductionMisuseError

Bases: RuntimeError

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