Endpoints

Full reference for @app.endpoint — decorator params, media file types, progress reporting, and queue behaviour.

@app.endpoint

Every public method on your APIPod app must be decorated with @app.endpoint. The decorator registers the route, validates inputs/outputs, and enqueues the job for async execution.

from apipod import APIPod
from media_toolkit import ImageFile

app = APIPod()

@app.endpoint(
    "/generate",
    queue_size=50,
)
def generate(prompt: str, width: int = 512, height: int = 512) -> ImageFile:
    image = my_model(prompt, width=width, height=height)
    return image

Decorator Parameters

Parameter	Type	Default	Description
`path`	`str`	`required`	URL route for this endpoint, e.g. "/generate".
`methods`	`list[str] \| None`	`None`	HTTP methods allowed (defaults to ["POST"] when None). POST is required for file uploads.
`max_upload_file_size_mb`	`int`	`None`	Maximum upload file size in MB.
`queue_size`	`int`	`500`	Maximum number of jobs waiting in the queue. Returns 429 when full.
`use_queue`	`bool`	`None`	Override queue auto-detection. Set False to force a synchronous endpoint even when a queue is configured.
`summary`	`str`	`None`	OpenAPI summary string shown in Swagger UI.
`tags`	`list[str]`	`None`	OpenAPI tags for grouping endpoints in Swagger UI.

Supported Media File Types

Use media-toolkit type hints in your function signature. APIPod automatically deserialises incoming base64 or URL payloads into the correct Python object.

Type Hint	Accepts	Python Object
`ImageFile`	base64, URL, file path, PIL.Image	`media_toolkit.ImageFile`
`AudioFile`	base64, URL, file path, bytes	`media_toolkit.AudioFile`
`VideoFile`	base64, URL, file path, bytes	`media_toolkit.VideoFile`
`str`	Plain JSON string	`str`
`int / float / bool`	Plain JSON scalar	`int \| float \| bool`

from apipod import APIPod
from media_toolkit import ImageFile, AudioFile

app = APIPod()

@app.endpoint("/swap-face")
def swap_face(source: ImageFile, target: ImageFile) -> ImageFile:
    return face_swap_model(source.to_pil(), target.to_pil())

@app.endpoint("/transcribe")
def transcribe(audio: AudioFile) -> dict:
    text = whisper_model(audio.to_bytes())
    return {"transcript": text}

Reporting Progress with JobProgress

Long-running tasks can stream progress back to the caller. Import JobProgress and call it inside your function. The SDK surfaces this as job.progress (0.0 – 1.0).

from apipod import APIPod, JobProgress
from media_toolkit import ImageFile

# Queue requires compute="serverless" and provider="localhost" (or a queue-enabling provider)
app = APIPod(compute="serverless", provider="localhost")

@app.endpoint("/generate")
def generate(job_progress: JobProgress, prompt: str, steps: int = 50) -> ImageFile:
    for i in range(steps):
        image = my_pipeline_step(prompt, i)
        job_progress.set_status((i + 1) / steps, f"Step {i + 1}/{steps}")
    return image

Progress is pushed over SSE while the job is running. Clients that call .get_result() directly will still receive the final value — they will not see intermediate updates.

Queue Behaviour

APIPod's job queue is not enabled by default. It activates when compute="serverless" is set (e.g. APIPod(compute="serverless", provider="localhost")). When enabled, every endpoint call returns a job_id immediately and jobs are executed in a FIFO queue. The default APIPod() (dedicated compute) uses a standard synchronous FastAPI router with no queue.

Config Key	Default	Description
`max_workers`	`1`	Number of concurrent jobs per replica.
`queue_size`	`500`	Max queued jobs before 429 is returned.
`job_ttl`	`3600`	Seconds a finished job result is retained.

{
  "queue": {
    "max_workers": 1,
    "queue_size": 500,
    "job_ttl": 3600
  }
}

Setting max_workers above 1 on a GPU service may cause out-of-memory errors if your model fills VRAM. Benchmark before increasing concurrency.

Return Types

The return value of your function is serialised to JSON automatically. media-toolkit objects are base64-encoded; plain Python dicts, lists, and scalars pass through unchanged.

from media_toolkit import ImageFile

@app.endpoint("/generate")
def generate(prompt: str) -> ImageFile:
    img = my_model(prompt)
    return img                  # serialised to base64 PNG automatically

@app.endpoint("/info")
def info() -> dict:
    return {"model": "my-model", "version": "1.0"}  # plain dict → JSON

Getting Started

Build