# Getting Started with Ray Serve (/integrations/rayserve/tutorials/getting-started) The KubeMQ Ray Serve adapter provides `KubeMQTaskProcessorAdapter` — an implementation of Ray Serve's `TaskProcessorAdapter` framework that backs queue-based async and sync ML inference with KubeMQ. This guide takes you from an empty environment to a working task handler that enqueues a job and polls for its result. ## Implementation steps [#implementation-steps] ### Prerequisites [#prerequisites] You need the following installed locally: | Requirement | Version | | ----------- | -------------------------------------- | | Python | 3.10 or newer | | Ray Serve | 2.50.0 or newer | | Docker | Any recent version (to run the broker) | The `kubemq-rayserve` package pulls in these runtime dependencies automatically: | Dependency | Version constraint | | ------------ | ------------------ | | `kubemq` | `>=4.1.5` | | `ray[serve]` | `>=2.50.0` | | `pydantic` | `>=2.0` | ### Install the Adapter [#install-the-adapter] Install the package from PyPI: ```bash 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: ```bash git clone https://github.com/kubemq-io/kubemq-rayserve.git cd kubemq-rayserve uv sync --group dev ``` ### Start a KubeMQ Broker [#start-a-kubemq-broker] Run KubeMQ in Docker: 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 [#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`: ```python title="standalone.py" 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. `max_retries` and `failed_task_queue_name` control retry and dead-letter behavior: a task that keeps failing is retried up to `max_retries` times before being routed to the DLQ. See the [configuration reference](/integrations/rayserve/reference/configuration) for every `KubeMQAdapterConfig` field. ### Enqueue a Task and Poll for the Result [#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`: ```python title="standalone.py" # 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 [#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: ```bash python examples/quickstart/hello_world.py ``` Every example defaults to a broker at `localhost:50000`. Override the address with the `KUBEMQ_ADDRESS` environment variable: ```bash KUBEMQ_ADDRESS=my-broker:50000 python examples/quickstart/hello_world.py ``` ### Deploy on Ray Serve (Production) [#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: ```python title="deployment.py" 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](/integrations/rayserve/scenarios/kubernetes-production-deployment) scenario. ### Next Steps [#next-steps]