Two endpoints, two shapes

One app whose endpoints take request shapes with nothing in common, so no single schema covers both.

Introduction

generate takes a prompt and upscale takes an image, so no single request schema covers both. Each handler's signature becomes its own, and the platform routes by path.

What they do share is the app: one queue, one worker pool, one GPU type. That is set on the app rather than in the file, which is why nothing here names hardware or scaling. Two endpoints that genuinely need different GPUs are two apps.

The app

model.py

"""Two endpoints on one app, with request shapes that have nothing in common."""

from __future__ import annotations

from runware_serverless import endpoint, serve

PIXEL_PNG = (
    "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+ip1sAAAAASUVORK5CYII="
)
"""A 1x1 white PNG, base64. Stands in for generated pixels."""

BASE_SIZE = 64
DEFAULT_STEPS = 4
DEFAULT_SCALE = 2


@serve
class ImageTools:
    def load(self) -> None:
        pass

    @endpoint
    def generate(self, prompt: str, steps: int = DEFAULT_STEPS) -> dict[str, object]:
        # An omitted optional field arrives as an explicit null, so the default
        # is applied here as well as in the signature.
        steps = DEFAULT_STEPS if steps is None else steps
        return {
            "handler": "generate",
            "prompt": prompt,
            "steps": steps,
            "width": BASE_SIZE,
            "height": BASE_SIZE,
            "image": PIXEL_PNG,
        }

    @endpoint
    def upscale(self, image: str, scale: int = DEFAULT_SCALE) -> dict[str, object]:
        scale = DEFAULT_SCALE if scale is None else scale
        return {
            "handler": "upscale",
            "scale": scale,
            "width": BASE_SIZE * scale,
            "height": BASE_SIZE * scale,
            "image": image,
        }

Deploying it

runware serverless deploy model.py --id image-tools --gpu-type l40s

Invoking it

Both endpoints are on the same app, so only the last segment of the path changes.

import { createClient } from '@runware/sdk'

const client = await createClient({ apiKey: process.env.RUNWARE_API_KEY })

const task = await client.invoke({
  appId: 'image-tools',
  endpointPath: 'generate',
  payload: {
    prompt: 'a red bicycle'
  }
})

import asyncio
import os

from runware import Runware


async def main():
    client = Runware(api_key=os.environ["RUNWARE_API_KEY"])
    try:
        task = await client.invoke(
            "image-tools",
            "generate",
            {
                "prompt": "a red bicycle"
            },
        )
    finally:
        await client.close()


asyncio.run(main())

curl https://api.serverless.runware.ai/v1/apps/image-tools/invoke-async/generate \
  -H "Authorization: Bearer $RUNWARE_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "taskId": "7c9e6679-7425-40de-944b-e07fc1f90ae7",
    "payload": {
      "prompt": "a red bicycle"
    }
  }'

echo '{"prompt":"a red bicycle"}' | runware serverless apps invoke image-tools generate -f -

{
  "taskId": "7c9e6679-7425-40de-944b-e07fc1f90ae7",
  "payload": {
    "prompt": "a red bicycle"
  }
}

upscale takes an image instead, and nothing in its body carries over from the call above.

import { createClient } from '@runware/sdk'

const client = await createClient({ apiKey: process.env.RUNWARE_API_KEY })

const task = await client.invoke({
  appId: 'image-tools',
  endpointPath: 'upscale',
  payload: {
    image: 'iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+ip1sAAAAASUVORK5CYII=',
    scale: 2
  }
})

import asyncio
import os

from runware import Runware


async def main():
    client = Runware(api_key=os.environ["RUNWARE_API_KEY"])
    try:
        task = await client.invoke(
            "image-tools",
            "upscale",
            {
                "image": "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+ip1sAAAAASUVORK5CYII=",
                "scale": 2
            },
        )
    finally:
        await client.close()


asyncio.run(main())

curl https://api.serverless.runware.ai/v1/apps/image-tools/invoke-async/upscale \
  -H "Authorization: Bearer $RUNWARE_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "taskId": "7c9e6679-7425-40de-944b-e07fc1f90ae7",
    "payload": {
      "image": "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+ip1sAAAAASUVORK5CYII=",
      "scale": 2
    }
  }'

cat <<'EOF' | runware serverless apps invoke image-tools upscale -f -
{
  "image": "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+ip1sAAAAASUVORK5CYII=",
  "scale": 2
}
EOF

{
  "taskId": "7c9e6679-7425-40de-944b-e07fc1f90ae7",
  "payload": {
    "image": "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+ip1sAAAAASUVORK5CYII=",
    "scale": 2
  }
}

steps and scale both have defaults in their signature, so you can leave them out. An omitted optional field arrives as an explicit null rather than as a missing key, which is why each handler applies its default a second time.

Testing it before you deploy

test_model.py

"""Each endpoint is a method, so each test is a call."""

from model import BASE_SIZE, PIXEL_PNG, ImageTools


def test_generate_reports_its_own_handler() -> None:
    model = ImageTools()
    model.load()

    result = model.generate("a red bicycle")

    assert result["handler"] == "generate"
    assert result["steps"] == 4
    assert result["image"] == PIXEL_PNG


def test_upscale_takes_a_different_shape_entirely() -> None:
    model = ImageTools()
    model.load()

    result = model.upscale(PIXEL_PNG, scale=4)

    assert result["handler"] == "upscale"
    assert result["width"] == BASE_SIZE * 4


def test_each_handler_applies_its_own_default() -> None:
    model = ImageTools()
    model.load()

    assert model.generate("a red bicycle", steps=None)["steps"] == 4
    assert model.upscale(PIXEL_PNG, scale=None)["scale"] == 2

The whole example is at image-tools/, with a README of its own.