tokenrail

tokenrail is a small Python library for running OpenAI Responses API jobs with a client.responses-style surface.

It focuses on:

• thread-based OpenAI batch execution
• structured output parsing for Pydantic models
• client-side RPM / TPM submit throttling
• OpenAI prompt-cache settings passed through unchanged, with read/write usage tracking
• per-model token / cost monitoring with ETA progress reporting
• resumable JSONL result writing with the full API response

Fully typed (PEP 561), supports Python 3.10+.

The library is in beta. The public API is still evolving, and breaking changes may occur between 0.x minor versions.

Installation

uv add tokenrail
# or
pip install tokenrail

To track an unreleased revision instead, depend on the Git repository directly:

[tool.uv.sources]
tokenrail = { git = "https://github.com/takumi0shibata/tokenrail", branch = "main" }

Set your OpenAI API key in the consuming project before use:

export OPENAI_API_KEY=...

Quick start

from tokenrail import BatchExecutor, RailClient, batch_items_from_queries

client = RailClient.openai(max_retries=6)

queries = {
    "1": [{"role": "user", "content": "Summarize this paper in 3 bullets."}],
    "2": [{"role": "user", "content": "Extract the key assumptions."}],
}

items = batch_items_from_queries(
    queries,
    model="gpt-5.4-mini-2026-03-17",
    reasoning_effort="medium",
    verbosity="low",
)

executor = BatchExecutor(
    client=client,
    max_workers=16,
    max_rpm=500,
    max_tpm=200_000,
    output_path="out/results.jsonl",
)

stats = executor.run(items)
print(stats.to_dict())

Result saving

output_path is required for every batch. Parent directories are created if needed. Each completed request is appended as one JSONL record, in completion order rather than input order.

By default, records contain the full normalized response: output text, parsed output, refusal, usage, billing, estimated cost, timing, metadata, status, and error. The original API response body is preserved under raw_response when available. The top-level id is your query id; the API's response id is kept under raw_response.id.

To save selected fields, pass a projector directly:

executor = BatchExecutor(
    client=client,
    output_path="out/selected-results.jsonl",
    projector=lambda response: {
        "text": response.output_text,
        "usage": response.usage.to_dict(),
    },
)
stats = executor.run(items)

The projector receives a NormalizedResponse and must return a dictionary with JSON-serializable values. tokenrail always adds or overrides id, error, and status with the actual result, so field selection cannot break resume or hide failures. A projector replaces the default record; unselected fields, including raw_response, are omitted.

Prompt caching

tokenrail forwards OpenAI's prompt-cache settings and input content as supplied. It does not detect shared prefixes, insert breakpoints, generate or split keys, prewarm caches, or impose cache-specific submit limits. These settings work with both responses.create(...) and responses.parse(...), including batch items.

Omitting these settings keeps the OpenAI API defaults, including implicit caching. On GPT-5.6 and later, writing an eligible prefix costs 1.25 times the ordinary input rate, so a prefix that is never reused can cost more than an uncached input. Setting support and cache eligibility depend on the model.

Use the API defaults:

from tokenrail import RailClient

client = RailClient.openai()
response = client.responses.create(model="gpt-6.1-sol", input="Summarize this paper.")

Cache only a reusable prefix on GPT-5.6 and later. Put the shared content in its own block and mark its end; content after the breakpoint stays outside the cache write. The prefix must render to at least 1,024 visible input tokens. Top-level instructions cannot contain a breakpoint, so supply reusable instructions in a developer content block:

from tokenrail import BatchExecutor, batch_items_from_queries

shared_block = {
    "type": "input_text",
    "text": "Shared rubric, examples, and reference material ...",  # At least 1,024 tokens.
    "prompt_cache_breakpoint": {"mode": "explicit"},
}
items = batch_items_from_queries(
    {
        item_id: [
            {"role": "developer", "content": [shared_block]},
            {"role": "user", "content": question},
        ]
        for item_id, question in {
            "paper-1": "Summarize paper one.",
            "paper-2": "Summarize paper two.",
        }.items()
    },
    model="gpt-6.1-sol",
    prompt_cache_options={"mode": "explicit"},
)
stats = BatchExecutor(client=client, output_path="out/cached-results.jsonl", max_workers=16, max_rpm=120).run(items)
print(stats.cached_tokens)       # Tokens read from cache.
print(stats.cache_write_tokens)  # Tokens newly written to cache.

Avoid prompt-cache reads and writes on GPT-5.6 and later by using explicit mode with no breakpoints:

response = client.responses.create(
    model="gpt-6.1-sol",
    input="A one-off prompt ...",
    prompt_cache_options={"mode": "explicit"},
)

