---
title: Taking an image as input | Runware Docs
url: https://runware.ai/docs/serverless/examples/watermark
description: Accept an image and return one, using the annotations that decode base64 into pixels before your handler runs.
relatedDocuments:
  - https://runware.ai/docs/serverless/writing-a-model
---
## 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**:

```python
"""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:

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

> [!WARNING]
> 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

```bash
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

**TypeScript**:

```typescript
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
    }
  }
})
```

**Python**:

```python
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**:

```bash
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
      }
    }
  }'
```

**CLI**:

```bash
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
```

**JSON**:

```json
{
  "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`.

> [!NOTE]
> 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**.