upivia — Python SDK

Typed wrapper over the Upivia v1 HTTP API: governed service requests, budgets, approvals, agents, workflows, streaming chat, triggers, storage, and knowledge. Version 0.2.0 has full parity with @agentwallet/sdk (TypeScript) and ships both a sync UpiviaClient and an async AsyncUpiviaClient.

Full guides live on the platform at /docs/sdk.

Install

The SDK lives inside the Upivia monorepo at packages/sdk-py/.

Not yet published to PyPI — publishing is planned soon as pip install upivia. Until then, install from the monorepo:

cd packages/sdk-py
uv sync --extra dev      # dev env (pytest, respx)
uv run pytest            # run the test suite

Or from another project: pip install -e path/to/packages/sdk-py. Requires Python ≥ 3.10; the only runtime dependency is httpx.

Quick start (sync)

from upivia import UpiviaClient

client = UpiviaClient(
    api_key="agent_key_xxx",            # or env UPIVIA_API_KEY
    base_url="https://www.upivia.com",  # or env UPIVIA_BASE_URL
)

result = client.service_requests.create(
    service="email",
    operation="send",
    payload={"to": "client@example.com", "subject": "Follow-up", "body": "Hi."},
)

if result["status"] == "executed":
    # Money is integer cents on the wire; format as dollars for display.
    print(f"sent for ${result['cost_cents'] / 100:.2f}")

Methods return parsed JSON dicts (None for 204 responses). The client is a context manager (with UpiviaClient(...) as client:).

Quick start (async)

AsyncUpiviaClient has the identical resource surface over httpx.AsyncClient — all methods are coroutines, iterators are async generators, and the chat turn is an async generator:

import asyncio
from upivia import AsyncUpiviaClient

async def main() -> None:
    async with AsyncUpiviaClient(api_key="agent_key_xxx", base_url="...") as client:
        result = await client.service_requests.create(
            service="text_generation",
            operation="generate",
            payload={"prompt": "One-line haiku about budgets."},
        )
        print(result["status"])

        async for row in client.audit_logs.iter(service="email"):
            print(row["event_type"])

asyncio.run(main())

Configuration

Constructor kwargs with env-var fallbacks:

Kwarg	Env fallback	Notes
api_key	UPIVIA_API_KEY, AGENTWALLET_API_KEY	Agent key, Bearer auth on agent endpoints
pat	UPIVIA_PAT	Personal access token for workspace endpoints
base_url	UPIVIA_BASE_URL, AGENTWALLET_BASE_URL	Required (raises ValueError when unresolved)
max_retries	—	Default 2 (0 disables)
timeout	—	Default 60s; SSE read timeout is unbounded
transport	—	httpx transport injection (tests)
default_headers, generate_idempotency_key	—	See docstrings

With env vars set, UpiviaClient() needs no arguments.

Auth modes

Every method documents its auth mode in its docstring:

• Agent key (api_key) — agent endpoints: service requests, spawn, budget check, memory read/write, delegation.
• PAT (pat) — workspace endpoints: balance, usage, audit logs, health, chat sessions/turns, agent requests, teams, workspaces, devices.
• Cookie session — endpoints marked cookie-session-only (most dashboard mutations: agent CRUD, workflows, triggers, storage, knowledge, scheduled tasks) reject PATs server-side; call them from a transport that carries the session cookie.

Service request outcomes

Branch on result["status"]:

r = client.service_requests.create(service=..., operation=..., payload=...)
match r["status"]:
    case "executed":          ...  # r["result"], r["cost_cents"]
    case "blocked":           ...  # r["reason_code"], r["message"] — policy/budget stop
    case "approval_required": ...  # r["approval_id"], r["expires_at"] — human gate
    case "failed":            ...  # r["reason_code"], r["message"] — provider error

Async/reconcilable operations may return status: "running" (HTTP 202, DEC-050). Poll service_requests.get(request_id) yourself, or:

done = client.service_requests.create_and_wait(
    service="voice_call", operation="create", payload={...},
    poll_interval=2.0, timeout=300.0,  # defaults shown; backoff ×1.5, cap 10s
)
# resolves at executed | failed | blocked (non-running responses return as-is)

Idempotency

service_requests.create auto-generates an idempotency key when none is supplied; pass idempotency_key= to pin your own. Replays with the same key and identical payload return the cached response; same key + different payload raises UpiviaError(kind="idempotency_conflict"). The key is also what makes the POST safe for automatic retries.

Streaming chat

chat.turn() yields ChatStreamEvent dataclasses (.event, .data) parsed from the SSE stream (PAT or cookie session):

session = client.chat.sessions.create(agent_id="agt_...")
for ev in client.chat.turn(session_id=session["id"], message="Summarize yesterday's spend."):
    if ev.event == "message_delta":
        print(ev.data.get("delta", ""), end="", flush=True)
    elif ev.event == "tool_pending":
        print("needs approval:", ev.data.get("approval_id"))
    elif ev.event == "done":
        break

Async: async for ev in client.chat.turn(...). Inline approvals: chat.resolve_approval(id, "approve"), then resume the turn with continue_from={"request_id": ..., "action": ...}.

Pagination

Cursor endpoints expose an auto-paginating iter() alongside list():

for row in client.audit_logs.iter(service="email"):          # sync
    print(row["event_type"], row["created_at"])

async for row in client.audit_logs.iter(service="email"):    # async client
    ...

Available on audit_logs.iter, agents.activity_iter, storage.objects.iter, knowledge.collections.iter, and triggers.iter.

Retries, timeouts, errors

