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. What the platform adds appears as namespaces on the same client: capabilities, invocations, callbacks, usage, file_store.

AsyncOpenAPI is the same over openai.AsyncOpenAI — the same namespaces, awaited: registering contracts, typed handles bound to Pydantic models, slot-aware file routing, the file store, usage queries, capabilities.execute_tool_call, and invocations.iterate (an async generator) all work exactly as on the sync client. The one documented exception is local recording (store=/cache=), which stays sync-only.

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

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

class DocSummaryOut(BaseModel):
    summary: str

# Publish the contract from the models — defines, drafts and freezes, then
# hands back a handle bound to what it just published.
summarize = client.capabilities.register(
    "doc-summary", input=DocSummaryIn, output=DocSummaryOut
)

# Or bind a version somebody already published.
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:

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.

Registering, staged or immediate

register is the mirror of get: it publishes the version and binds it, so what comes back is invocable on the next line. The capability is defined only when it does not already exist, so a deploy step is safe to re-run — the second run publishes v2 rather than failing on a name that is taken.

Publishing freezes both schemas forever, so there are two ways to stage:

# a handle you keep, published later from the console
summarize = client.capabilities.register(
    "doc-summary", input=DocSummaryIn, output=DocSummaryOut, publish=False
)

# the version document, to read before freezing it
version = client.capabilities.draft("doc-summary", input=DocSummaryIn, output=DocSummaryOut)
client.capabilities.publish_version("doc-summary", version["version"])

input/output take models or raw JSON Schema, and from a shell:

openapi publish doc-summary --input in.json --output out.json [--draft]
openapi capabilities             # every capability in the organization
openapi versions doc-summary     # its history, drafts included

The same listings in code are client.capabilities.list() and client.capabilities.versions("doc-summary") — what the console shows, without leaving the editor.

Registering is management, not inference: it authenticates with a provisioning key (pk_…) carrying the contracts scope, not the gw_ key the same client makes requests with. Pass provisioning_key= or set OPENAPI_PROVISIONING_KEY — and a publish job can hold only that: OpenAPI(provisioning_key="pk_…") constructs without an inference key, and the first gateway-bound call raises MissingApiKey instead of quietly needing a credential the process was promised not to hold. A key scoped to contracts alone cannot mint credentials that spend your credits — which is what a CI job should hold.

Files in, files out

A file slot is part of the contract like any other field — declare it with File and the schema, the encoding, and the decoding all fall out of the one annotation:

from pathlib import Path
from typing import Annotated
from openapi_ai import File, MediaType, capability_contract

class RedactIn(BaseModel):
    document: Annotated[File, MediaType("application/pdf")]
    reason: str

class RedactOut(BaseModel):
    redacted: Annotated[File, MediaType("application/pdf")]

capability_contract(RedactIn, RedactOut)   # file slots included — publish as usual

redact = client.capabilities.get(
    "doc-redact", version=1, input_model=RedactIn, output_model=RedactOut,
)
out = redact(RedactIn(document=Path("contract.pdf"), reason="pii"))
out.redacted.save("contract-redacted.pdf")   # .bytes and .media_type too

Untyped calls are just as direct — a Path or bytes value encodes itself, and output file slots come back as File objects on a bound handle:

result = client.capabilities.get("doc-redact", version=1)(
    input={"document": Path("contract.pdf"), "reason": "pii"}
)
result["output"]["redacted"].save("contract-redacted.pdf")

The contract decides how the bytes travel, and the code above does not change: inline slots (contentEncoding: "base64") ride the payload, bounded by the payload cap (10 MB default); by-reference slots (format: "file-id") upload to the platform file store first and the payload carries only the id — outputs download back into the same File objects. The store itself is client.file_store when you want an id without an invocation.

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 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_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_async

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

A malformed dispatch raises DispatchParseError (a ValueError) naming the missing field — answer it with a 400; your own bugs stay 500s. When rotating the dispatch secret, pass secrets=[NEW, OLD] and both stay valid through the deploy, with no timing leak about which one matched.

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.

