Skip to content
SoundverseSoundverse 2.0 — Engineering Docs

Add a new generation tool worker

A tool is how the platform generates something — a song, stems, album art, a conversion. Under the hood a tool is a long-running queue worker, not a web service: no HTTP, no ingress. It polls the task queue, runs your model, uploads results, and drains on SIGTERM. You write three things — an input model, an output model, and a process() method — and the WorkerFleet in soundverse-py does registration, task-claiming, heartbeats, uploads, usage reporting, and shutdown.

What you’ll touch

Section titled “What you’ll touch”
  • Directorycore-tool-name/
    • Directoryapp/
      • Directorytools/
        • name.py your input model + output model + tool class
        • __init__.py register your tool in ALL_TOOLS
      • main.py set SERVICE_NAME
    • pyproject.toml set the package name
    • Directory.github/workflows/
      • deploy-aca-staging.yml set service_name
      • deploy-aca-prod.yml set service_name

Steps

Section titled “Steps”

  1. Scaffold. Copy template-core-tool into a new repo core-tool-<name>.

  2. Write app/tools/<name>.py — the three parts:

    app/tools/my_tool.py
    from pydantic import BaseModel, Field
    from soundverse.tools.base import BaseTool, ToolOutput
    from soundverse.tools.context import TaskContext
    

    # 1) INPUT model — a plain pydantic model. Field(description=...) and
    #    constraints (ge/le/min_length…) become the MCP input JSON schema the
    #    agent sees, and are validated before process() runs.
    class MyInput(BaseModel):
        prompt: str = Field(description="What to generate")
        seconds: int = Field(default=30, ge=5, le=120, description="Clip length")
    

    # 2) OUTPUT model — subclass ToolOutput. Field naming is a contract:
    #    a field named `text` → the primary result; OutputFile/AudioFile/
    #    ImageFile/VideoFile (or lists) → auto-uploaded, deduped assets;
    #    everything else → metadata_json.
    class MyOutput(ToolOutput):
        text: str
    

    # 3) The tool class. process() is SYNCHRONOUS (runs in a thread pool).
    class MyTool(BaseTool[MyInput, MyOutput]):
        def process(self, input: MyInput, ctx: TaskContext) -> MyOutput:
            ctx.emit_progress(10, "starting")
            # … call your model …
            ctx.report_usage(input.seconds)   # raw usage; the worker NEVER bills
            return MyOutput(text="done")

    Use the ctx (TaskContext) for everything that reaches the platform: resolve_credentials(), set_streaming_url(), emit_progress(), emit_partial(), upload_file(), report_usage(), live_audio_ingest(), and the user_id / workspace_id / project_id scope.

  3. Declare platform config so the tool gets priced. Pricing is the step people forget — see the dedicated guide. Declare it as class attributes; the fleet upserts it to the registry on boot.

    app/tools/my_tool.py (class attributes)
    from soundverse.tools.config import LicensePricing
    from soundverse_proto.generation.tool.v1 import tool_pb2
    

    class MyTool(BaseTool[MyInput, MyOutput]):
        model = "my-model"
        operation = "generate"
        cost_unit = "second"
        usage_field = "seconds"          # makes the reserve-time hold quantity-aware
        pricing = (
            LicensePricing(
                license=tool_pb2.LICENSE_STANDARD,
                cost_mode=tool_pb2.COST_MODE_DYNAMIC,
                cost_base=10,             # reserve-time hold / floor
                cost_increment=1,         # per reported usage unit
            ),
            # …one entry per license you accept…
        )

  4. Register it. Import your class in app/tools/__init__.py and add an instance to ALL_TOOLS (one worker can serve several tools). app/main.py only reads ALL_TOOLS.

  5. Rename the service in four places — name in pyproject.toml, SERVICE_NAME in app/main.py, and service_name in both deploy-aca-staging.yml and deploy-aca-prod.yml. Delete the example (rm app/tools/example_tool.py tests/test_example_tool.py), then run make check (ruff + pyright strict + pytest).

  6. Registration is automatic on boot. WorkerFleet calls RegisterTool (with input_schema_json + output_schema_json derived from your models), then applies the idempotent UpsertToolConfig / UpsertToolLicensePricing / UpsertToolRateLimits / UpsertToolCredentials. Your code is the source of truth — edit a value, restart the worker, it re-applies.

  7. Restart core-mcp. easy to miss core-mcp caches tool input/output schemas in memory. Redeploying the worker updates the DB registry row, but a changed schema is not picked up until core-mcp is restarted. See the core-mcp field guide.

  8. Deploy. Ensure the org secrets/vars exist (prefixed STAGING_/PROD_; the deploy engine strips the prefix and injects them): INTERNAL_RPC_SECRET, REDIS_ADDR, CORE_DATABASE_GRPC, CORE_STORAGE_GRPC, plus any provider key. Push to the staging/prod branch to deploy. See Deploy a service.

The traps that will actually bite you

Section titled “The traps that will actually bite you”

What you get for free

Section titled “What you get for free”

Because the panel is registry-only and schema-driven, your Pydantic input model is the UI — the SaaS AI Tools panel renders a form from your input_schema_json. No frontend code. You also get task claiming, heartbeats/leases, progress + partial events, deduped uploads, usage reporting, and graceful drain — all from the fleet.

Section titled “Related”