Getting Started with Ray Serve

Prerequisites

You need the following installed locally:

The kubemq-rayserve package pulls in these runtime dependencies automatically:

Install the Adapter

Install the package from PyPI:

uv pip install kubemq-rayserve

If you want to work against the source — for example to run the bundled examples or the test suite — clone the repository and sync the development dependency group instead:

git clone https://github.com/kubemq-io/kubemq-rayserve.git
cd kubemq-rayserve
uv sync --group dev

Start a KubeMQ Broker

Run KubeMQ in 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 the gRPC port the adapter connects to — it is the default broker address (localhost:50000) used throughout this guide. Port 9090 exposes the broker's shared HTTP server (REST and the AI-agent connectors); the dashboard, where you can inspect queues, channels, and message flow while your tasks run, runs separately on port 8080.

Write a Task Handler and Set Up the Adapter

A task handler is a plain Python function. Register it with the adapter under a name, and the adapter dispatches matching enqueued tasks to it.

In production, Ray Serve calls initialize() for you with a TaskProcessorConfig (shown in the last step). For standalone usage outside a deployment, you supply a small config object with the same fields the adapter reads — queue_name, max_retries, failed_task_queue_name, and unprocessable_task_queue_name:

import time
from kubemq_rayserve import KubeMQAdapterConfig, KubeMQTaskProcessorAdapter

# 1. Define a task handler
def classify(text: str) -> dict:
    return {"label": "POSITIVE", "score": 0.95}

# 2. Configure and initialize the adapter
config = KubeMQAdapterConfig(address="localhost:50000")
adapter = KubeMQTaskProcessorAdapter(config)

# In production, Ray Serve calls initialize() with TaskProcessorConfig.
# For standalone usage:
class TaskConfig:
    queue_name = "inference-tasks"
    max_retries = 3
    failed_task_queue_name = "inference-tasks-dlq"
    unprocessable_task_queue_name = ""

adapter.initialize(consumer_concurrency=2, task_processor_config=TaskConfig())
adapter.register_task_handler(classify, name="classify")
adapter.start_consumer()

KubeMQAdapterConfig(address="localhost:50000") points the adapter at the broker you just started. initialize() creates the underlying SDK clients and stores the concurrency setting (consumer_concurrency=2 runs two parallel consumers), and start_consumer() launches the queue-polling thread that pulls tasks off inference-tasks and dispatches them to your registered classify handler.

Enqueue a Task and Poll for the Result

With the consumer running, enqueue a task by name. enqueue_task_sync() serializes the task, sends it to the queue, and immediately returns a TaskResult with a generated id and a PENDING status — it does not wait for processing. You then poll get_task_status_sync(task_id) until the status becomes SUCCESS:

# 3. Enqueue a task
result = adapter.enqueue_task_sync(task_name="classify", args=["great product"])
print(f"Task ID: {result.id}, Status: {result.status}")

# 4. Poll for result
for _ in range(30):
    status = adapter.get_task_status_sync(result.id)
    if status.status == "SUCCESS":
        print(f"Result: {status.result}")
        break
    time.sleep(0.5)

# 5. Cleanup
adapter.stop_consumer()

Each get_task_status_sync() call peeks the result channel and returns the current TaskResult. Once the handler finishes, the status flips to SUCCESS and status.result holds the handler's return value — here, {"label": "POSITIVE", "score": 0.95}. Always call stop_consumer() when you are done: it cancels the polling token, stops the background threads, and closes the SDK clients.

Run a Packaged Example Instead

The repository ships a self-contained quickstart you can run directly against your broker. It performs the same async enqueue-and-poll flow shown above, using a unique channel name per run:

python examples/quickstart/hello_world.py

Every example defaults to a broker at localhost:50000. Override the address with the KUBEMQ_ADDRESS environment variable:

KUBEMQ_ADDRESS=my-broker:50000 python examples/quickstart/hello_world.py

Deploy on Ray Serve (Production)

For real deployments you do not call initialize() yourself. Instead, you decorate a Ray Serve deployment with @task_consumer, passing a TaskProcessorConfig that names the adapter class and its KubeMQAdapterConfig. Ray Serve wires up the consumer, and each method invocation runs as a task pulled from the queue:

from ray import serve
from kubemq_rayserve import KubeMQAdapterConfig, KubeMQTaskProcessorAdapter

@serve.deployment(ray_actor_options={"num_gpus": 1})
@task_consumer(
    task_processor_config=TaskProcessorConfig(
        adapter_class=KubeMQTaskProcessorAdapter,
        adapter_config=KubeMQAdapterConfig(
            address="kubemq:50000",
            auth_token="your-jwt-token",
        ),
        queue_name="inference-tasks",
        max_retries=3,
        failed_task_queue_name="inference-tasks-dlq",
    )
)
class ModelDeployment:
    def __init__(self):
        self.model = load_model()

    def __call__(self, data: str) -> dict:
        return self.model.predict(data)

The adapter_config here points at an in-cluster broker (kubemq:50000) and supplies a JWT via auth_token. For the complete Kubernetes deployment — broker, Ray cluster, RayService, and queue-depth autoscaling — see the Kubernetes production deployment scenario.

Next Steps