Rehearse before the platform judges you. Your first nonconforming output should fail a test, not burn a production attempt (pip install "openapi-ai-sdk[testing]"):

from openapi_ai.testing import ContractCheck, MockDispatcher

check = ContractCheck.from_file("contract.json")   # openapi contract doc-summary@3 > contract.json
mock = MockDispatcher(check, secret="fs_test")

dispatch = mock.dispatch(input={"text": "hello"})  # authorized, pre-validated, like the platform's
response = my_endpoint(dispatch.body, dispatch.headers)      # your real handler
push_result(job.completion_url, output, transport=mock.transport)
assert mock.completed[0].output == output

The mock answers completions exactly as the platform does — 200 on conforming output, the 422 nonconforming_output envelope with paths, 410 once the token is spent — so your CompletionRejected handling is exercised too. check.assert_output(result) is the one-line version, and openapi conform doc-summary@3 --output @sample.json (or --contract contract.json offline) is the same verdict from CI. Validation is pinned to JSON Schema Draft 2020-12, the draft the gateway itself validates with.

Run your inference through the credential the dispatch carries. When the contract sponsors platform inference, the job arrives with a gateway credential scoped to that attempt — the enterprise's requirements (allowed model providers, jurisdictions, retention) are enforced on it, its cost is theirs, and the work is attributable:

client = job.client()          # already bound to this attempt's credential
client.chat.completions.create(model="…", messages=[…])

Pull mode, for providers that cannot make outbound calls: acknowledge the dispatch with your job id, then answer status polls — the platform GETs your registered status URL template (…/jobs/{job_id}/status) with the same bearer secret, on an exponential backoff until the attempt's deadline:

from openapi_ai import pull_ack, pull_pending, pull_result

@app.post("/dispatch")                       # completion.url is null in pull mode
async def dispatch(request: Request):
    verify_dispatch(request.headers.get("Authorization"), secret=DISPATCH_SECRET)
    job = DispatchedJob.parse(await request.body())
    job_id = queue(job)
    return pull_ack(job_id)

@app.get("/jobs/{job_id}/status")            # the registered status URL template
async def status(job_id: str, request: Request):
    verify_dispatch(request.headers.get("Authorization"), secret=DISPATCH_SECRET)
    result = finished.get(job_id)
    return pull_result(result) if result is not None else pull_pending()

The full wire contract — poll cadence, terminal-read-once semantics, what counts as pending — is vendored in docs/fulfillment-protocol.md.

That is the whole integration. A contract without sponsorship dispatches no credential and job.client() says so (NoGatewayCredential) rather than handing you something that would be refused.

Routing, attribution, and usage

Everything below rides extra_body on standard OpenAI calls:

from openapi_ai import provider, metadata, extra_body, session_id

client.chat.completions.create(
    model="openai/gpt-4.1-mini",
    messages=[...],
    extra_body=extra_body(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).

The openapi CLI

Installed with the package — the same surface, from a shell:

openapi chat "say hello" --model openai/gpt-4.1-mini --session c42
openapi invoke doc-summary@3 --input '{"text": "..."}'
openapi invoke doc-redact@1 --file document=contract.pdf --save redacted=clean.pdf
openapi submit doc-summary@3 --input @input.json --wait
openapi invocation inv-… --wait
openapi contract doc-summary@3
openapi credits
openapi key
openapi pull doc-summary@3

JSON on stdout for machines, the assistant's words for chat, refusals on stderr with the typed code and paths. The key comes from OPENAPI_API_KEY (or --api-key), the base URL from OPENAPI_BASE_URL.

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.

Naming a contract across organizations

A capability name belongs to the organization that published it, not to the platform: two organizations may both have a doc-summary, against entirely different schemas. Every command accepts the qualified form when you want to be explicit about which one you mean:

openapi invoke acme/doc-summary@3 --input '{"text": "..."}'
openapi pull acme/doc-summary@3        # writes acme/doc_summary_v3.py

The organization is checked, never used to look anything up — your API key decides which organization a call reaches. Naming a different one is refused before the request is sent, rather than quietly invoking your own contract of the same name. Unqualified references behave exactly as before, including where pull writes.