Automatic retries (default 2) apply to network errors and 429/502/503/504 — GETs always, mutations only when idempotent (e.g. service-request creates carrying an Idempotency-Key). Retry-After is honored on 429; otherwise exponential backoff with full jitter (0.5s · 2ⁿ, cap 8s).

All failures raise UpiviaError. Switch on err.kind:

kind	Meaning
"network"	request never reached the server
"unauthorized"	401/403
"rate_limited"	429 (err.retry_after = parsed Retry-After seconds)
"idempotency_conflict"	409 with code idempotency_key_payload_mismatch
"http"	any other non-2xx
"invalid_response"	body wasn't JSON

err.status, err.code, err.server_message, err.body, err.retry_after, and err.request_id (x-request-id) are populated when available.

Resource surface

Same layout on both clients (async methods are coroutines):

Resource	Methods
service_requests	create · get · create_and_wait
balance	get(team_id=)
usage	list(agent_id=, from_=, to=, limit=, include_chart=)
approvals	list · approve · reject
audit_logs	list · iter
agents	list · create · get · update · delete · reset_key · clone · transfer · set_budget · enable_service · disable_service · health · fleet_health · activity · activity_iter · skills · remove_skill
spawn	create · estimate
delegate	create · list · get_task · update_task · reattach
memory	list · search · create · update · delete · graph
workflows	list · create · get · update · delete · create_version · publish_version (governed) · unpublish_version · share · export · run_and_wait
workflows.runs	list · create · get · cancel · rerun_failed · retry_step
workflows.agents	list · grant · revoke
workflows.from_template	list · create
scheduled_tasks	list · create · get · update · delete · runs
agent_requests	list · create · resolve
budget_check	check(agent_id) (free platform.check_budget meta-op)
chat	turn (SSE generator) · resolve_approval
chat.sessions	list · create · get · delete · delete_all
triggers	list · iter · create · get · update · delete · fire (HMAC)
storage.objects	list · iter · get · delete · restore · upload · presign · confirm · download
knowledge.collections	list · iter · create · get · delete
knowledge.documents	create · get · delete
services	list (public catalog; prices in integer cents)
teams	list · switch · budget · allocate_budget · members.list · members.update
workspaces	list · switch
budget_requests	create · list · resolve
devices	heartbeat · sessions.{list,update,command,delete,cwd} · commands.update
platform	health · readiness · agent_docs

Note: workflow publishing is governed — publish_version creates a PublishRequest an admin must approve; the version is not live immediately.

Firing a webhook trigger (HMAC)

triggers.create returns the signing secret exactly once (never re-readable). fire() serializes the payload (compact JSON), computes HMAC-SHA256 of that exact raw string with the secret, and sends X-AgentWallet-Signature: sha256=<hex> — no Bearer header:

created = client.triggers.create({
    "agent_id": "agt_...",
    "kind": "webhook",
    "operation": "email.send",
    "payload_template": {"to": "ops@example.com", "subject": "Alert", "body": "{{message}}"},
})
# Store created["secret"] now — it is shown only once.

fired = client.triggers.fire(
    created["trigger"]["id"],
    {"message": "disk usage at 91%"},
    secret=created["secret"],
)
# 202 even when downstream dispatch was rejected — check fired["dispatch_status"].

Storage: upload and presign

# Direct multipart upload (≤100 MB), cookie session:
obj = client.storage.objects.upload(
    filename="hello.txt", content=b"hello", content_type="text/plain"
)

# Larger files: presign → PUT → confirm:
pre = client.storage.objects.presign(
    filename="big.bin", content_type="application/octet-stream", size_bytes=500_000_000
)
httpx.put(pre["upload_url"], content=big_bytes)
client.storage.objects.confirm(pre["object"]["id"], sha256=digest)

# Resolve a time-limited signed download URL (redirect is not followed):
info = client.storage.objects.download(obj["object"]["id"])
print(info["url"])

LangGraph integration

Route LangGraph agents' tool calls through Upivia's policy/budget/approval/audit pipeline instead of calling providers directly:

pip install 'upivia[langgraph]'   # adds langchain-core

from upivia import UpiviaClient
from upivia.integrations.langgraph import UpiviaToolkit

client = UpiviaClient(api_key="agent_key_xxx", base_url="...")

toolkit = UpiviaToolkit(client)          # catalog-driven: GET /api/v1/services →
tools = toolkit.as_langchain_tools()     # one StructuredTool per service.operation

# Then hand `tools` to your LangGraph agent/graph as usual.

Tool generation is catalog-driven by default (one tool per operation, with the operation's input_schema attached as args_schema when supported). When the catalog is unreachable — or with UpiviaToolkit(client, offline=True) — it falls back to 11 hardcoded default operations. Pass operations=[(service, operation, description), ...] to pin an explicit set.

Every tool result starts with a discriminated STATUS: <status> line (executed | approval_required | blocked | failed | running | error) so agents react to governance outcomes (budget blocks, approval pauses) without crashing the loop. toolkit.as_callables() returns raw callables with no langchain dependency.

Custom transport (tests)

Inject an httpx transport for offline tests:

import httpx
from upivia import UpiviaClient

def handler(req: httpx.Request) -> httpx.Response:
    return httpx.Response(200, json={"amount_cents": 5000})

client = UpiviaClient(
    base_url="http://test.local",
    transport=httpx.MockTransport(handler),
)

See tests/test_client.py and tests/test_async_client.py for the full pattern.

More

• CHANGELOG.md — full 0.2.0 method list and back-compat notes.
• examples/python-quickstart.py and examples/python-async-quickstart.py — runnable demos.
• Platform docs: /docs/sdk and the machine-readable spec at GET /api/v1/agent-docs.