Choose keys, breakpoints, and reuse according to your workload. When supplying a key, keep it stable across related requests. tokenrail records cache read/write tokens and estimates their costs; it does not guarantee a hit or a saving. Conflicting values supplied in normal request arguments and extra_body raise ValueError before sending. store=False controls response storage, not prompt caching. See the OpenAI prompt caching guide for current model support, pricing, and cache lifetime.

Structured output batches

Pass a Pydantic model as text_format when building batch items. BatchExecutor will call responses.parse(...) for those items and store the validated object on response.output_parsed.

from pydantic import BaseModel

from tokenrail import BatchExecutor, RailClient, batch_items_from_queries


class PaperSummary(BaseModel):
    title: str
    key_assumptions: list[str]


client = RailClient.openai(max_retries=6)

items = batch_items_from_queries(
    {
        "paper-1": [{"role": "user", "content": "Extract the title and assumptions from this paper: ..."}],
        "paper-2": [{"role": "user", "content": "Extract the title and assumptions from this paper: ..."}],
    },
    model="gpt-5.4-mini-2026-03-17",
    reasoning_effort="medium",
    text_format=PaperSummary,
)

executor = BatchExecutor(
    client=client,
    output_path="out/structured-results.jsonl",
    max_workers=16,
    projector=lambda response: {
        "summary": response.output_parsed.model_dump(mode="json") if response.output_parsed else None,
        "refusal": response.refusal,
        "usage": response.usage.to_dict(),
    },
)

stats = executor.run(items)
print(stats.to_dict())

Use response_format={...} with client.responses.create(...) when you want to provide a raw JSON Schema yourself. Use text_format=YourModel for Pydantic parsing; response_format and text_format cannot be used together. Do not set verbosity on structured output batches that use text_format, because the OpenAI SDK's responses.parse(...) path does not accept that combination.

Configuration notes

• max_retries configures the OpenAI Python SDK client's built-in retry behavior. tokenrail does not add its own retry loop on top.
• Omitting timeout (or passing None) preserves the SDK's default timeout. Pass a number of seconds to override it.
• max_rpm and max_tpm are optional client-side submit limits. When a limit is set, BatchExecutor waits before submitting more work instead of raising its effective concurrency above the configured rate.
• TPM throttling estimates future usage from completed requests; it is not a hard API token limit. SDK retries are not separately counted by the submit limiter.
• Request failures are captured as error records (saved to the JSONL file and counted in stats) rather than raised, so one failing item does not abort the batch.
• API responses with a status other than completed, or with an API error, are counted as errors. Incomplete responses retain partial output and usage for diagnostics. background=True is rejected before sending because tokenrail does not poll background responses.
• With the OpenAI SDK, structured-output parsing failures return a response with error set and output_parsed=None, preserving the raw response, partial text, usage, and estimated cost. Check response.error before using parsed output. Capturing the raw response does not issue another API request; transport errors and invalid request arguments still raise on direct client calls.
• base_url is passed through to the OpenAI Python SDK for callers that need an SDK-level custom endpoint.

Resume behavior

BatchExecutor reads successful query ids from output_path before it starts. An id is skipped when a valid record has no error and its status is completed (or absent in legacy records). Failed, unfinished, and corrupt records are retried on re-running the batch. Existing JSONL files from earlier releases remain readable.

Use the same output path and query ids only when resuming the same inputs and request settings. Select a different path for a different batch. Resume does not provide exactly-once API execution: an interruption can cause another API charge and another record for the same id. Consumers should use the latest successful record for each id.

Projector, JSON serialization, and file-writing errors propagate from run(). Results that were not saved successfully remain eligible for retry. JSONL writes are flushed and synced to disk. A broken final line is repaired before appending; complete final records without a newline are preserved. Writes are coordinated within one executor; do not concurrently write the same destination from separate executors or processes.

The default record preserves partial output, usage, cost, and raw_response for failed requests, including local structured-output parsing failures. These failures remain retryable even when the API returned status="completed", because the top-level error records the parsing failure.

Batch ids must be non-empty and unique after conversion to strings. If an item supplies request_id, it must equal BatchItem.id. Empty ids, duplicate ids, and mismatches raise ValueError before any requests in that batch are sent. Query ids are stored as JSON values and are not used as file names.

Progress output

BatchExecutor automatically tracks statistics, keeps request-specific details short, and prints batch metrics separately:

tokenrail · 100 requests
   PAYER: openai — costs are covered
  0001  ok   req-0001      model=gpt-5.6-terra  1.3k tok (40% cached / 0% cache-write)  $0.002000  oai  1.4s
