The soundverse-py SDK

DeployedNot a service

soundverse-py is the shared Python library every backend worker imports — the agent, Sansaarm, stitch, media, gpu, and the template-core-tool scaffold all build on it. It is not a deployable service; it has no main entrypoint of its own. The distributed package is named soundverse (you write from soundverse.tools import WorkerFleet), and it wraps the generated soundverse_proto stubs so a tool author never touches raw JSON, storage RPCs, or the task queue.

Two halves matter to almost everyone:

• WorkerFleet — the runtime that runs your tool. It registers tools, claims tasks, keeps leases alive, uploads results, reports usage, and drains on shutdown.
• TaskContext (ctx) — the API surface your process() method talks to. Everything that reaches the platform — progress events, uploads, credentials, usage, chat, metadata — goes through ctx.

If you are writing a tool, start at Add a tool worker; this page explains the machinery underneath it.

Where it sits

• Directorysoundverse-py/

  • Directorysrc/soundverse/

    • Directorytools/

      • fleet.py WorkerFleet — the poll/claim/run/drain runtime
      • context.py TaskContext + LiveAudioIngest
      • base.py BaseTool / ToolOutput / the @tool decorator
      • config.py LicensePricing / RateLimit / ToolCredentials
      • files.py OutputFile / AudioFile / ImageFile / VideoFile
      • inputs.py FileInput / AudioInput / ImageInput / VideoInput
      • convert.py conversion target selection
    • config.py SoundverseEnv / ServiceRegistryEnv (env → settings)
    • Directoryclients/ SoundverseClient — the internal gRPC/Connect client

      • …
    • keys.py the {env}:{namespace}: Redis key contract
    • telemetry.py OTLP setup + span/tracer helpers

Pulled at build time, pinned to main

Workers depend on soundverse-py pinned @main (the submodule tracks the main branch), so Docker resolves it at image-build time. An SDK change only reaches a worker after it is merged to soundverse-py main and the consuming worker is rebuilt — rebuilding a worker before the merge lands just pulls the old @main. To test an unmerged SDK locally, use the editable overlay described in Test unmerged soundverse-py changes.

WorkerFleet: the worker runtime

A tool worker is a long-running poller with no ingress — no HTTP, no FastAPI. app/main.py builds a SoundverseClient, hands the fleet its list of BaseTool instances (ALL_TOOLS), and calls fleet.start(), which blocks in the poll loop until SIGTERM. One process can host several tools.

app/main.py (the whole bootstrap)

from soundverse.clients.rpc import SoundverseClient
from soundverse.config import config
from soundverse.tools import WorkerFleet
from app.tools import ALL_TOOLS

client = SoundverseClient(config)
fleet = WorkerFleet(
    config, client, ALL_TOOLS,
    environment=worker_settings.environment,   # must match the core-mcp it feeds
    max_concurrency=worker_settings.max_concurrency,
    poll_interval=worker_settings.poll_interval,
)
fleet.start()   # blocks; drains in-flight tasks + deregisters on SIGTERM

flowchart TD
  subgraph Fleet["WorkerFleet — one process, N tools"]
    reg["register + upsert config"]
    poll["poll loop — claim per tool"]
    lanes["leaf lane / orchestrator lane"]
    run["run process() in thread pool"]
    hb["per-task heartbeat lease"]
    asm["assemble GenerationOutput"]
  end
  db[("core-database\ntask queue + tool registry")]
  redis[("Redis\ntask:wake + event stream")]
  storage["core-storage\nblob upload + transcode"]

  reg -->|"RegisterTool / UpsertTool*"| db
  redis -.->|"task:wake"| poll
  poll -->|"ClaimNextTask (SKIP LOCKED)"| db
  poll --> lanes --> run
  run -->|"emit_progress / emit_partial"| redis
  run -->|"upload_file"| storage
  hb -->|"HeartbeatTask"| db
  run --> asm -->|"CompleteTask / FailTask"| db

  classDef data     fill:#8b5cf6,color:#fff,stroke:#6d28d9
  classDef worker   fill:#10b981,color:#fff,stroke:#047857
  classDef external fill:#f59e0b,color:#111,stroke:#b45309
  class reg,poll,lanes,run,hb,asm worker
  class db,storage data
  class redis external

1. Registration + idempotent config

On boot the fleet stamps every tool with the resolved environment and calls RegisterTool (sending the input/output JSON schemas derived from your Pydantic models). Immediately after, it upserts the tool’s declared platform config — UpsertToolConfig, UpsertToolLicensePricing, UpsertToolRateLimits, UpsertToolCredentials. Those RPCs are idempotent upserts, so your code is the source of truth: edit a pricing value, restart the worker, and the row re-applies. See Tool pricing model and Add tool pricing.

