Python SDK

TaskDaemon python sdk guide

Python SDK

The Python SDK provides a clean interface for writing TaskDaemon handlers.

Installation

The published Python SDK supports the examples below:

python -m pip install taskdaemon==0.1.0

Alternatively, install a reviewed source revision:

git clone https://github.com/jona62/TaskDaemon-Handlers.git taskdaemon-handlers
python -m pip install ./taskdaemon-handlers/python

Use a reviewed SDK revision or release for deployment. Keep diagnostic logs on stderr; the handler protocol uses stdout for one response per task. A request starts with attempt: 1; retries require a valid retryable error and remaining stored budget.

Basic Usage

from taskdaemon import run, Task, Success, Error

def handler(task: Task) -> Success | Error:
    name = task.task_data.get("name", "World")
    return Success({"message": f"Hello, {name}!"})

run(handler)

Task Object

The Task dataclass contains:

@dataclass
class Task:
    task_id: str      # Unique task identifier
    task_type: str    # Handler type (e.g., "resize")
    task_data: dict   # Input data from client
    attempt: int      # Current attempt number (starts at 1)

Return Types

Success

Return Success with a result (any JSON-serializable value):

return Success({"processed": True, "count": 42})

Error

Return Error with an error message and optional retryable flag:

# Non-retryable error (default)
return Error("Invalid input: missing required field")

# Retryable error (will be retried)
return Error("Service temporarily unavailable", retryable=True)

Complete Example

This image handler needs pillow and requests in addition to the SDK:

import base64
from io import BytesIO
import requests
from PIL import Image
from taskdaemon import run, Task, Success, Error

def handler(task: Task) -> Success | Error:
    data = task.task_data

    # Download image
    resp = requests.get(data["image_url"], timeout=30)
    img = Image.open(BytesIO(resp.content))

    if task.task_type == "resize":
        img = img.resize((data["width"], data["height"]), Image.Resampling.LANCZOS)
    elif task.task_type == "thumbnail":
        size = data["size"]
        img.thumbnail((size, size), Image.Resampling.LANCZOS)

    # Encode as base64
    buffer = BytesIO()
    img.save(buffer, format="PNG")
    b64 = base64.b64encode(buffer.getvalue()).decode()

    return Success({
        "size": [img.width, img.height],
        "format": "png",
        "data": b64
    })

if __name__ == "__main__":
    run(handler)

Error Handling

Exceptions are automatically caught and converted to non-retryable errors:

def handler(task: Task) -> Success | Error:
    # If this raises, it becomes Error(str(exception), retryable=False)
    value = task.task_data["required_field"]
    return Success({"value": value})

For explicit error handling:

def handler(task: Task) -> Success | Error:
    if "required_field" not in task.task_data:
        return Error("Missing required_field")

    try:
        result = external_api_call()
        return Success(result)
    except ConnectionError:
        return Error("API unavailable", retryable=True)
    except ValueError as e:
        return Error(f"Invalid data: {e}")

Dockerfile

FROM python:3.11-slim

RUN python -m pip install taskdaemon==0.1.0 pillow requests

COPY handler.py /app/
WORKDIR /app

CMD ["python", "-u", "handler.py"]

Use python -u for unbuffered output to ensure responses are sent immediately.

Handler Configuration

[handlers.resize]
image = "image-handler:latest"
instances = 4
timeout = 30

[handlers.thumbnail]
image = "image-handler:latest"
instances = 2
timeout = 15

Multiple task types can use the same image - the handler receives task.task_type to differentiate.