── 50/100 · 50% · 00:00:14 · ETA 00:00:14 · 58 rpm · 74k tpm · $0.100 (oai 100% / dev $0.000) · cache r40%/w2%
!! PAYER SWITCH: openai → developer at 53/100 (00:00:15) — now billed to you
  0053  ok   req-0053      1.2k tok (38% cached / 0% cache-write)  $0.001900  DEV  1.1s
Done 100/100 · 99 ok / 1 errors · 00:00:29
Total $0.198 — openai $0.104 (53%) / developer $0.094 (47%)
Prompt cache: 48k read / 2.4k written
Payer switches: 1

Periodic summaries are emitted every 50 completed requests or after 30 seconds when another result arrives. Payer transitions require three consecutive known payer observations, and ANSI colors are enabled automatically for a terminal.

Pass progress=False to disable all progress output:

stats = BatchExecutor(
    client=client,
    output_path="out/results.jsonl",
    progress=False,
).run(items)

Statistics are still collected and returned as a StatsSnapshot. Display intervals and formatting use built-in defaults.

Request lines show oai for openai, DEV for developer, and the original string for any other payer value. A missing or empty payer is shown as ?. The monitor reads the payer from cost, falling back to billing.payer when unavailable. Unknown payer values count toward unknown_payer_requests but do not drive payer transitions.

Cost tracking

Standard-tier base prices in USD per million tokens, checked against the official OpenAI pricing page on October 8, 2026:

GPT-5.6 Sol uses the current promotional rate, available at least through November 21, 2026 according to OpenAI.

• GPT-5.6 and GPT-6 cache writes are estimated at 1.25 times each model's ordinary input rate and are tracked separately from cache reads.
• Costs use the checked-in base, standard-tier pricing table (tokenrail.catalog). The estimate does not apply GPT-5.6/GPT-6 long-context rates above 272K input tokens, Batch/Fast/Flex/Ultrafast pricing, or regional uplifts. The official pricing page is authoritative.
• Models without a pricing entry get cost=None. If an unregistered model partially matches an older catalog entry, tokenrail emits ModelCatalogFallbackWarning once per model and names the capability and pricing entries used as fallbacks.
• OpenAI cost allocation is inferred from billing.payer in the response body. When payer == "openai", the nominal request cost is counted as OpenAI-covered rather than developer-billed.
• reasoning_effort is gated to supported gpt-5, gpt-6, and o-series models in the checked-in capability registry.
• GPT-6 Astra and GPT-6.1 Sol support low, medium, high, xhigh, and max reasoning efforts; temperature and top_p are rejected before sending a request. GPT-6 Sol and Luna additionally support none, which must be explicitly selected to use temperature or top_p. These checks apply to both responses.create(...) and responses.parse(...). See the GPT-6 guide.

Fallback warnings can be filtered with Python's standard warning controls:

import warnings

from tokenrail.catalog import ModelCatalogFallbackWarning

warnings.filterwarnings("ignore", category=ModelCatalogFallbackWarning)

Development

uv sync
uv run pytest
uv run ruff check src tests

Live API checks are opt-in and spend tokens. Set OPENAI_API_KEY in the environment or a local .env. Start with the minimal payer probe; this first request may be billed to you:

TOKENRAIL_RUN_LIVE_TESTS=1 uv run pytest tests/test_live_openai.py::test_live_payer_probe -q -s

The remaining tests require TOKENRAIL_LIVE_FREE_TOKENS_REMAINING to contain a verified remaining balance for the applicable organization-wide model group. Check the current program terms, project eligibility, and usage dashboard, and pause other consumers of that shared quota before supplying this value. Leave it unset when the balance is unknown: only the initial probe runs, and the five remaining tests are skipped. Then, with a verified balance, run TOKENRAIL_RUN_LIVE_TESTS=1 uv run pytest tests/test_live_openai.py -q -s. Every invocation includes a new probe and consumes a new budget.

These checks use gpt-6-luna, capped outputs, and no SDK retries. Generation requests are serialized even when batch workers run concurrently. Before each generation, the transport calls the input-token count endpoint with the actual input and structured schema, then reserves input tokens plus the output cap against both a 100,000-token total ceiling and the verified free-token balance. Reservations are never refunded. The output cap includes reasoning tokens; cache and reasoning details are not added again to total usage.

A payer other than openai, unknown payer, missing or inconsistent usage, preflight failure, or transport failure blocks all subsequent sends. A response with payer openai confirms only that request's allocation; it does not promise future free usage. The response that reveals a payer switch may already be billed, and the caller-supplied balance cannot prevent other processes from consuming the shared quota. The normal test command skips all live checks. During 0.5.0 validation, the original five live checks used 7,719 tokens over ten requests; that historical result does not establish today's remaining balance.

License

MIT