symfonic.core.codex_provider

codex_provider

CodexOAuthProvider — DEV/TEST USE ONLY.

Calls OpenAI's Codex backend using a ChatGPT Plus/Pro/Team subscription token instead of a metered OPENAI_API_KEY. Credentials are supplied via environment variables (the same shape openwiki writes); this provider does NOT run an interactive OAuth flow and does NOT read the Codex CLI's on-disk token store.

When to use

• Local development against Codex models without provisioning a separate metered OPENAI_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.

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. Unlike :class:~symfonic.core.oauth_provider.AnthropicOAuthProvider -- whose credentials_resolver is a real multi-tenant seam that must keep working in prod -- Codex 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.

ToS caveat

Replaying a ChatGPT-subscription token to the Codex backend from a third-party program is subscription-auth reuse. OpenAI ships "Sign in with ChatGPT" for its own Codex CLI, but programmatic reuse elsewhere may be throttled or disallowed under your plan. Treat this as the same risk class as :class:~symfonic.core.oauth_provider.AnthropicOAuthProvider — dev/test only.

Credentials (environment variables)

• OPENAI_CHATGPT_ACCESS_TOKEN (required) subscription access token.
• OPENAI_CHATGPT_ACCOUNT_ID (optional) → chatgpt-account-id request header, which the Codex backend uses to attribute usage to the right plan.
• OPENAI_CHATGPT_BASE_URL (optional) override the Codex backend base URL (defaults to the ChatGPT Codex endpoint).

You obtain these by signing in with the Codex CLI (codex login) or an openwiki-style flow and exporting the resulting values, e.g.::

export OPENAI_CHATGPT_ACCESS_TOKEN=...
export OPENAI_CHATGPT_ACCOUNT_ID=...

CodexCredentialError

Bases: RuntimeError

Raised when a ChatGPT subscription token cannot be resolved.

CodexOAuthProvider

CodexOAuthProvider(
    *,
    access_token: str | None = None,
    account_id: str | None = None,
    base_url: str | None = None,
    use_responses_api: bool = True,
    allow_production: bool = False,
)

DEV/TEST provider that calls the Codex backend via a ChatGPT token.

Parameters:

