openapi-ai-sdk 0.1.0

OpenAI-compatible client for the OpenAPI Router gateway, with routing, attribution and usage.

openapi-ai-sdk (Python)

pip install openapi-ai-sdk      # installs the openapi_ai module

from openapi_ai import OpenAPI, provider, metadata, routing, session

client = OpenAPI(api_key="gw_…", metadata={"env": "prod"})

with session("conversation-42"):
    response = client.chat.completions.create(
        model="openai/gpt-4.1-mini",
        messages=[{"role": "user", "content": "hello"}],
        extra_body=routing(
            provider(order=["groq"], sort="throughput"),
            metadata(tenant="acme"),
        ),
    )

print(response.usage.cost)                            # USD, on every response
client.usage.generations(metadata={"tenant": "acme"})  # what that tenant spent

OpenAPI is an openai.OpenAI — a subclass that overrides nothing, so migrating is one import and one name. Streaming, tool calls, retries and typed responses are the OpenAI SDK's own, and anything built on that SDK — LangChain, LlamaIndex, Instructor — takes this client unchanged. What the platform adds appears as namespaces on the same client: capabilities, invocations, callbacks, usage. There is an AsyncOpenAPI twin over openai.AsyncOpenAI, and router(...) remains as a factory spelling of the same client.

What is ours

provider(...)	which provider serves the request: order, only, ignore, sort, data_collection
models(...)	fallback models, tried in order
transforms(...)	middle-out prompt compression
metadata(...)	labels recorded with the request and filterable afterwards
routing(...)	merges the above into one extra_body
session(...)	attributes every request in the block to one conversation
generation, generations, credits, key_info	what it cost, and what a set of calls cost
client.capabilities, client.invocations	call capability contracts, sync or async — see below
client.callbacks.verify(...)	constant-time verification of signed result webhooks
client.usage	generation, generations, credits, key_info, namespaced

One deployment, https://openapi.ai/v1, so there is nothing to configure. Point base_url somewhere else for a local stack. Where inference physically runs is a routing choice per request — provider(jurisdictions=["EU"]) — not a hostname.

Recording what you spent

from openapi_ai import router, SqliteStore, summary

store = SqliteStore("~/.local/state/openapi-ai/calls.db")
client = router(api_key="gw_…", store=store)

client.chat.completions.create(..., extra_body=metadata(tenant="acme"))
summary(store, metadata={"tenant": "acme"})
# {'calls': 12, 'billable': 11, 'cached': 1, 'errors': 0, 'cost': 0.0241, ...}

Content is not stored by default. Ids, model, routing, tokens, cost, latency and your labels are; prompts and completions need store_io=True. This mirrors the server, where I/O capture is opt-in per key and off by default — a client that wrote transcripts to disk by default would invert that decision onto a machine nobody reviewed. A request sent with data_collection="deny" never has its content stored at all, whatever the flag says.

Backends: SqliteStore (recommended), JsonFileStore (grep-able, single process), PostgresStore (shared; needs the postgres extra, and makes whoever runs it a data controller for what it holds).

Caching

store = SqliteStore("calls.db", store_io=True)   # a cache needs a body to replay
client = router(api_key="gw_…", store=store, cache=True)

Non-streaming only, mirroring the gateway's own refusal to cache streams. Note what a client-side hit skips: it never reaches the gateway, so credits, free-tier counters, budgets, guardrails and rate limits see nothing. That is why it is opt-in.

The cache key covers everything that changes the answer — including provider preferences, model suffixes like :online, and data_collection — and excludes what does not, so labelling a request does not give it its own cache entry. It also mixes in a fingerprint of your API key, so a shared store cannot serve one organization's response to another.

Capabilities

Capability contracts invert the usual integration: your organization defines a named, versioned interface — an input schema and an output schema — providers register endpoints that fulfill it, and the platform validates both directions and retries across fulfillments. You integrate against the contract, never a provider API.

summarize = client.capabilities.get("doc-summary", version=3)

# Synchronous: the handle is callable; blocks until a conforming result.
result = summarize(input={"text": "..."})

# Asynchronous: a live handle now, the result when the work finishes.
job = summarize.submit(input={"text": "..."})
print(job.wait().output["summary"])

Binding name and version once is the point: published contracts are frozen, callers integrate against exactly one schema pair, and neither the gateway nor the SDK invents a floating "latest". Three unhappy endings, kept apart on purpose:

• a refusal (nonconforming input, the in-flight cap, exhausted fulfillments) raises CapabilityInvocationError at the call site, carrying the typed code, the JSON-pointer paths, and the attempt trail;
• a failed invocation is a terminal outcome the platform already retried — job.failed and job.error read it as data, while job.output raises InvocationFailed so the happy path cannot quietly read a None;
• a timeout in wait() raises InvocationTimeout and changes nothing: the job keeps running server-side.

Long-running work can skip polling: pass callback_url= to submit, then verify what arrives — client.callbacks.verify(raw_body, signature_header, secret="whsec_…") does the constant-time, tolerance-checked HMAC and returns the parsed document. AsyncOpenAPI serves the same namespaces with await.

Capabilities as model tools

tools = [client.capabilities.get("doc-summary", version=3).as_tool()]
# … model answers with tool_calls …
messages.append(client.capabilities.execute_tool_call(call))

A contract is already a tool definition; execute_tool_call runs the call through the platform and answers with the tool message. Model-caused failures (hallucinated arguments, nonconforming input) become tool-message content, never exceptions — an agent loop that crashes on a bad model turn is a loop nobody can run.

Typed contracts, generated

A published contract is a frozen pair of JSON Schemas — a type registry the platform already runs. Pull one into your codebase:

openapi-ai pull doc-summary@3

writes doc_summary_v3.py: TypedDicts for both sides plus invoke/submit helpers typed end to end, so the type checker holds your calls to the same schema pair the provider is held to at runtime. Commit the file; regenerate only to bind a different version — the contract cannot change under you, because published versions are frozen. The generator handles the common schema subset and degrades to Any (never a crash) for constructs the platform validates but a TypedDict cannot express. The key comes from --api-key or OPENAPI_AI_API_KEY.

Fulfilling capabilities (providers)

The other side of the contract, for provider teams: your endpoint is your own server, and the SDK provides the pieces every endpoint needs — the constant-time bearer check, the parsed job, completion with the platform's answers surfaced as typed outcomes, and the pull-mode shapes spelled once.

from openapi_ai import DispatchedJob, verify_dispatch, push_result, push_failure

verify_dispatch(request.headers.get("Authorization"), secret=DISPATCH_SECRET)
job = DispatchedJob.parse(await request.body())     # input already conforms
push_result(job.completion_url, {"summary": "…"})   # or push_failure(...)

A nonconforming result raises CompletionRejected with the JSON-pointer paths — and the token is burned; rejected.gone marks a consumed or expired token. An honest push_failure is authoritative by default: the platform does not shop the same input to another provider.

Development

python3 -m venv .venv && .venv/bin/pip install -e ".[dev]"
PYTHONPATH=. .venv/bin/python -m pytest tests/

tests/test_matches_the_gateway.py reads the gateway source and the committed spec, so a value this package validates locally cannot silently drift from what the server actually accepts.