Registration retries five times, then falls back to discovering an already-registered row by (model, operation) via GetActiveTools so a transient DB blip doesn’t strand the worker. If a tool still has no id afterward, start() raises — a worker that can’t claim its tool’s tasks fails fast rather than heartbeating as “healthy” while doing nothing.

ENVIRONMENT must match the core-mcp it feeds

RegisterTool writes the tool row under the fleet’s environment (from the ENVIRONMENT env var). core-mcp only advertises tools whose row matches its own ENVIRONMENT (GetActiveTools). A worker registered under a different environment than the MCP it feeds registers a row nobody exposes — the tool silently never appears. See core-mcp and the empty-panel entry in Troubleshooting.

2. Claiming — Postgres queue + a wake signal

The poll loop calls ClaimNextTask per tool. That RPC is an atomic UPDATE … RETURNING over generation.tasks (SKIP LOCKED), so multiple replicas never grab the same row — see Task queue on Postgres. An empty queue is the steady state, not an error: the DB codegen maps a no-row claim to NotFound, which the fleet treats as “nothing to claim” and moves on.

Polling is the floor, but pickup is near-instant thanks to a best-effort pub/sub nudge. The gateway (and core-mcp) publish a task id on the env-namespaced channel {env}:common:task:wake right after QueueTask commits; a subscriber thread flips a wake event so the poll loop re-sweeps immediately instead of waiting for the next tick. A dropped wake only delays pickup to the jittered poll fallback — the jitter keeps replicas’ fallback polls desynchronised.

Redis DB index must match the gateway's

The wake channel and the live event stream are scoped to a Redis logical DB index, carried in the URL path of REDIS_ADDR (…/0). If a worker and the gateway resolve different DBs, every live progress event is silently dropped (the gateway’s XREAD reads a different DB than the worker’s XADD writes). The fleet logs its resolved host:port/db on boot and stamps a worker fingerprint key so a drift is one greppable line. There is intentionally no separate DB knob — put the index in REDIS_ADDR.

3. Two concurrency lanes (the orchestrator deadlock)

The fleet runs tasks in a ThreadPoolExecutor bounded by two semaphores, not one. Most tools are leaf tools and draw from the leaf lane (MAX_CONCURRENCY). But an orchestrator tool — one that spends its run blocking on other fleet tasks — gets its own lane.

The agent is the motivating case: core-tool-agent hosts both the agent tiers and their leaf sub-tools (lyric_writer, song_namer, album_art) in one fleet. An agent task holds a slot for its whole run while awaiting those leaves. With a single shared pool, enough concurrent agents would hold every slot while the leaves they wait on sit queued and unclaimable — a starvation deadlock. Separate lanes make the waits-for graph a DAG with no resource cycle, so leaf tools always have slots and make progress regardless of orchestrator load.

You opt in with one class attribute:

core-tool-agent/app/tools/agent_tool.py

class AgentTool(BaseTool[AgentInput, AgentOutput]):
    is_orchestrator = True   # scheduled on the orchestrator lane, never the leaf lane

The orchestrator lane only exists when the fleet actually hosts an orchestrator, so a leaf-only fleet (Sansaarm) spends no extra threads. Sizes are per-service: core-tool-agent reads MAX_CONCURRENCY (leaf, default 10) and ORCHESTRATOR_CONCURRENCY (default 16 — idle-waiting agents are cheap, so size it generously above the leaf cap). Crucially, when the orchestrator lane is saturated the poll loop continues the sweep rather than breaking — a full agent lane must never stop leaf claiming.

4. Heartbeats & leases

Two heartbeats run in parallel. A per-task thread calls HeartbeatTask on a fixed interval to renew the task’s time-based lease so the queue’s zombie-reaper doesn’t reclaim an in-flight task (see Task queue). A separate worker heartbeat writes a short-TTL {env}:worker:active:{worker_id} key so operators can see live workers.

5. Run → assemble → report

For each claimed task the fleet builds a TaskContext, continues the trace carried on the task row’s W3C traceparent (the queue hop is invisible to auto-instrumentation, so the parent is extracted explicitly — see Observability), and runs on_start → parse_input → bind_file_inputs → process → assemble_output. Then it calls CompleteTask with the serialized output and the raw reported usage, or FailTask on an exception.

