Taking an image as input

Accept an image and return one, using the annotations that decode base64 into pixels before your handler runs.

Introduction

An image reaches your endpoint as a base64 string, and decoding it is work every image app does. The package has two annotations that do it for you, before your handler runs.

This app lays a mark over an image, and the work is a stand-in. It is pixel arithmetic on the CPU, nothing in model.py touches the GPU or loads a weight, and you would not deploy it. It is the smallest thing that needs a real image on both sides, so the image can be the whole subject.

What it shows is what every app taking an image has in common, whether yours upscales, edits or runs a ControlNet pass.

The app

model.py

"""An image arrives as base64, and the annotation is what turns it into pixels."""

from __future__ import annotations

import base64
import io
from enum import Enum

from pydantic import BaseModel, Field

from runware_serverless import serve
from runware_serverless.schema import Base64RGBAImage, Base64RGBImage

DEFAULT_OPACITY = 0.65
DEFAULT_MARGIN = 24
OPAQUE = 255


class Corner(str, Enum):
    top_left = "top-left"
    top_right = "top-right"
    bottom_left = "bottom-left"
    bottom_right = "bottom-right"


class Layers(BaseModel):
    base: Base64RGBImage = Field(description="The image to stamp, base64 encoded.")
    mark: Base64RGBAImage = Field(description="The mark to lay over it, base64 encoded.")
    corner: Corner = Field(default=Corner.bottom_right, description="Where the mark goes.")
    opacity: float = Field(default=DEFAULT_OPACITY, ge=0, le=1)
    margin: int = Field(default=DEFAULT_MARGIN, ge=0, description="Gap from the two edges, in pixels.")


class Stamped(BaseModel):
    image: str = Field(description="The result as a base64 PNG.")
    width: int
    height: int


@serve
class Watermark:
    requirements = ["pillow==13.0.0"]

    def load(self) -> None:
        pass

    def predict(self, layers: Layers) -> Stamped:
        canvas = layers.base.convert("RGBA")
        mark = _faded(layers.mark, layers.opacity)
        canvas.alpha_composite(mark, _position(canvas.size, mark.size, layers))

        flattened = canvas.convert("RGB")
        buffer = io.BytesIO()
        flattened.save(buffer, format="PNG")

        return Stamped(
            image=base64.b64encode(buffer.getvalue()).decode("ascii"),
            width=flattened.width,
            height=flattened.height,
        )


def _faded(mark: object, opacity: float) -> object:
    # Converted here rather than trusted: the annotation on `mark` preserves the
    # mode it arrived in, so a mark with no alpha channel reaches this unchanged.
    faded = mark.convert("RGBA")  # pyright: ignore[reportAttributeAccessIssue]
    faded.putalpha(faded.getchannel("A").point(lambda value: round(value * opacity)))

    return faded


def _position(
    canvas: tuple[int, int],
    mark: tuple[int, int],
    layers: Layers,
) -> tuple[int, int]:
    right = canvas[0] - mark[0] - layers.margin
    bottom = canvas[1] - mark[1] - layers.margin

    return {
        Corner.top_left: (layers.margin, layers.margin),
        Corner.top_right: (right, layers.margin),
        Corner.bottom_left: (layers.margin, bottom),
        Corner.bottom_right: (right, bottom),
    }[layers.corner]

What the annotation does

Base64RGBImage and Base64RGBAImage validate the incoming string into a PIL.Image, so layers.base is already pixels when your handler runs. The field publishes itself as a described string instead of an opaque one:

{ "type": "string", "format": "base64", "description": "Image in base64 format" }

Put them on a field of a model, never on a parameter. Coercion looks for a Pydantic model to validate the parameter against, and these are images, not models, so a bare def predict(self, image: Base64RGBImage) is never coerced.

Your published contract goes with it. The same annotation, on a bare parameter and on a field:

	published	the handler gets
`image: Base64RGBImage`	`{}`	`str`
the same type on a field	a described base64 string	a `PIL.Image`

Nothing fails either way. One of them just answers a different question than the one it was asked, and tells your callers nothing while it does it.

What the names convert

Base64RGBImage converts to RGB. Anything becomes three channels, and the container the bytes arrived in is gone with the conversion, so format is None afterwards.

Base64RGBAImage converts nothing. A grayscale PNG stays grayscale and an image with no alpha channel does not gain one. It is the right choice for a mark whose transparency you want kept, and it is why this handler converts the mark itself before compositing instead of assuming one arrived.

On the way out

Both refuse to serialize, so a response field annotated with one sends a placeholder instead of your image. That is why Stamped.image is a plain string the handler encodes.

Coercion is a request-side feature. The value your handler returns goes out as it is.

Deploying it

runware serverless deploy model.py --id watermark --gpu-type rtx-pro-6000

The app declares Pillow, because the annotations need it at serving time. Pinning it means your app behaves the same wherever it runs, whatever the runtime happens to carry.

Calling it

import { createClient } from '@runware/sdk'

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

const task = await client.invoke({
  appId: 'watermark',
  endpointPath: 'predict',
  payload: {
    layers: {
      base: '<base64 photo>',
      mark: '<base64 logo with alpha>',
      corner: 'bottom-right',
      opacity: 0.5
    }
  }
})

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(
            "watermark",
            "predict",
            {
                "layers": {
                    "base": "<base64 photo>",
                    "mark": "<base64 logo with alpha>",
                    "corner": "bottom-right",
                    "opacity": 0.5
                }
            },
        )
    finally:
        await client.close()


asyncio.run(main())

curl https://api.serverless.runware.ai/v1/apps/watermark/invoke-async/predict \
  -H "Authorization: Bearer $RUNWARE_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "taskId": "7c9e6679-7425-40de-944b-e07fc1f90ae7",
    "payload": {
      "layers": {
        "base": "<base64 photo>",
        "mark": "<base64 logo with alpha>",
        "corner": "bottom-right",
        "opacity": 0.5
      }
    }
  }'

cat <<'EOF' | runware serverless apps invoke watermark predict -f -
{
  "layers": {
    "base": "<base64 photo>",
    "mark": "<base64 logo with alpha>",
    "corner": "bottom-right",
    "opacity": 0.5
  }
}
EOF

{
  "taskId": "7c9e6679-7425-40de-944b-e07fc1f90ae7",
  "payload": {
    "layers": {
      "base": "<base64 photo>",
      "mark": "<base64 logo with alpha>",
      "corner": "bottom-right",
      "opacity": 0.5
    }
  }
}

The result is a base64 PNG at output.predictions[0].image, the same size as the base.

Sending only base and mark works, and the mark lands bottom right at the declared opacity, not at nothing, because each of those defaults is published and the platform sends it. A null would not have validated against Corner, float or int.

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 of roughly 105 seconds. The next call answers 200 in 2 seconds.