Typed requests and responses
Describe your API with Pydantic models so callers see every field, and so defaults reach your handler instead of nulls.
Introduction
The build derives your app's contract from your handler signatures, so your signature is not a private detail. It is the documentation your callers read, and the validation the platform applies before your code runs.
The work here is small on purpose, reading an image's metadata, so the request and the response are the subject.
The app
"""A request and a response the platform can describe, which is what pydantic buys."""
from __future__ import annotations
import base64
import io
from enum import Enum
from pydantic import BaseModel, Field
from runware_serverless import endpoint, serve
MAX_PALETTE = 8
class Orientation(str, Enum):
landscape = "landscape"
portrait = "portrait"
square = "square"
class Source(BaseModel):
image: str = Field(description="The image to read, base64 encoded, with no data URI prefix.")
palette: bool = Field(default=False, description="Also return the most common colors.")
class Size(BaseModel):
width: int
height: int
class Color(BaseModel):
red: int
green: int
blue: int
class Report(BaseModel):
format: str = Field(description="The container the bytes are in, such as PNG or JPEG.")
size: Size
orientation: Orientation
bytes: int = Field(description="How many bytes the image decoded to.")
palette: list[Color] = Field(default_factory=list)
@serve
class TypedEndpoints:
requirements = ["pillow==13.0.0"]
def load(self) -> None:
pass
@endpoint
def describe(self, source: Source) -> Report:
# Here rather than at the top of the file. The build imports this file
# with the platform's own dependencies, not the app's, so nothing in
# requirements exists until a worker runs.
from PIL import Image
raw = base64.b64decode(source.image, validate=True)
with Image.open(io.BytesIO(raw)) as image:
width, height = image.size
return Report(
format=image.format or "unknown",
size=Size(width=width, height=height),
orientation=_orientation(width, height),
bytes=len(raw),
palette=_palette(image) if source.palette else [],
)
def _orientation(width: int, height: int) -> Orientation:
if width > height:
return Orientation.landscape
if height > width:
return Orientation.portrait
return Orientation.square
def _palette(image: object) -> list[Color]:
# Quantizing rather than counting pixels: getcolors returns nothing at all
# once an image holds more distinct colors than the limit it was given, which
# every photograph does.
quantized = image.convert("RGB").quantize(colors=MAX_PALETTE) # pyright: ignore[reportAttributeAccessIssue]
channels = quantized.getpalette() or []
by_frequency = sorted(quantized.getcolors() or [], reverse=True)
return [
Color(
red=channels[index * 3],
green=channels[index * 3 + 1],
blue=channels[index * 3 + 2],
)
for _, index in by_frequency
]Parameters and model fields
An optional parameter publishes its type, and nothing else. No default, and not a word of prose:
{ "steps": { "type": "integer" } }A field on a Pydantic model publishes both:
{
"palette": {
"type": "boolean",
"default": false,
"description": "Also return the most common colors.",
"title": "Palette"
}
}That difference reaches your handler. The platform fills an omitted optional from the published schema, so a parameter whose default it never saw reaches you as an explicit null and you apply the default a second time. A field's default is in the schema, so palette arrives as false and you re-apply nothing.
This is the one reason to prefer a model even for a single optional flag. It is the difference between palette: bool = False meaning what it says and meaning palette: bool | None.
The response schema
Your return annotation is the response schema, and the build reads nothing else.
| The handler returns | Callers are told |
|---|---|
dict[str, str] | {"type": "object"}, and nothing about what is in it |
| a type the build cannot resolve | {}, meaning anything at all |
| a Pydantic model | every field, with its type, its prose and its nesting |
A model holding another model publishes both, and an enum publishes its values, so orientation documents itself as one of three known strings instead of as a string.
Deploying it
runware serverless deploy model.py --id typed-endpoints --gpu-type rtx-pro-6000No weights and no GPU work, so the hardware only has to exist.
Calling it
The parameter is a model, so its fields sit under the parameter's own name. That extra level in the body is what the typed contract costs.
import { createClient } from '@runware/sdk'
const client = await createClient({ apiKey: process.env.RUNWARE_API_KEY })
const task = await client.invoke({
appId: 'typed-endpoints',
endpointPath: 'describe',
payload: {
source: {
image: '<base64 PNG>',
palette: true
}
}
})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(
"typed-endpoints",
"describe",
{
"source": {
"image": "<base64 PNG>",
"palette": True
}
},
)
finally:
await client.close()
asyncio.run(main())curl https://api.serverless.runware.ai/v1/apps/typed-endpoints/invoke-async/describe \
-H "Authorization: Bearer $RUNWARE_API_KEY" \
-H "Content-Type: application/json" \
-d '{
"taskId": "7c9e6679-7425-40de-944b-e07fc1f90ae7",
"payload": {
"source": {
"image": "<base64 PNG>",
"palette": true
}
}
}'cat <<'EOF' | runware serverless apps invoke typed-endpoints describe -f -
{
"source": {
"image": "<base64 PNG>",
"palette": true
}
}
EOF{
"taskId": "7c9e6679-7425-40de-944b-e07fc1f90ae7",
"payload": {
"source": {
"image": "<base64 PNG>",
"palette": true
}
}
}The report comes back at output.predictions[0], with every field the schema promised.
Omitting palette answers the same way, with an empty palette instead of a validation error. That is the published default arriving as false, since a null would not validate against bool at all.
Measured on an RTX PRO 6000: the first call takes about 1m45s and answers 202 with the task still pending, because a sync invoke has a deadline and the worker is starting behind it. The next call answers 200 in 2 seconds.