openapi-ai-sdk 0.1.1

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

openapi-ai-sdk

The Python SDK for OpenAPI — a drop-in replacement for the OpenAI client, plus the one thing no other gateway offers: capability contracts, versioned schema-frozen interfaces you call like functions while the platform guarantees what goes in and what comes out.

pip install openapi-ai-sdk

Drop-in for the OpenAI SDK

Change one import and one class name; everything else is your existing code.

from openapi_ai import OpenAPI   # was: from openai import OpenAI

client = OpenAPI(api_key="gw_…")

response = client.chat.completions.create(
    model="openai/gpt-4.1-mini",
    messages=[{"role": "user", "content": "hello"}],
)

print(response.usage.cost)   # USD, on every response

OpenAPI is an openai.OpenAI — a subclass that overrides nothing. 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. AsyncOpenAPI is the same over openai.AsyncOpenAI. What the platform adds appears as namespaces on the same client: capabilities, invocations, callbacks, usage.

Capability contracts

The reason this SDK exists. Instead of integrating a provider's API, your organization defines a contract — a named, versioned pair of JSON Schemas — and providers register endpoints that fulfill it. The platform validates input before dispatch and output before you see it, retries across fulfillments, and never returns nonconforming data. Published versions are frozen forever: the schema you integrate against today cannot change under you.

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

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

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

version is required on purpose: there is no floating "latest", so publishing v4 breaks nobody on v3.

From Pydantic models, typed end to end

Define the contract and the call site from one pair of models:

from pydantic import BaseModel
from openapi_ai import capability_contract

class DocSummaryIn(BaseModel):
    text: str
    max_words: int | None = None

class DocSummaryOut(BaseModel):
    summary: str

capability_contract(DocSummaryIn, DocSummaryOut)
# → {"input_schema": …, "output_schema": …} — publish this as the contract version

summarize = client.capabilities.get(
    "doc-summary", version=3,
    input_model=DocSummaryIn, output_model=DocSummaryOut,
)

out = summarize(DocSummaryIn(text="long document"))   # → DocSummaryOut
out.summary                                           # attributes, not dict keys

job = summarize.submit(DocSummaryIn(text="long document"))
job.wait().output                                     # → DocSummaryOut

Input is validated client-side before anything leaves the process (a dict is accepted too, and validated through the model); output comes back as the model the platform already guaranteed conformance for. What you publish and what you type-check are one definition, not two hand-maintained copies.

Failures are typed, and kept deliberately distinct:

What happened	How you see it
Refusal — nonconforming input, in-flight cap, exhausted fulfillments	CapabilityInvocationError at the call site, with the typed code, JSON-pointer paths, and attempt trail
Failed invocation — terminal, already retried by the platform	Data: job.failed, job.error; reading job.output raises InvocationFailed so the happy path can't quietly get None
wait() timeout	InvocationTimeout — changes nothing; the job keeps running server-side

For long-running work, skip polling: pass callback_url= to submit, then verify the signed webhook that arrives — client.callbacks.verify(raw_body, signature_header, secret="whsec_…") does the constant-time, tolerance-checked HMAC and returns the full document.

Contracts as tools, for any model

A published contract already is a tool definition:

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

execute_tool_call runs the model's call through the platform and answers with the role:"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. The platform can also run the loop server-side, or serve your contracts over MCP — see invoking capabilities.

Static types, generated from the contract

openapi-ai pull doc-summary@3

writes doc_summary_v3.py: TypedDicts for both sides plus typed invoke/submit helpers, so your type checker enforces the same schema pair the provider is held to at runtime. Commit the file — it cannot rot, because published versions are frozen. Constructs a TypedDict cannot express degrade to Any, never to a crash. The key comes from --api-key or OPENAPI_AI_API_KEY.

Fulfilling contracts (the provider side)

Your endpoint is your own server; the SDK ships the pieces every fulfiller needs:

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 JSON-pointer paths — and the one-time completion token is burned (rejected.gone marks a consumed or expired one). An honest push_failure is authoritative by default: the platform does not shop the same input to another provider.

Routing, attribution, and usage

Everything below rides extra_body on standard OpenAI calls:

Helper	What it does
provider(...)	which provider serves the request: order, only, ignore, sort, jurisdictions, data_collection
models(...)	fallback models, tried in order
transforms(...)	middle-out prompt compression
metadata(...)	labels recorded with the request, filterable afterwards
routing(...)	merges the above into one extra_body
session_id(...)	attributes this request to a conversation, right on the call
session(...)	the same, for every request in a with block
client.usage	generation, generations, credits, key_info — what it cost, per call or per label

from openapi_ai import provider, metadata, routing, session_id

client.chat.completions.create(
    model="openai/gpt-4.1-mini",
    messages=[...],
    extra_body=routing(provider(order=["groq"], sort="throughput"),
                       metadata(tenant="acme"),
                       session_id("conversation-42")),
)

client.usage.generations(metadata={"tenant": "acme"})   # that tenant's spend

The default deployment is https://openapi.ai/v1; point base_url at your own stack to self-host. Where inference physically runs is a per-request routing choice (provider(jurisdictions=["EU"])), not a hostname.

Local call log and cache

from openapi_ai import OpenAPI, SqliteStore, summary

store = SqliteStore("~/.local/state/openapi-ai/calls.db", store_io=True)
client = OpenAPI(api_key="gw_…", store=store, cache=True)   # cache needs bodies to replay

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

Content is never stored by default — ids, model, routing, tokens, cost, latency, and your labels are; prompts and completions require an explicit store_io=True, mirroring the server's own opt-in stance. A request sent with data_collection="deny" never has its content stored at all. The cache (opt-in, non-streaming, needs store_io=True) keys on everything that changes the answer and fingerprints your API key, so a shared store cannot serve one organization's response to another — and note that a client-side hit never reaches the gateway, so budgets, guardrails, and rate limits see nothing. Backends: SqliteStore (recommended), JsonFileStore, PostgresStore (the postgres extra).

Documentation

• Capability contracts — concepts and the invocation lifecycle
• Invoking capabilities — sync, async, tools, MCP, generated types
• Tutorials — your first capability, end to end

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 accepts.

License

Apache-2.0 — use it against the hosted platform or any self-hosted deployment.