API Reference

This page is the authoritative reference for the kubemq-rayserve API: every exported symbol, the adapter's methods, the extension methods, the autoscaling policy, the metrics dictionary, and the TaskResult status model. For the configuration fields and dependencies, see the configuration reference. For the design model, see Concepts.

Package surface

kubemq-rayserve implements Ray Serve's TaskProcessorAdapter framework — the first non-Celery adapter — and is published as version 1.0.0 (Development Status: Beta) under the MIT license. The public API consists of exactly three symbols, exported from kubemq_rayserve:

from kubemq_rayserve.adapter import KubeMQTaskProcessorAdapter
from kubemq_rayserve.autoscaling import kubemq_queue_depth_policy
from kubemq_rayserve.config import KubeMQAdapterConfig

__all__ = [
    "KubeMQAdapterConfig",
    "KubeMQTaskProcessorAdapter",
    "kubemq_queue_depth_policy",
]

KubeMQTaskProcessorAdapter

KubeMQTaskProcessorAdapter implements Ray Serve's TaskProcessorAdapter ABC (7 abstract + 4 optional methods) plus two extension methods. The constructor stores the config only — no network connections are made until initialize() runs.

from kubemq_rayserve import KubeMQAdapterConfig, KubeMQTaskProcessorAdapter

adapter = KubeMQTaskProcessorAdapter(KubeMQAdapterConfig(address="localhost:50000"))

Core methods

These are the methods Ray Serve's framework calls during the deployment lifecycle.

Convenience aliases drop the _sync suffix and point at the same implementations:

Extension methods

These two methods are not part of the TaskProcessorAdapter ABC. They are KubeMQ-specific additions that exploit Queries and Events.

Method signatures and raises

The two methods most likely to raise on the caller's thread are enqueue_task_sync and query_task_sync. Their exact signatures and exception contracts are below.

enqueue_task_sync

def enqueue_task_sync(
    self,
    task_name: str,
    args: Any = None,
    kwargs: Any = None,
    **options: Any,
) -> TaskResult:
    ...

A "PENDING" result is stored on the result channel before the queue message is sent, so a fast consumer cannot overwrite a SUCCESS result with a late PENDING store. Returns TaskResult(id=task_id, status="PENDING", result=None).

query_task_sync

def query_task_sync(
    self,
    task_name: str,
    args: Any = None,
    kwargs: Any = None,
    timeout: int = 30,
    **options: Any,
) -> TaskResult:
    ...

Sends a KubeMQ Query on the task channel and blocks until the consumer's Query subscription responds. The effective timeout is timeout or config.sync_inference_timeout. The returned TaskResult carries the handler's status (SUCCESS / FAILURE) and result.

kubemq_queue_depth_policy

kubemq_queue_depth_policy is a Ray Serve custom autoscaling policy. It queries KubeMQ queue depth via the Python SDK and returns a desired replica count for each deployment in ctxs.

def kubemq_queue_depth_policy(
    ctxs: dict[str, Any],
    kubemq_address: str = "localhost:50000",
    queue_name: str = "",
    tasks_per_replica: int = 5,
    auth_token: str = "",
) -> tuple[dict[str, int], dict]:
    ...

The function returns a (decisions, state) tuple: decisions maps each deployment_id to a desired replica count, and state carries the observed queue_depth and computed desired_replicas for observability.

The scaling formula is:

desired_replicas = max(1, math.ceil(queue_depth / tasks_per_replica))

There is a hard floor of 1 replica — scale-to-zero is not supported, and Ray Serve's built-in downscale_delay_s handles cooldown when the queue is empty. On any error the policy logs a warning and returns each deployment's current replica count unchanged, so a transient broker outage never forces a scale-to-zero. A module-level, lock-protected client cache reuses a single gRPC connection across the ~10s policy invocation interval. See the autoscaling guide for both the in-process policy and the KEDA path.

from functools import partial
from kubemq_rayserve import kubemq_queue_depth_policy

policy = partial(
    kubemq_queue_depth_policy,
    kubemq_address="kubemq:50000",
    queue_name="inference-tasks",
    tasks_per_replica=10,
)

Metrics

get_metrics_sync() returns a dict[str, Any] with exactly 8 entries. Gauge metrics query KubeMQ live on each call; counter and histogram metrics are tracked in-memory by a thread-safe collector.

metrics = adapter.get_metrics_sync()
# {
#   "queue_depth": 12,
#   "in_flight": 2,
#   "dlq_depth": 0,
#   "tasks_enqueued_total": 148,
#   "tasks_completed_total": {"SUCCESS": 144, "FAILURE": 4},
#   "task_processing_duration_seconds": {"min": 0.04, "max": 1.21, "avg": 0.18, "count": 148},
#   "result_storage_retries_total": 0,
#   "consumer_poll_latency_seconds": 0.003,
# }

Task statuses and TaskResult

Every async API returns a Ray Serve TaskResult with three fields: id, status, and result. The status is one of five string values that track a task through its lifecycle.

The stored result payload (returned by the peek-based backend) carries more than the three TaskResult fields:

{
  "task_id": "f1c2...",
  "status": "SUCCESS",
  "result": { "label": "POSITIVE", "score": 0.95 },
  "created_at": "2026-06-01T10:00:00.000000+00:00",
  "completed_at": "2026-06-01T10:00:00.180000+00:00",
  "error": null
}

get_task_status_sync() returns TaskResult(status="PENDING", result=None) whenever the result channel does not exist yet, so polling a not-yet-processed task is always safe.

Progress event schema

report_progress(task_id, pct, detail="") publishes a KubeMQ Event on a derived channel. Subscribe to that channel to stream progress without touching the result backend.

def long_task(adapter, task_id, items):
    for i, item in enumerate(items):
        process(item)
        adapter.report_progress(task_id, (i + 1) / len(items) * 100, detail=f"item {i + 1}")

See Progress Tracking for the full subscriber-side walkthrough.

Result-channel naming and expiry

Each task's result lives on its own KubeMQ queue channel, named by concatenating the configured prefix with the task id:

{result_channel_prefix}{task_id}

With the default prefix this is rayserve-result-{task_id}. Results are written with expiration_in_seconds = result_expiry_seconds (default 3600), so stale results self-purge after the TTL. Writes use a purge-then-write pattern (ack_all_queue_messages then send_queue_message) so state transitions overwrite cleanly. Storage retries up to 3 times with backoff and never re-raises — for expensive ML inference, losing a result is preferable to reprocessing the task, so on persistent failure the backend falls back to a bounded in-memory dict.

Appendix: kubemq-rayserve vs CeleryTaskProcessorAdapter

Ray Serve's reference adapter is Celery-backed. The table below contrasts the two for teams choosing a task backend.