# API Reference (/integrations/rayserve/reference/api)



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](/integrations/rayserve/reference/configuration). For the design model, see [Concepts](/integrations/rayserve/concepts).

## Package surface [#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`:

```python title="kubemq_rayserve/__init__.py"
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",
]
```

| Symbol                       | Kind                            | Purpose                                                                                             |
| ---------------------------- | ------------------------------- | --------------------------------------------------------------------------------------------------- |
| `KubeMQAdapterConfig`        | Pydantic `BaseModel`            | Connection and behavior configuration ([reference](/integrations/rayserve/reference/configuration)) |
| `KubeMQTaskProcessorAdapter` | `TaskProcessorAdapter` subclass | Queue-based async + Query-based sync inference engine                                               |
| `kubemq_queue_depth_policy`  | Function                        | Ray Serve custom autoscaling policy driven by queue depth                                           |

<Callout type="info">
  Everything outside these three symbols (the `_internal` package, `ResultBackend`,
  `MetricsCollector`) is private and may change between releases. Do not import it.
</Callout>

## KubeMQTaskProcessorAdapter [#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.

```python
from kubemq_rayserve import KubeMQAdapterConfig, KubeMQTaskProcessorAdapter

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

### Core methods [#core-methods]

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

| Method                                                            | Description                                                                                                  |
| ----------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------ |
| `initialize(consumer_concurrency=1, **kwargs)`                    | Create SDK clients and store concurrency + `task_processor_config` fields. Must run before any other method. |
| `register_task_handler(func, name=None)`                          | Register a callable as a named task handler. Falls back to `func.__name__` when `name` is `None`.            |
| `enqueue_task_sync(task_name, args=None, kwargs=None, **options)` | Serialize the task, send to the KubeMQ queue, return `TaskResult(status="PENDING")`.                         |
| `get_task_status_sync(task_id)`                                   | Peek the result channel and return a `TaskResult` with the current status.                                   |
| `start_consumer(**kwargs)`                                        | Start the queue polling thread plus the Query subscription thread.                                           |
| `stop_consumer(timeout=10.0)`                                     | Cancel the token, stop threads, and close SDK clients.                                                       |
| `health_check()`                                                  | Return `True` if the broker is reachable via SDK `ping()`.                                                   |
| `cancel_task_sync(task_id)`                                       | Soft cancel — overwrite the result with `CANCELLED`. Always returns `True`.                                  |
| `get_metrics_sync()`                                              | Return the 8-metric dict (see [Metrics](#metrics)).                                                          |

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

| Alias             | Target                 |
| ----------------- | ---------------------- |
| `enqueue_task`    | `enqueue_task_sync`    |
| `get_task_status` | `get_task_status_sync` |
| `get_metrics`     | `get_metrics_sync`     |
| `cancel_task`     | `cancel_task_sync`     |

### Extension methods [#extension-methods]

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

| Method                                                                      | Description                                                                                         |
| --------------------------------------------------------------------------- | --------------------------------------------------------------------------------------------------- |
| `query_task_sync(task_name, args=None, kwargs=None, timeout=30, **options)` | Synchronous inference via a KubeMQ Query. Blocks until the handler responds or the timeout elapses. |
| `report_progress(task_id, pct, detail="")`                                  | Publish a progress event via KubeMQ Events on `{queue_name}.progress`.                              |

## Method signatures and raises [#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 [#enqueue_task_sync]

```python
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)`.

| Raises                      | When                                          |
| --------------------------- | --------------------------------------------- |
| `ValueError`                | `task_name` is empty.                         |
| `TypeError`                 | `args` or `kwargs` are not JSON-serializable. |
| `KubeMQConnectionError`     | The broker is unreachable.                    |
| `KubeMQAuthenticationError` | The auth token is invalid.                    |
| `RuntimeError`              | `initialize()` was not called first.          |

<Callout type="warn">
  Arguments are serialized with `json.dumps`. NumPy arrays, tensors, and other non-JSON types raise
  `TypeError` with guidance to convert first (for example `ndarray.tolist()`). Pluggable serializers
  are planned for a future release.
</Callout>

### query\_task\_sync [#query_task_sync]

```python
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.

| Raises               | When                                          |
| -------------------- | --------------------------------------------- |
| `KubeMQTimeoutError` | The query times out with no response.         |
| `TypeError`          | `args` or `kwargs` are not JSON-serializable. |

## kubemq\_queue\_depth\_policy [#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`.

```python
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]:
    ...
```

