Echo

The smallest app that works: two endpoints, no weights and no GPU work, so a failure in it is the platform rather than your model.

Introduction

Nothing here can be slow or run out of memory, which is the point. Deploy it when you want to prove the loop end to end, or when something is failing and you need to know whether the problem is your model or the platform.

The app

model.py

"""The smallest app that works: no weights, no GPU work, two endpoints."""

from __future__ import annotations

from runware_serverless import endpoint, serve


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

    @endpoint
    def echo(self, message: str) -> dict[str, object]:
        return {
            "message": message,
            "uppercase": message.upper(),
            "length": len(message),
        }

    @endpoint
    def reverse_message(self, message: str, repeat: int = 1) -> dict[str, object]:
        repeat = 1 if repeat is None else repeat
        return {
            "message": message,
            "reversed": message[::-1] * repeat,
        }

Deploying it

runware serverless deploy model.py --id echo --gpu-type l40s

Invoking it

The app id you deployed with and the handler name are the whole address. Nothing in echo is slow, so you can wait for the result on the same connection.

import { createClient } from '@runware/sdk'

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

const task = await client.invoke({
  appId: 'echo',
  endpointPath: 'echo',
  payload: {
    message: 'hello'
  }
}, {
  deliveryMethod: 'sync'
})

import asyncio
import os

from runware import InvokeOptions, Runware


async def main():
    client = Runware(api_key=os.environ["RUNWARE_API_KEY"])
    try:
        task = await client.invoke(
            "echo",
            "echo",
            {
                "message": "hello"
            },
            options=InvokeOptions(delivery_method="sync"),
        )
    finally:
        await client.close()


asyncio.run(main())

curl https://api.serverless.runware.ai/v1/apps/echo/invoke-sync/echo \
  -H "Authorization: Bearer $RUNWARE_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "taskId": "7c9e6679-7425-40de-944b-e07fc1f90ae7",
    "payload": {
      "message": "hello"
    }
  }'

echo '{"message":"hello"}' | runware serverless apps invoke echo echo --sync -f -

{
  "taskId": "7c9e6679-7425-40de-944b-e07fc1f90ae7",
  "payload": {
    "message": "hello"
  }
}

The second handler is reverse_message in Python and reverse-message in the URL. Underscores in a method name become hyphens in its path, and nothing else about the name changes.

import { createClient } from '@runware/sdk'

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

const task = await client.invoke({
  appId: 'echo',
  endpointPath: 'reverse-message',
  payload: {
    message: 'hello',
    repeat: 2
  }
}, {
  deliveryMethod: 'sync'
})

import asyncio
import os

from runware import InvokeOptions, Runware


async def main():
    client = Runware(api_key=os.environ["RUNWARE_API_KEY"])
    try:
        task = await client.invoke(
            "echo",
            "reverse-message",
            {
                "message": "hello",
                "repeat": 2
            },
            options=InvokeOptions(delivery_method="sync"),
        )
    finally:
        await client.close()


asyncio.run(main())

curl https://api.serverless.runware.ai/v1/apps/echo/invoke-sync/reverse-message \
  -H "Authorization: Bearer $RUNWARE_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "taskId": "7c9e6679-7425-40de-944b-e07fc1f90ae7",
    "payload": {
      "message": "hello",
      "repeat": 2
    }
  }'

cat <<'EOF' | runware serverless apps invoke echo reverse-message --sync -f -
{
  "message": "hello",
  "repeat": 2
}
EOF

{
  "taskId": "7c9e6679-7425-40de-944b-e07fc1f90ae7",
  "payload": {
    "message": "hello",
    "repeat": 2
  }
}

Testing it before you deploy

serve and endpoint hand the class and its methods back unchanged, so there is nothing to mock and no test client to stand up.

test_model.py

"""Your own tests, written the way you would write any others."""

from model import EchoModel


def test_echo_reports_the_message() -> None:
    model = EchoModel()
    model.load()

    assert model.echo("hi") == {"message": "hi", "uppercase": "HI", "length": 2}


def test_reverse_applies_its_default() -> None:
    model = EchoModel()
    model.load()

    assert model.reverse_message("abc")["reversed"] == "cba"


def test_reverse_repeats() -> None:
    model = EchoModel()
    model.load()

    assert model.reverse_message("abc", repeat=2)["reversed"] == "cbacba"


def test_an_omitted_optional_arrives_as_null() -> None:
    """The platform sends an omitted optional field as an explicit ``null``.

    A Python default does not fire for that, which is why the handler applies
    it again. This is the one thing about the request shape worth pinning.
    """
    model = EchoModel()
    model.load()

    assert model.reverse_message("abc", repeat=None)["reversed"] == "cba"

The whole example is at echo/, with a README of its own.