Python SDK Guide

unitysvc-sellers ships a typed Python SDK that the usvc_seller CLI is itself built on. Anything the CLI can do — upload a catalog, list services, mutate promotions, run connectivity tests — you can do programmatically via unitysvc_sellers.Client (sync) or unitysvc_sellers.AsyncClient (async).

The SDK is generated from the seller OpenAPI spec, so request and response models are fully typed: your editor will autocomplete fields and your type checker will catch mistakes before they hit the wire.

Installation

pip install unitysvc-sellers

Authentication

Both Client and AsyncClient read credentials from the environment by default:

Variable	Purpose
UNITYSVC_SELLER_API_KEY	Seller API key (svcpass_…). Get one from the UnitySVC dashboard.
UNITYSVC_SELLER_API_URL	Base URL for the seller API. Defaults to https://seller.unitysvc.com/v1.

from unitysvc_sellers import Client

client = Client()              # reads both env vars, raises if key missing
client = Client(api_key="...") # explicit key, env URL (or default)
client = Client(api_key="...", base_url="https://seller.unitysvc.com/v1")

The seller context is encoded entirely in the API key — there is no seller_id argument to pass. Both clients are safe to use as context managers and will close their underlying httpx session on exit:

from unitysvc_sellers import Client

with Client() as client:
    for svc in client.services.list():
        print(svc.name)

The list response is iterable directly; you can still inspect .next_cursor / .has_more for manual pagination, or call client.services.iter_all() to walk every page.

Sync vs async: when to use which

You are writing…	Use
A one-off script or a Jupyter notebook	Client
A Celery task or other already-sync worker	Client
An async def function, FastAPI handler, or asyncio app	AsyncClient
A CLI tool where you want to overlap multiple requests	AsyncClient

The two are a one-to-one mirror: every sync method has an async counterpart with the same name and signature, just awaitable.

Sync example

from unitysvc_sellers import Client

with Client() as client:
    services = client.services.list(limit=50)
    for svc in services:
        print(svc.id, svc.name, svc.status)

Async example

import asyncio
from unitysvc_sellers import AsyncClient

async def main():
    async with AsyncClient() as client:
        services = await client.services.list(limit=50)
        for svc in services:
            print(svc.id, svc.name, svc.status)

asyncio.run(main())

Resource namespaces

Both clients expose the same eight resource namespaces as lazy properties:

Namespace	Underlying endpoints	Purpose
client.services	/seller/services/*	List, get, upload, mutate status / routing / pricing, delete services
client.templates	/seller/templates/*	Browse the platform service-template catalog (read-only: list / get)
client.instances	/seller/instances/*	Create a service from a template, and list / get / delete your instances
client.promotions	/seller/promotions/*	CRUD on seller-funded promotion codes
client.groups	/seller/service-groups/*	CRUD on service groups
client.documents	/seller/documents/*	Fetch document file content, execute (gateway dispatch), update test
client.secrets	/seller/secrets/*	Create, rotate, list, and delete encrypted seller secrets
client.tasks	/seller/tasks/*	Poll Celery task state for async operations (notably service upload)

The async client exposes the same namespaces with Async prefixes on the resource classes (AsyncServices, AsyncGroups, AsyncPromotions, …), but the attribute names on the client are identical — so you can rename Client to AsyncClient in a code snippet and the rest just needs awaits.

Active-record style

The services, groups, and promotions namespaces all return active-record wrappers rather than raw generated dataclasses:

• client.services.get(id) returns a Service object; client.services.list(...) returns a ServiceList whose items are Services.
• client.groups.get(id) / .upsert(...) / .update(...) return Group objects; .list(...) returns a GroupList.
• client.promotions.get(id) / .upsert(...) / .update(...) return Promotion objects; .list(...) returns a PromotionList.

Each wrapper:

1. Forwards field access to the underlying generated record via __getattr__, so svc.id, grp.display_name, promo.code all work transparently.
2. Carries methods bound to its id, so you can write svc.update({"status": "pending"}) instead of client.services.update(svc.id, {"status": "pending"}).
3. Is iterable (the list wrappers), so for svc in client.services.list(): ... walks the current page of Service objects directly. .next_cursor / .has_more / .next_page() remain available for manual pagination, and client.services.iter_all(...) walks every page automatically.

Manager-style calls (client.services.update(service_id, body)) are still supported for the case where you only have an id from a webhook or external system — the wrappers are sugar, not a replacement.

The documents, secrets, and tasks namespaces stay as plain function-style managers — they have a single short verb each (or no mutating verbs at all) and don't benefit from the active-record ceremony.

client.services

The services resource covers the full seller-catalog lifecycle.

Manager methods on client.services:

Method	Description
list(cursor=, limit=, …)	Cursor-paged list returning a :class:ServiceList (iterable, has next_page).
iter_all(...)	Generator that walks every page automatically.
get(service_id)	Full service detail wrapped as a :class:Service.
run_tests(service_id, *, document_id=None, force=False, ...)	Queue a server-side gateway-then-upstream diagnostic and block until complete. Returns a typed :class:RunTestsResult with per-(doc × iface) rows and pass/fail counts.
upload(body)	POST provider + offering + listing bundle. Backend responds with a task id.
update(service_id, body)	Patch fields — status, visibility, routing_vars, list_price.
delete(service_id, dryrun=)	Remove a service. (Most flows should use update({"status": "deprecated"}).)

Service wrappers expose the same operations pre-bound to the service id: svc.refresh(), svc.update(body), svc.delete(), svc.run_tests(), plus the svc.submit() shortcut for transitioning status to "pending".

Listing services

client.services.list(...) returns a ServiceList — iterable directly over the current page of Service wrappers, with next_cursor / has_more available for manual pagination:

with Client() as client:
    # One page at a time:
    services = client.services.list(limit=100, status="active")
    for svc in services:
        print(svc.id, svc.name, svc.service_type)
    # Then services.next_page() for the next page, or:
    while services.has_more:
        services = services.next_page()
        for svc in services:
            ...

    # Or have the SDK walk every page for you:
    for svc in client.services.iter_all(status="active"):
        print(svc.id, svc.name)

Fetching one service

client.services.get(...) returns a Service active-record wrapper. Field access is forwarded to the underlying record, and the wrapper carries methods bound to that service id:

svc = client.services.get("be098e7d-59e1-498a-bc4f-e389eb61c70b")
print(svc.name, svc.status)
for iface in svc.interfaces:
    print(" iface:", iface.name, iface.base_url)
for doc in svc.documents or []:
    print(" doc:", doc.title, doc.category, doc.test_status)

Mutating status

Service exposes the same mutations as the manager — pre-bound to its service id:

svc = client.services.get("be098e7d-59e1-498a-bc4f-e389eb61c70b")

svc.submit()                                 # shortcut: status -> "pending"
svc.update({"visibility": "public"})         # arbitrary patch
svc.update({"list_price": {"type": "constant", "price": "1.00"}})

svc = svc.refresh()                          # re-fetch to observe transitions
print(svc.status)

svc.delete()                                  # remove (most flows prefer status="deprecated")

If you only have a service id from a webhook, the manager-style calls still work: client.services.update(service_id, {"status": "pending"}), client.services.delete(service_id).

client.templates

The platform service-template catalog (read-only) — discover what you can instantiate. list the active templates and get one's parameter schema. Creating a service from a template lives on client.instances (below).

Manager methods on client.templates:

Method	Description
list(skip=, limit=, service_type=)	Active templates you can instantiate.
get(template_id)	One template's metadata + parameter schema.

client.instances

Template instances — the SDK counterpart of the dashboard's Create from template flow. create renders a template into a draft service (and, with auto_submit=True, also submits it for review); list / get / delete manage your instances.

Manager methods on client.instances:

Method	Description
create(template_id, parameters=, name=, auto_submit=False)	Render a template into a draft service; auto_submit=True also submits it for review. Returns instance_id + ingest task_id.
list(skip=, limit=)	Your instances, with derived service status.
get(instance_id)	One instance: parameters, template metadata, linked service.
delete(instance_id)	Delete the instance record (the linked service is not unpublished).

from unitysvc_sellers import Client

with Client() as client:
    for tpl in client.templates.list():          # discover (catalog)
        print(tpl.name, tpl.version)

    result = client.instances.create(            # create a draft
        "openai-compatible-llm",
        parameters={
            "api_base_url": "https://api.example.com/v1",
            "api_key_secret_name": "UPSTREAM_API_KEY",  # the secret NAME, never the value
            "input_price": 1.00,
        },
        name="my-llm",
        # Draft by default; pass auto_submit=True to also submit for review now.
    )
    # Poll the ingest task to a verdict with client.tasks if you need to block.
    print(result["instance_id"], result["task_id"])

Secret-typed parameters take the name of a secret you created with client.secrets, never the key value. Capability pools opt in the same way — create from a pool-named template and the resulting service joins /p/<pool> at the pool's uniform terms.

client.promotions

Seller-funded promotion codes — discounts the seller pays for, to bootstrap demand or reward specific customers.

Manager methods on client.promotions:

Method	Description
list(cursor=, limit=, status=)	Cursor-paged list returning a PromotionList (iterable, has next_page).
iter_all(...)	Generator that walks every page automatically.
get(promotion_id)	Fetch one by id, returned as a Promotion.
upsert(body)	Create-or-update by name (idempotent), returned as a Promotion.
update(promotion_id, body)	Partial patch by id, returned as a Promotion.
delete(promotion_id)	Permanent delete. Returns None.

Promotion wrappers expose the same operations pre-bound to the promotion id: promo.refresh(), promo.update(body), promo.delete().

Creating, listing, and mutating promotions

from unitysvc_sellers import Client

with Client() as client:
    # Create or update by name. Returns a Promotion bound to its id.
    promo = client.promotions.upsert({
        "name": "summer2026",
        "code": "SUMMER2026",
        "pricing": {"type": "percentage_off", "percentage": "25.00"},
        "scope": {"service_type": "llm"},
        "apply_at": "request",
        "priority": 10,
        "status": "active",
    })
    print(promo.id, promo.code, promo.status)

    # Iterate active promotions on the current page.
    for p in client.promotions.list(limit=100, status="active"):
        print(p.code, p.status)

    # Or have the SDK walk every page for you.
    for p in client.promotions.iter_all(status="active"):
        print(p.code, p.status)

    # Pause via the active-record handle (no need to thread the id through).
    promo.update({"status": "paused"})

    # Re-fetch after the mutation to observe the new state.
    promo = promo.refresh()
    assert promo.status == "paused"

    # Delete (permanent).
    promo.delete()

If you only have a promotion id (e.g. from a webhook), the manager-style calls still work: client.promotions.update(promotion_id, {"status": "paused"}), client.promotions.delete(promotion_id).

client.groups

Service groups bundle related services together for routing, discovery, or group-wide pricing.

Manager methods on client.groups:

Method	Description
list(cursor=, limit=, status=)	Cursor-paged list returning a GroupList (iterable, has next_page).
iter_all(...)	Generator that walks every page automatically.
get(group_id)	Fetch one by id, returned as a Group.
upsert(body)	Create-or-update by name (idempotent), returned as a Group.
update(group_id, body)	Partial patch by id, returned as a Group.
delete(group_id)	Permanent delete. Returns None.

Group wrappers expose the same operations pre-bound to the group id: grp.refresh(), grp.update(body), grp.delete().

Creating, listing, and mutating groups

with Client() as client:
    # Upsert returns a Group with .id already set.
    grp = client.groups.upsert({
        "name": "premium-llms",
        "display_name": "Premium LLMs",
        "owner_type": "seller",
        "membership_rules": {
            "include_tags": ["premium", "llm"],
        },
    })
    print(grp.id, grp.name, grp.display_name)

    # Iterate one page of the catalog.
    for g in client.groups.list(limit=50):
        print(g.name, g.display_name, g.status)

    # Or walk every page in one loop.
    for g in client.groups.iter_all():
        print(g.name)

    # Mutate via the active-record handle.
    grp.update({"display_name": "Premium LLMs (curated)"})

    # Re-fetch after the mutation.
    grp = grp.refresh()
    assert grp.display_name == "Premium LLMs (curated)"

    # Delete the group entirely.
    grp.delete()

Manager-style calls (client.groups.update(group_id, body), client.groups.delete(group_id)) remain available for the case where you only have an id.

client.documents

Operate on documents attached to a service (code examples, connectivity tests, terms, logos).

with Client() as client:
    doc = client.documents.get("5c937b4a-980a-4a69-bb90-a1f93816de1d")
    print(doc.title, doc.mime_type)
    if doc.file_content:
        print(doc.file_content[:200])

    # Record an externally-run test result. Useful when the CLI or
    # an in-house CI ran the script locally and wants to POST the
    # per-interface outcome back.
    client.documents.update_test(
        doc.id,
        {
            "status": "success",
            "tests": {
                "default": {"status": "success", "exit_code": 0},
            },
        },
    )

client.documents.execute(doc_id, force=True) dispatches a single document to the backend Celery worker for execution against the gateway. To run every executable document on a service across every active interface — with gateway-then-upstream attribution on failures — use client.services.run_tests(...) (or the equivalent svc.run_tests() instance method) instead. The CLI usvc_seller services run-tests <service_id> is a thin wrapper over the same SDK call.

client.secrets

Manage encrypted seller secrets (API keys, tokens, credentials). Values are write-only — only metadata is ever returned by the API.

The shape mirrors GitHub's secrets API: there is no separate create or rotate — set(name, value) does both in one idempotent call.

Method	Parameters	Returns	Description
list(skip=0, limit=100)	skip: int, limit: int	SecretsPublic	List secrets (metadata only)
get(name)	name: str	SecretPublic	Get one secret's metadata by name
set(name, value)	name: str, value: str	SecretPublic	Idempotent create-or-replace
delete(name)	name: str	None	Permanently delete a secret

Secret names must be uppercase with underscores (e.g. OPENAI_API_KEY, STRIPE_SECRET). Names starting with __ are reserved for platform use.

from unitysvc_sellers import Client

with Client() as client:
    # Create or rotate — `set` is idempotent.
    client.secrets.set("OPENAI_API_KEY", "sk-proj-abc123...")

    # List all secrets (metadata only).
    for s in client.secrets.list().data:
        print(s.name, s.created_at, s.last_used_at)

    # Get one secret's metadata.
    meta = client.secrets.get("OPENAI_API_KEY")
    print(meta.name, meta.updated_at)

    # Rotate by calling set again with a new value.
    client.secrets.set("OPENAI_API_KEY", "sk-proj-new456...")

    # Delete (immediate effect — services referencing it will break).
    client.secrets.delete("OPENAI_API_KEY")

Write-only values

The API never returns secret values. If you lose the value, rotate it with a new one. There is no "get value" endpoint.

client.tasks

Several seller endpoints are fire-and-forget: they accept your request, queue a Celery task, and return a task_id. The tasks resource lets you poll for the real outcome. Both methods are variadic — pass one or many ids in a single call.

Method	Description
get(*task_ids)	One request, returns {task_id → status_dict} for each id (up to 100/call).
wait(*task_ids, timeout=, poll_interval=, on_update=)	Poll until every id reaches a terminal state, or time out.

Terminal Celery states the SDK recognises are completed and failed. Both methods return a dict keyed by task_id; each status entry typically contains task_id, status, result, and (for failures) error / message fields.

with Client() as client:
    result = client.services.upload({"services": [...]})
    task_id = result.task_id

    final = client.tasks.wait(task_id, timeout=300, poll_interval=2)
    status = final[task_id]
    if status["status"] == "completed":
        print("service_id =", status["result"]["service_id"])
    else:
        print("failed:", status.get("error") or status.get("message"))

upload_directory — full catalog upload helper

For the common case of "upload a whole seller catalog directory, wait for every task, write each folder's service.json so the next run knows what already exists", use the upload_directory helper from unitysvc_sellers.upload. This is the same code path the usvc_seller specs upload CLI command uses.

from pathlib import Path
from unitysvc_sellers import Client
from unitysvc_sellers.upload import upload_directory

with Client() as client:
    result = upload_directory(
        client,
        data_dir=Path("./specs"),
        dryrun=False,
        task_wait_timeout=600.0,
        task_poll_interval=2.0,
    )

print(f"services: {result.services.success}/{result.services.total}")
print(f"promotions: {result.promotions.success}/{result.promotions.total}")
print(f"groups: {result.groups.success}/{result.groups.total}")

for err in result.services.errors:
    print(" FAILED", err["file"], "→", err["error"])

if result.total_failed:
    raise SystemExit(1)

Useful keyword arguments:

• dryrun=True — call every endpoint with the dryrun flag where supported; skips task polling entirely (dryrun runs synchronously on the backend).
• upload_services=False / upload_promotions=False / upload_groups=False — upload only a subset of the catalog.
• on_progress=callable — a hook that fires on every state transition. Signature is on_progress(kind, status, name, detail), where kind is "service", "promotion", or "group" and status transitions through "queued" → ("ok" or "error" or "dryrun"). Useful when embedding the upload into a larger progress bar or web UI.
• task_wait_timeout / task_poll_interval — control how long the helper is willing to wait for the async ingest tasks to finish.

Returns an UploadResult dataclass with per-resource services / promotions / groups counts (total / success / failed / errors).

Running connectivity tests

The usvc_seller services run-tests CLI queues a server-side diagnostic that runs each executable document (connectivity tests and code examples) across every active access interface from inside the cluster — the same network path customers hit. On any iface-level gateway failure it falls back to an upstream-mode probe so the result attributes the fault as platform_fault (gateway broken, upstream fine) or upstream_fault (both fail).

The SDK exposes the same call directly:

from unitysvc_sellers import AsyncClient


async def run_diagnostic(service_id: str) -> int:
    async with AsyncClient() as client:
        result = await client.services.run_tests(service_id)
        # result is a RunTestsResult dataclass — see services.py
        for row in result.results:
            mark = "✓" if row.status == "success" else "✗"
            attribution = f" → {row.outcome}" if row.outcome else ""
            print(f"  {mark} {row.document_title} [{row.interface_name}]: "
                  f"{row.status}{attribution}")
        return result.fail_count

Sync equivalent:

with Client() as client:
    result = client.services.run_tests(
        service_id,
        document_id=None,    # optional: restrict to one document
        force=False,         # re-execute already-passing rows
        poll_interval=2.0,   # seconds between task-status polls
        timeout=600.0,       # hard cap on total wait
    )
    print(f"passed: {result.passed}  fail_count: {result.fail_count}")

The active-record shortcut svc.run_tests() pre-binds the service id; everything else is the same.

What the backend does under the hood:

1. Snapshots service.status and temporarily elevates draft / rejected to pending so the gateway routing layer will resolve. Restored in a finally block — a worker crash mid-run can't leave the service permanently elevated.
2. Iterates every executable document × every active access interface; renders gateway-mode and runs the script in a worker subprocess.
3. On any iface-level gateway failure, renders+executes the same document in upstream mode (cached per-doc — a single upstream config drives all ifaces) for fault attribution.
4. Persists per-(doc × iface) results to Document.meta.test.tests[<iface_id>] (readable later via usvc_seller services show-test --doc-id <id> for full stdout/stderr).
5. Restores the snapshotted status and returns the summary.

The SDK call blocks until the task completes; under the hood it polls GET /v1/seller/tasks/{task_id} every poll_interval seconds until the task reaches a terminal state.

Exceptions

All SDK errors derive from SellerSDKError. The subclasses map directly to HTTP status classes so you can catch them narrowly:

Exception	Raised when…
AuthenticationError	401 — key missing, invalid, or expired
PermissionError	403 — key is valid but not authorised for this action
NotFoundError	404 — resource does not exist
ValidationError	400 / 422 — request body failed server validation
ConflictError	409 — e.g., service name already taken
RateLimitError	429 — retry with the Retry-After hint
ServerError	5xx — backend trouble
APIError	Catch-all parent of the above
SellerSDKError	Root. Also covers non-HTTP failures (config, auth).

from unitysvc_sellers import (
    Client,
    NotFoundError,
    ValidationError,
    APIError,
)

try:
    svc = client.services.get("00000000-0000-0000-0000-000000000000")
except NotFoundError:
    print("service does not exist")
except ValidationError as exc:
    print("bad request:", exc.detail)
except APIError as exc:
    print(f"HTTP {exc.status_code}:", exc)

Every APIError carries:

• status_code (int)
• detail (the parsed ErrorResponse from the backend, usually a dict, or the raw text for non-JSON responses)
• response_body (the raw bytes, for debugging)

End-to-end example: mirror a catalog into CI

Combine the pieces into a CI-friendly script that uploads a catalog, waits for all ingest tasks, and fails the build on any per-service errors:

#!/usr/bin/env python3
"""CI entrypoint: upload the catalog under ./data and report results."""

import sys
from pathlib import Path

from unitysvc_sellers import Client
from unitysvc_sellers.upload import upload_directory


def main() -> int:
    data_dir = Path("./specs").resolve()

    def progress(kind: str, status: str, name: str, detail: str = "") -> None:
        icon = {"queued": "·", "ok": "✓", "error": "✗", "dryrun": "→"}.get(status, "?")
        print(f"  {icon} {kind}: {name} [{status}] {detail}")

    with Client() as client:
        result = upload_directory(
            client,
            data_dir=data_dir,
            on_progress=progress,
            task_wait_timeout=900.0,
            task_poll_interval=3.0,
        )

    print()
    print(f"services:   {result.services.success}/{result.services.total} ok")
    print(f"promotions: {result.promotions.success}/{result.promotions.total} ok")
    print(f"groups:     {result.groups.success}/{result.groups.total} ok")

    for err in (*result.services.errors, *result.promotions.errors, *result.groups.errors):
        print(f"  FAILED {err.get('file', '?')} → {err.get('error')}")

    return 1 if result.total_failed else 0


if __name__ == "__main__":
    sys.exit(main())

Drop that in scripts/upload_catalog.py, wire UNITYSVC_SELLER_API_KEY and UNITYSVC_SELLER_API_URL into your CI secrets, and call it from a GitHub Actions workflow:

- name: Upload catalog
  env:
    UNITYSVC_SELLER_API_KEY: ${{ secrets.UNITYSVC_SELLER_API_KEY }}
    UNITYSVC_SELLER_API_URL: ${{ secrets.UNITYSVC_SELLER_API_URL }}
  run: python scripts/upload_catalog.py

See also

• SDK Reference — auto-generated class and method documentation from source docstrings.
• Getting Started — install, configure, and upload your first catalog.
• CLI Reference — the complete command and option listing.
• Operate Live Services and Generate a Catalog — higher-level catalog patterns.
• File Schemas — the provider / offering / listing payload shapes the SDK accepts.