Metadata-Version: 2.4
Name: openapi-ai-sdk
Version: 0.7.0
Summary: Python SDK for openapi.ai — a drop-in extension of the OpenAI client, plus capability contracts: schema-frozen interfaces called like functions.
Project-URL: Homepage, https://openapi.ai/docs/sdk/python
Project-URL: Repository, https://github.com/sireto/openapi-ai-sdk
Project-URL: Issues, https://github.com/sireto/openapi-ai-sdk/issues
Project-URL: Documentation, https://openapi.ai/docs/sdk
License-Expression: Apache-2.0
License-File: LICENSE
Requires-Python: >=3.10
Requires-Dist: httpx==0.28.1
Requires-Dist: openai==2.9.0
Requires-Dist: pydantic==2.13.4
Provides-Extra: dev
Requires-Dist: black==26.5.1; extra == 'dev'
Requires-Dist: jsonschema>=4.18; extra == 'dev'
Requires-Dist: pytest-asyncio==1.4.0; extra == 'dev'
Requires-Dist: pytest==9.1.1; extra == 'dev'
Requires-Dist: ruff==0.14.5; extra == 'dev'
Provides-Extra: postgres
Requires-Dist: psycopg[binary]==3.2.*; extra == 'postgres'
Provides-Extra: testing
Requires-Dist: jsonschema>=4.18; extra == 'testing'
Description-Content-Type: text/markdown

# openapi-ai-sdk

[![PyPI](https://img.shields.io/pypi/v/openapi-ai-sdk)](https://pypi.org/project/openapi-ai-sdk/)
[![Python](https://img.shields.io/pypi/pyversions/openapi-ai-sdk)](https://pypi.org/project/openapi-ai-sdk/)
[![License](https://img.shields.io/badge/license-Apache--2.0-blue)](./LICENSE)

The Python SDK for [OpenAPI](https://openapi.ai) — 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.

```bash
pip install openapi-ai-sdk
```

## Drop-in for the OpenAI SDK

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

```python
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.

```python
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:

```python
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:**

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

### 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:

```python
# 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:

```sh
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:

```python
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:

```python
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:

```python
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](https://openapi.ai/docs/capabilities/invoking).

### Static types, generated from the contract

```bash
openapi pull doc-summary@3
```

writes `doc_summary_v3.py`: `TypedDict`s 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:

```python
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]"`):

```python
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:

```python
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:

```python
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](../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:

| 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 |
| `extra_body(...)` | merges the above into one `extra_body` (alias: `routing`) |
| `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 |

```python
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

```python
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:

```bash
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](https://openapi.ai/docs/capabilities) — concepts and the invocation lifecycle
- [Invoking capabilities](https://openapi.ai/docs/capabilities/invoking) — sync, async, tools, MCP, generated types
- [Tutorials](https://openapi.ai/docs/tutorials/first-capability) — your first capability, end to end

## Development

```bash
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](./LICENSE) — 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:

```sh
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.
