Overview

Long-running inference — large language model generations, batch jobs, or multi-stage pipelines — can take seconds to minutes to complete. Rather than leaving callers blind until a result lands, kubemq-rayserve lets a task handler publish incremental progress updates that any external client can subscribe to in real time.

Progress tracking is built on KubeMQ Events (the fire-and-forget pub/sub pattern exposed through the SDK's PubSubClient). The adapter publishes each update as an EventMessage on a dedicated {queue_name}.progress channel, and subscribers receive them live. This is one of the capabilities the Celery adapter lacks out of the box — with Celery, progress reporting requires custom signals, whereas here it is a single built-in method.

Progress events use KubeMQ Events, which are fire-and-forget. They are delivered to whatever subscribers are connected at publish time and are not persisted. The task result itself is always durable through the queue-peek result backend — progress events are an observability stream layered on top of it.

How `report_progress` works

report_progress is an extension method on KubeMQTaskProcessorAdapter. It is meant to be called from inside a task handler to announce how far along the work is.

signature

def report_progress(self, task_id: str, pct: float, detail: str = "") -> None:
    ...

Parameter	Type	Description
`task_id`	`str`	Task identifier the progress update belongs to
`pct`	`float`	Progress percentage, `0`–`100`
`detail`	`str`	Optional human-readable detail string

When called, the adapter publishes an EventMessage on the {queue_name}.progress channel. The event body is a JSON object and the event also carries lightweight tags for cheap, body-free filtering:

Body: {"task_id": ..., "pct": ..., "detail": ...}
Tags: {"task_id": ..., "pct": ...} (the pct tag is the integer percentage as a string)

Crucially, publishing is best-effort: if the underlying send_event call fails, the adapter logs a warning rather than raising. A failed progress update will never crash or fail the task it is reporting on.

adapter.py — report_progress

def report_progress(self, task_id: str, pct: float, detail: str = "") -> None:
    """Publish a progress update via KubeMQ Events.

    Callable from within task handlers to report progress.
    """
    body = json.dumps({"task_id": task_id, "pct": pct, "detail": detail}).encode("utf-8")
    event = EventMessage(
        channel=f"{self._queue_name}.progress",
        body=body,
        tags={"task_id": task_id, "pct": str(int(pct))},
    )
    try:
        self._pubsub_client.send_event(event)
    except Exception as exc:
        logger.warning("Failed to send progress event: %s", exc)

v1.0.0 limitation. Progress events are fire-and-forget KubeMQ Events — they are not persisted, so a subscriber that connects after an event was published will not see it. In addition, a handler must know its own task_id to report against, and in v1.0.0 that id is threaded explicitly into the handler as a keyword argument (the task_id kwarg workaround shown below) rather than being injected automatically.

Prerequisites

Progress tracking requires a running KubeMQ broker. Start one locally with Docker:

docker run -d \  --name kubemq \  -p 50000:50000 \  -p 9090:9090 \  -e KUBEMQ_TOKEN=YOUR_LICENSE_KEY \  europe-docker.pkg.dev/kubemq/images/kubemq:next

Port 50000 is KubeMQ's gRPC endpoint used by both the adapter and the PubSubClient. Port 9090 exposes the shared HTTP server used by connector endpoints such as KEDA scaling.

Reporting progress from a handler

A handler receives its task_id as an explicit keyword argument and calls adapter.report_progress at meaningful stage boundaries. The handler below reports 0, 33, 66, and 100 percent, each with a short detail string describing the phase. The producer pre-generates the task_id and passes it into the task via kwargs so the handler and any subscriber share the same id.

progress_subscriber.py — analyze handler

from kubemq_rayserve import KubeMQAdapterConfig, KubeMQTaskProcessorAdapter

# Module-level adapter reference so the handler can reach it
adapter: KubeMQTaskProcessorAdapter | None = None


def analyze(data: str, task_id: str = "") -> dict:
    """Handler that reports progress via explicit task_id kwarg."""
    assert adapter is not None
    adapter.report_progress(task_id, 0, "Starting analysis")
    time.sleep(0.3)

    adapter.report_progress(task_id, 33, "Phase 1 complete")
    time.sleep(0.3)

    adapter.report_progress(task_id, 66, "Phase 2 complete")
    time.sleep(0.3)

    adapter.report_progress(task_id, 100, "Analysis finished")
    return {"analyzed": data, "phases": 2}

The handler is registered and started like any other task handler, then enqueued with a known task_id threaded through kwargs:

progress_subscriber.py — enqueue with a known task_id

import uuid

adapter.register_task_handler(analyze, name="analyze")
adapter.start_consumer()

# Pre-generate a known task_id and pass it as a kwarg
known_task_id = f"task-{uuid.uuid4().hex[:8]}"
result = adapter.enqueue_task_sync(
    "analyze",
    kwargs={"data": "test input", "task_id": known_task_id},
)
print(f"enqueued task_id={result.id}")

Subscribing to progress events

An external client observes progress by building a PubSubClient and subscribing to the {channel}.progress channel with an EventsSubscription. The on_receive_event_callback decodes event.body as the update is delivered. Because progress runs on KubeMQ Events, the subscriber sees updates live as the handler emits them.

progress_subscriber.py — external subscriber

from kubemq import (
    CancellationToken,
    ClientConfig,
    EventReceived,
    EventsSubscription,
    PubSubClient,
)

progress_channel = f"{channel}.progress"
received_events: list[str] = []

def on_event(event: EventReceived) -> None:
    body = event.body.decode("utf-8") if event.body else ""
    received_events.append(body)
    print(f"  [subscriber] progress event: {body}")

def on_error(err: str) -> None:
    print(f"  [subscriber] error: {err}")

cancel_token = CancellationToken()
pubsub_config = ClientConfig(address=BROKER, client_id=f"progress-sub-{uuid.uuid4().hex[:8]}")
pubsub_client = PubSubClient(config=pubsub_config)

subscription = EventsSubscription(
    channel=progress_channel,
    group="",
    on_receive_event_callback=on_event,
    on_error_callback=on_error,
)
pubsub_client.subscribe_to_events(subscription, cancel_token)
print(f"Subscribed to progress channel: {progress_channel}")

Subscribe before enqueuing the task. Since the events are not persisted, any update published while no subscriber is connected is lost. The example pauses briefly (time.sleep(0.5)) after subscribing to let the subscription establish before the work begins.

When the subscriber and adapter run together — as in examples/quickstart/progress_hello_world.py — the callback can parse the JSON body directly to read pct and detail:

progress_hello_world.py — decoding pct and detail

import json

progress_events: list[dict] = []

def on_event(event) -> None:
    try:
        data = json.loads(event.body)
        progress_events.append(data)
        print(f"  [Progress] pct={data.get('pct', 0)}% detail={data.get('detail', '')}")
    except Exception:
        pass

subscription = EventsSubscription(
    channel=f"{channel}.progress",
    on_receive_event_callback=on_event,
    on_error_callback=on_error,
)
pubsub.subscribe_to_events(subscription, cancel_token)

Rich detail messages

The detail argument is free-form text, so it can carry descriptive, item-level status such as "Processing image 3/10". The handler in examples/progress_tracking/progress_with_detail.py walks a simulated batch and computes a percentage per item while emitting a descriptive detail string at each step.

progress_with_detail.py — per-item detail strings

def process_images(image_count: int = 5, task_id: str = "") -> dict:
    """Handler that processes images with rich progress detail strings."""
    assert adapter is not None

    adapter.report_progress(task_id, 0, f"Starting batch of {image_count} images")
    time.sleep(0.2)

    processed = 0
    for i in range(1, image_count + 1):
        pct = int((i / image_count) * 80) + 10  # 10% to 90%
        adapter.report_progress(task_id, pct, f"Processing image {i}/{image_count}")
        time.sleep(0.2)
        processed += 1

    adapter.report_progress(task_id, 95, f"Generating summary for {processed} images")
    time.sleep(0.2)

    adapter.report_progress(task_id, 100, "Batch complete")
    return {"processed_count": processed, "status": "success"}

Multi-stage pipeline progress

For pipelines that span multiple handlers, each stage reports progress independently against its own task_id while contributing to a shared percentage range. The two-stage example in examples/progress_tracking/multi_stage_progress.py splits the work so preprocessing covers 0–50% and inference covers 50–100%, producing a single continuous progress arc across the pipeline.

multi_stage_progress.py — two-stage progress ranges

def preprocess(data: str, task_id: str = "") -> dict:
    """Stage 1: Preprocessing with progress 0-50%."""
    assert adapter is not None

    adapter.report_progress(task_id, 0, "Stage 1: Starting preprocessing")
    time.sleep(0.2)
    adapter.report_progress(task_id, 10, "Stage 1: Validating input")
    time.sleep(0.2)
    adapter.report_progress(task_id, 25, "Stage 1: Normalizing data")
    time.sleep(0.2)
    adapter.report_progress(task_id, 40, "Stage 1: Tokenizing")
    time.sleep(0.2)
    adapter.report_progress(task_id, 50, "Stage 1: Preprocessing complete")
    return {"preprocessed": data.lower().strip(), "tokens": len(data.split())}


def inference(preprocessed: str, token_count: int = 0, task_id: str = "") -> dict:
    """Stage 2: Inference with progress 50-100%."""
    assert adapter is not None

    adapter.report_progress(task_id, 55, "Stage 2: Loading model")
    time.sleep(0.2)
    adapter.report_progress(task_id, 65, "Stage 2: Running inference")
    time.sleep(0.3)
    adapter.report_progress(task_id, 80, "Stage 2: Post-processing results")
    time.sleep(0.2)
    adapter.report_progress(task_id, 95, "Stage 2: Formatting output")
    time.sleep(0.1)
    adapter.report_progress(task_id, 100, "Stage 2: Inference complete")
    return {"prediction": "positive", "confidence": 0.94, "token_count": token_count}

Each stage is enqueued in turn, with the output of stage 1 feeding stage 2. Both stages publish to the same {queue_name}.progress channel, so a single subscriber observes the entire pipeline:

multi_stage_progress.py — chaining the stages

# Stage 1: preprocess
stage1_task_id = f"task-{uuid.uuid4().hex[:8]}"
result1 = adapter.enqueue_task_sync(
    "preprocess",
    kwargs={"data": input_data, "task_id": stage1_task_id},
)
# ... poll result1 until SUCCESS/FAILURE, capturing stage1_output ...

# Stage 2: inference, fed by stage 1 output
stage2_task_id = f"task-{uuid.uuid4().hex[:8]}"
result2 = adapter.enqueue_task_sync(
    "inference",
    kwargs={
        "preprocessed": stage1_output.get("preprocessed", ""),
        "token_count": stage1_output.get("tokens", 0),
        "task_id": stage2_task_id,
    },
)

End-to-end quickstart

examples/quickstart/progress_hello_world.py ties the pieces together in a single script: it registers a handler that reports 25, 50, 75, and 100 percent, subscribes to the progress channel, enqueues a task with a known task_id, and polls for the final result while progress events stream in.

progress_hello_world.py — handler

def process_data(data: str, task_id: str = "") -> dict:
    """Handler that reports progress at each stage."""
    if adapter is not None and task_id:
        adapter.report_progress(task_id, 25, "Starting processing")
    time.sleep(0.3)

    if adapter is not None and task_id:
        adapter.report_progress(task_id, 50, "Halfway done")
    time.sleep(0.3)

    if adapter is not None and task_id:
        adapter.report_progress(task_id, 75, "Almost done")
    time.sleep(0.3)

    if adapter is not None and task_id:
        adapter.report_progress(task_id, 100, "Complete")
    return {"processed": data, "status": "done"}

Run it against a live broker:

terminal

uv run python examples/quickstart/progress_hello_world.py

Expected output interleaves the streamed progress events with the final result:

output

Enqueuing task with known task_id=task-1a2b3c4d...
  enqueued: task_id=...
  [Progress] pct=25% detail=Starting processing
  [Progress] pct=50% detail=Halfway done
  [Progress] pct=75% detail=Almost done
  [Progress] pct=100% detail=Complete

Result: status=SUCCESS result={'processed': 'hello', 'status': 'done'}

Total progress events received: 4

Progress Tracking

Overview

How `report_progress` works

Prerequisites

Reporting progress from a handler

Subscribing to progress events

Rich detail messages

Multi-stage pipeline progress

End-to-end quickstart

Next steps

Async Inference

API Reference

On this page