scitex_genai.gateway

Structured provider gateways for external agent harnesses.

class scitex_genai.gateway.AnthropicStreamTranslator(requested_model, message_id=<factory>, started=False, stopped=False, next_index=0, open_blocks=<factory>, block_types=<factory>, input_tokens=0, output_tokens=0)[source]

Statefully translate Codex Responses SSE objects to Anthropic SSE.

requested_model: str

message_id: str

started: bool = False

stopped: bool = False

next_index: int = 0

open_blocks: dict[str, int]

block_types: dict[int, str]

input_tokens: int = 0

output_tokens: int = 0

translate(event)[source]

Return type:: list[str]

__init__(requested_model, message_id=<factory>, started=False, stopped=False, next_index=0, open_blocks=<factory>, block_types=<factory>, input_tokens=0, output_tokens=0)

class scitex_genai.gateway.CodexAccount(alias, credential, in_flight=0, last_used_at=0.0, cooldown_until=0.0, refresh_lock=<factory>, primary_used_percent=None, secondary_used_percent=None, usage_refreshed_at=0.0)[source]

Runtime scheduling state for one provider-qualified account.

alias: str

credential: CodexCredential

in_flight: int = 0

last_used_at: float = 0.0

cooldown_until: float = 0.0

refresh_lock: Lock

primary_used_percent: float | None = None

secondary_used_percent: float | None = None

usage_refreshed_at: float = 0.0

property usage_score: float

property qualified_id: str

__init__(alias, credential, in_flight=0, last_used_at=0.0, cooldown_until=0.0, refresh_lock=<factory>, primary_used_percent=None, secondary_used_percent=None, usage_refreshed_at=0.0)

class scitex_genai.gateway.CodexAccountPool(accounts, *, choose=None)[source]

Select accounts per session and rotate around temporary failures.

Every new session uses choose after quota and concurrent-load filtering. This includes a one-account pool: the selector receives a one-element list instead of bypassing rotation with a singleton shortcut.

__init__(accounts, *, choose=None)[source]

classmethod discover(homes=None)[source]

Return type:: CodexAccountPool

async acquire(session_id='', *, exclude=None)[source]

Return type:: CodexAccount

async release(account)[source]

Return type:: None

async cool_down(account, seconds)[source]

Return type:: None

async update_usage(account, *, primary_used_percent, secondary_used_percent)[source]

Return type:: None

class scitex_genai.gateway.CodexBackend(pool, transport, usage_client=None)[source]

Apply account scheduling and failover around CodexTransport.

__init__(pool, transport, usage_client=None)[source]

async refresh_usage()[source]

Return type:: None

async stream(payload, *, session_id='')[source]

Return type:: AsyncIterator[dict[str, Any]]

class scitex_genai.gateway.CodexCredential(path, access_token, refresh_token, account_id, expires_at, email='')[source]

One refreshable Codex subscription credential.

path: Path

access_token: str

refresh_token: str

account_id: str

expires_at: float

email: str = ''

classmethod load(path)[source]

Return type:: CodexCredential

needs_refresh(*, now=None, skew=300.0)[source]

Return type:: bool

async refresh(*, post=None)[source]

Refresh in place and atomically update the source auth.json.

Return type:: None

__init__(path, access_token, refresh_token, account_id, expires_at, email='')

class scitex_genai.gateway.CodexTransport(*, base_url='https://chatgpt.com/backend-api', client=None)[source]

Make raw Codex subscription requests; never run returned tools.

__init__(*, base_url='https://chatgpt.com/backend-api', client=None)[source]

async stream(payload, account, *, session_id='')[source]

Return type:: AsyncIterator[dict[str, Any]]

class scitex_genai.gateway.CodexUsageClient(client=None)[source]

Read quota windows without exposing tokens or account identifiers.

__init__(client=None)[source]

async fetch(account)[source]

Return type:: tuple[float | None, float | None]

async refresh_pool(pool)[source]

Return type:: None

scitex_genai.gateway.anthropic_to_codex(body, *, session_id='')[source]

Build a Codex Responses request without executing any tools.

Return type:: dict[str, Any]

scitex_genai.gateway.codex_events_to_anthropic(events, *, model)[source]

Collect Codex events into one non-streaming Anthropic message.

Return type:: dict[str, Any]

scitex_genai.gateway.create_app(backend, *, api_key=None)[source]

Create the FastAPI app without importing server dependencies at import time.

Return type:: Any