opperai 2.0.0b4

Python SDK for the Opper Task API

Opper Python SDK

Python client for the Opper API.

Install

pip install opperai

Quick Start

from opperai import Opper

opper = Opper()  # uses OPPER_API_KEY env var

result = opper.call("summarize", input={"text": "Long article..."})
print(result.data)

# Stream a function
for chunk in opper.stream("summarize", input={"text": "Long article..."}):
    if chunk.type == "content":
        print(chunk.delta, end="")
    if chunk.type == "complete":
        print(chunk.data)

Schema Support

Pass Pydantic models, dataclasses, TypedDicts, or raw JSON Schema dicts for input_schema and output_schema — the SDK resolves them to JSON Schema automatically.

from pydantic import BaseModel

class Summary(BaseModel):
    summary: str
    entities: list[str]

result = opper.call(
    "extract",
    input={"text": "Marie Curie was a physicist in Paris."},
    output_schema=Summary,
)
result.data.summary   # str — typed!
result.data.entities  # list[str]

Dataclasses, TypedDicts, and plain dicts also work. See 01a_using_schemas.py and 01b_using_other_schemas.py.

Observability

Use trace() as a decorator or context manager to group calls under a single trace span. Nesting works naturally.

@opper.trace("my-pipeline")
def run():
    a = opper.call("step-1", input="hello")
    b = opper.call("step-2", input=a.data)

# or as a context manager
with opper.trace("my-pipeline") as span:
    opper.call("step-1", input="hello")

Examples

#	Example	What it shows
00	First call	Simplest possible call
01a	Pydantic schemas	Type-safe output with Pydantic
01b	Other schemas	Dataclass, TypedDict, raw dict
02	Streaming	Stream deltas + complete event
03a	Tools (call)	Tool definitions with call()
03b	Tools (stream)	Tool call chunks in streaming
04a	Generate image	Image generation
04b	Describe image	Vision / image description
04c	Edit image	Image editing
05	Audio	Text-to-speech + speech-to-text
06	Video	Video generation
07	Embeddings	Vector embeddings + similarity
08	Function mgmt	List, get, revisions, delete
09	Observability	Tracing with decorator + context manager
09b	Manual tracing	Manual span creation
09c	Traces	List, get, and inspect traces
10	Models	List available models
12	Knowledge base	Semantic search with knowledge bases
13	Web tools	Web search and URL fetch (beta)

Run a single example:

export OPPER_API_KEY="your-key"
uv run python examples/getting-started/00_your_first_call.py

Run all examples:

uv run python examples/run_all.py

Configuration

Parameter	Default	Env Var
api_key	—	OPPER_API_KEY
base_url	https://api.opper.ai	OPPER_BASE_URL
headers	{}	—

Error Handling

from opperai import ApiError

try:
    opper.call("my-fn", input="hello")
except ApiError as e:
    print(e.status, e.body)

Async Support

All methods have _async variants:

result = await opper.call_async("summarize", input={"text": "..."})

async for chunk in opper.stream_async("summarize", input={"text": "..."}):
    print(chunk.delta, end="")

Requirements

• Python 3.10+
• Optional: pip install opperai[pydantic] for Pydantic schema support

License

MIT