Name	Type	Description	Default
access_token	str | None	ChatGPT subscription access token. Defaults to the OPENAI_CHATGPT_ACCESS_TOKEN env var.	None
account_id	str | None	Plan/account id. Defaults to OPENAI_CHATGPT_ACCOUNT_ID; when present it is sent as the chatgpt-account-id header so the backend bills the right plan.	None
base_url	str | None	Codex backend base URL. Defaults to OPENAI_CHATGPT_BASE_URL then the ChatGPT Codex endpoint.	None
use_responses_api	bool	Route via the OpenAI Responses API (the Codex backend's protocol). Default True; set False only if you point base_url at a Chat-Completions-compatible gateway. On the default (Codex) path the backend rejects max_output_tokens, so config.max_tokens is NOT forwarded; it is honoured only on the use_responses_api=False gateway path.	True
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
CodexCredentialError	No access token supplied via arg or env.
CodexProductionMisuseError	A production env var is set and allow_production=True was not passed. Fires regardless of how the token is supplied (env var OR explicit access_token=).

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

def __init__(
    self,
    *,
    access_token: str | None = None,
    account_id: str | None = None,
    base_url: str | None = None,
    use_responses_api: bool = True,
    allow_production: bool = False,
) -> None:
    token = access_token or os.environ.get(_ACCESS_TOKEN_ENV)
    if not token:
        msg = (
            f"No ChatGPT subscription token found. Set "
            f"${_ACCESS_TOKEN_ENV} (sign in via `codex login` and export "
            f"the token), or pass access_token=..."
        )
        raise CodexCredentialError(msg)

    # Refuse production regardless of how the token was supplied.
    # AnthropicOAuthProvider skips its guard for an explicit token/
    # resolver because that resolver IS its reviewed multi-tenant seam;
    # Codex has no such seam — access_token= merely mirrors the same
    # OPENAI_CHATGPT_ACCESS_TOKEN env var and carries no extra deployment
    # intent — so allow_production=True is the SOLE escape. Guards
    # accidental subscription billing under multi-tenant prod load.
    if _is_production_env() and not allow_production:
        msg = (
            "CodexOAuthProvider is for dev/test only. Detected a "
            "production environment (SYMFONIC_ENV/ENVIRONMENT/ENV/NODE_ENV). "
            "Use OpenAIProvider with a metered OPENAI_API_KEY for "
            "production. Pass allow_production=True to override "
            "(not recommended)."
        )
        raise CodexProductionMisuseError(msg)

    self._access_token = token
    self._account_id = account_id or os.environ.get(_ACCOUNT_ID_ENV)
    self._base_url = (
        base_url or os.environ.get(_BASE_URL_ENV) or _DEFAULT_CODEX_BASE_URL
    )
    self._use_responses_api = use_responses_api

    logger.info(
        "CodexOAuthProvider initialized: token=%s account_id=%s base_url=%s",
        _token_fingerprint(self._access_token),
        self._account_id or "<none>",
        self._base_url,
    )

token_fingerprint property

token_fingerprint: str

Short, non-leaking identifier for the active token.

get_chat_model

get_chat_model(config: ModelConfig) -> BaseChatModel

Return a ChatOpenAI pointed at the Codex backend.

The subscription access token is passed as api_key — LangChain sends it as Authorization: Bearer <token> — alongside the chatgpt-account-id header and the Codex base URL.

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

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

    The subscription access token is passed as ``api_key`` — LangChain
    sends it as ``Authorization: Bearer <token>`` — alongside the
    ``chatgpt-account-id`` header and the Codex base URL.
    """
    try:
        from langchain_openai import ChatOpenAI
    except ImportError as exc:
        raise ImportError(
            "langchain-openai is required for CodexOAuthProvider. "
            "Install it via: pip install symfonic-core[openai]",
        ) from exc

    kwargs: dict[str, Any] = {
        "model": config.model_name,
        "temperature": config.temperature,
        "max_retries": config.max_retries,
        "base_url": self._base_url,
        "api_key": self._access_token,
        "use_responses_api": self._use_responses_api,
        "default_headers": self._default_headers(config),
        # Emit a usage chunk at the end of every SSE stream so
        # input/output token counts are non-zero on streamed calls.
        "stream_usage": True,
    }
    if self._use_responses_api:
        # Codex-backend wire requirements (verified live). Gated on the
        # Responses-API path only, so the documented non-Codex
        # Chat-Completions gateway override (``use_responses_api=False``)
        # is not saddled with Codex-specific fields a strict gateway may
        # reject:
        #  - store=False  -> backend rejects server-side storage (400);
        #                    Chat Completions already defaults it off.
        #  - streaming=True -> backend accepts streaming only (400);
        #                    ``ainvoke`` aggregates the stream.
        #  - max_tokens omitted -> backend rejects max_output_tokens (400).
        kwargs["store"] = False
        kwargs["streaming"] = True
    else:
        # Non-Codex gateway path: honour the caller's output cap.
        kwargs["max_tokens"] = config.max_tokens
    if config.timeout_seconds is not None:
        kwargs["timeout"] = config.timeout_seconds
    # The OpenAI chat API has no ``top_k``; only ``top_p`` is threaded.
    if config.top_p is not None:
        kwargs["top_p"] = config.top_p

    # Log-only auth-failure observability: a 401/403 at request time (an
    # expired/invalid subscription token) emits ONE actionable WARNING.
    # Never raises, never reads the body — control flow/streaming untouched.
    self._install_auth_failure_log_hook(kwargs)

    # Codex backend 400s on role:system; the subclass rewrites the produced
    # payload to role:developer. Gated on the Responses-API path only, so
    # the documented Chat-Completions gateway override
    # (use_responses_api=False) keeps plain system messages a strict
    # gateway may require — it returns a plain ChatOpenAI, unchanged.
    model_cls = _codex_chat_model_cls() if self._use_responses_api else ChatOpenAI
    return model_cls(**kwargs)

CodexProductionMisuseError

Bases: RuntimeError

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