Output assembly is by convention (defined on ToolOutput): a field named text becomes the primary text; OutputFile fields — including AudioFile/ImageFile/VideoFile and lists — are uploaded to core-storage and become deduped assets; everything else folds into metadata_json.

Output JSON is snake_case, on purpose

CompleteTask serializes the GenerationOutput with MessageToJson(output, preserving_proto_field_name=True). The protobuf default is camelCase, which silently dropped every multi-word asset field (blob_hash, file_id, mime_type) — only url/role survived — so sub-tool deliverables never surfaced as song cards. The snake_case proto field names are the contract the agent loop’s asset reader consumes and the gateway parses leniently. Don’t flip this back.

The worker never bills

process() only calls ctx.report_usage(units) — raw usage, no money. core-mcp prices and settles against it (cost_base is the reserve-time hold checked before generation; cost_increment is charged per reported unit at settlement in dynamic mode). See The token ledger & money path.

6. Graceful drain

SIGTERM/SIGINT (ACA sends SIGTERM on scale-down and redeploy) sets the stop event and wakes the idle poll loop. The executor shuts down wait=True so in-flight tasks finish, the worker:active key is deleted, and telemetry is flushed so the last task.run spans export before exit. Keep process() synchronous — it runs in the thread pool; do async work inside asyncio.run(...).

TaskContext: the API surface

ctx is how a tool talks to the platform. It carries the task’s scope as read-only properties (user_id, workspace_id — falling back to the global default workspace — project_id, task_id, message_id) and exposes the methods below. Every RPC it makes is internally authenticated (the channel carries the INTERNAL_RPC_SECRET bearer) and scoped to the task.

Method	What it does
emit_progress(pct, msg)	Push a progress event onto the live Redis stream the gateway relays.
emit_partial(data)	Emit an interim structured result (agent envelopes, previews).
set_streaming_url(url)	Publish an interim URL of the file being generated (visible live).
text_stream() / push_text_delta(...)	Stream token deltas + a mutable live-text snapshot.
log(msg)	Emit a log event onto the task stream.
resolve_credentials()	Fetch this task’s resolved provider secret/endpoint/model override.
upload_file(data=…, mime_type=…, role=…)	Persist bytes/path/URL via core-storage → a deduped OutputAsset.
add_asset(url=…, mime_type=…, …)	Append a pre-uploaded asset (orchestrators surfacing a sub-tool result); no re-upload.
report_usage(units)	Record raw usage for core-mcp to price/settle.
get_task(id) / list_messages()	Poll another generation; load the project’s prior chat turns.
create_message(...) / update_message(...)	Persist the durable chat transcript (orchestrator tools).
set_license(...), set_attributes(...), set_display_name(...), link_cover(...), add_lineage(...)	Durable per-file metadata into storage.* side tables — best-effort, never fails a run.
live_audio_ingest(...)	Open a live-to-VOD session: stream bytes, get an HLS preview URL live, then finalize to a permanent asset.

Durable metadata goes in side tables, not the JSONB bag

New per-file data belongs in storage.attributes via ctx.set_attributes(...) (and the typed helpers above), not the freeform metadata blob. These calls are deliberately best-effort — metadata polish must never fail a generation. See Add durable per-file metadata.

The full, per-argument catalogue lives in the TaskContext API reference.

Config & the internal client

The SDK also owns the settings + client wiring workers share instead of re-implementing. SoundverseEnv (import the config singleton) reads the fleet-wide env-var names: INTERNAL_RPC_SECRET (the bearer on every internal call; legacy INTERNAL_AUTH_SECRET is accepted as a fallback alias), REDIS_ADDR / REDIS_PASSWORD, and the nested ServiceRegistryEnv addresses CORE_DATABASE_GRPC, CORE_STORAGE_GRPC, CORE_MCP_GRPC / CORE_MCP_URL. TLS is decided per upstream (CORE_DATABASE_USE_TLS / CORE_STORAGE_USE_TLS, default on — staging/prod front those services with TLS-only ACA ingress on :443); set them to a false value for local plaintext services. SoundverseClient builds the authenticated Connect/gRPC channels from those addresses — never hardcode a service host.

Related

• Add a tool worker — the recipe that sits on top of this runtime
• TaskContext API reference — every ctx.* method, per argument
• Task queue on Postgres — the claim/lease/reaper mechanics
• Tool pricing model — what the config upsert wires up
• core-tool-agent — the orchestrator that needs its own lane
• core-mcp — prices, settles, and advertises registered tools
• Test unmerged soundverse-py changes — the local overlay