| Argument            | Type             | Default             | Description                                       |
| ------------------- | ---------------- | ------------------- | ------------------------------------------------- |
| `ctxs`              | `dict[str, Any]` | required            | Maps `deployment_id` to its `AutoscalingContext`. |
| `kubemq_address`    | `str`            | `"localhost:50000"` | KubeMQ broker address (host:port).                |
| `queue_name`        | `str`            | `""`                | Queue channel name to monitor.                    |
| `tasks_per_replica` | `int`            | `5`                 | Target number of queued tasks per replica.        |
| `auth_token`        | `str`            | `""`                | JWT authentication token.                         |

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:

```python
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](/integrations/rayserve/how-to/autoscaling) for
both the in-process policy and the KEDA path.

```python title="serve_config.py"
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 [#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.

| Metric                             | Type               | Description                                                                                 |
| ---------------------------------- | ------------------ | ------------------------------------------------------------------------------------------- |
| `queue_depth`                      | Gauge (`int`)      | Current waiting messages in the task queue, from `list_queues_channels().incoming.waiting`. |
| `in_flight`                        | Gauge (`int`)      | Tasks currently being processed by handlers.                                                |
| `dlq_depth`                        | Gauge (`int`)      | Waiting messages in the failed-task (DLQ) channel.                                          |
| `tasks_enqueued_total`             | Counter (`int`)    | Total tasks enqueued since startup.                                                         |
| `tasks_completed_total`            | Counter (`dict`)   | Completed counts by status: `{"SUCCESS": int, "FAILURE": int}`.                             |
| `task_processing_duration_seconds` | Histogram (`dict`) | Running stats: `{"min", "max", "avg", "count"}`.                                            |
| `result_storage_retries_total`     | Counter (`int`)    | Total result-storage retry attempts.                                                        |
| `consumer_poll_latency_seconds`    | Gauge (`float`)    | Duration of the last `receive_queue_messages` poll.                                         |

```python
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,
# }
```

<Callout type="info">
  `task_processing_duration_seconds` uses O(1) running statistics — `min`, `max`, `sum`, and `count`
  — rather than storing every duration, so memory stays flat under sustained load. Before any task
  completes, the histogram reports all-zero stats. See the [metrics guide](/integrations/rayserve/how-to/metrics) for
  collection and dashboard patterns.
</Callout>

## Task statuses and TaskResult [#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.

| Status      | Meaning                                                                            |
| ----------- | ---------------------------------------------------------------------------------- |
| `PENDING`   | Enqueued (or no result yet on the channel); awaiting a consumer.                   |
| `STARTED`   | Reserved status string for in-progress reporting.                                  |
| `SUCCESS`   | Handler returned; `result` holds the JSON-serializable return value.               |
| `FAILURE`   | Handler raised after retries were exhausted; an `error` field is stored alongside. |
| `CANCELLED` | Result was overwritten by `cancel_task_sync()` (soft cancel).                      |

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

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

| Field       | Value                                                                                |
| ----------- | ------------------------------------------------------------------------------------ |
| Channel     | `{queue_name}.progress`                                                              |
| Body (JSON) | `{ "task_id": str, "pct": float, "detail": str }`                                    |
| Tags        | `{ "task_id": str, "pct": str }` (the `pct` tag is the integer-truncated percentage) |

```python
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](/integrations/rayserve/how-to/progress-tracking) for the full subscriber-side walkthrough.

## Result-channel naming and expiry [#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:

```text
{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 [#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.

| Feature            | kubemq-rayserve                                   | CeleryTaskProcessorAdapter            |
| ------------------ | ------------------------------------------------- | ------------------------------------- |
| Message Broker     | KubeMQ (Kubernetes-native)                        | Redis / RabbitMQ                      |
| Sync Inference     | Built-in (`query_task_sync`)                      | Not available                         |
| Progress Tracking  | Built-in (`report_progress`)                      | Requires custom signals               |
| Autoscaling Policy | `kubemq_queue_depth_policy`                       | Manual HPA configuration              |
| DLQ Monitoring     | `on_dlq` callback                                 | Requires Celery signals + custom code |
| Result Backend     | Queue-peek (no extra infra)                       | Requires separate Redis/DB            |
| KEDA Integration   | Native ([kubemq-keda-scaler](/integrations/keda)) | Requires custom KEDA scaler           |
| Kubernetes-native  | Yes (single binary, no Erlang)                    | No (Redis/RabbitMQ dependencies)      |
| Health Check       | SDK `ping()`                                      | Celery `inspect().ping()`             |
| Setup Complexity   | 1 dependency (KubeMQ)                             | 2+ dependencies (broker + backend)    |