Handler Base

Shared base library for building NATS-based AI/ML handler services.

Installation

pip install handler-base

Or from Gitea:

pip install git+https://git.daviestechlabs.io/daviestechlabs/handler-base.git

Quick Start

from handler_base import Handler, Settings
from nats.aio.msg import Msg

class MyHandler(Handler):
    async def setup(self):
        # Initialize your clients
        pass
    
    async def handle_message(self, msg: Msg, data: dict):
        # Process the message
        result = {"processed": True}
        return result

if __name__ == "__main__":
    MyHandler(subject="my.subject").run()

Features

• Handler base class - NATS subscription, graceful shutdown, signal handling
• NATSClient - Connection management, JetStream, msgpack serialization
• Settings - Pydantic-based configuration from environment
• HealthServer - Kubernetes liveness/readiness probes
• Telemetry - OpenTelemetry tracing and metrics
• Service clients - HTTP wrappers for AI services

Service Clients

from handler_base.clients import (
    STTClient,      # Whisper speech-to-text
    TTSClient,      # XTTS text-to-speech
    LLMClient,      # vLLM chat completions
    EmbeddingsClient,  # BGE embeddings
    RerankerClient,    # BGE reranker
    MilvusClient,      # Vector database
)

Configuration

All settings via environment variables:

Variable	Default	Description
NATS_URL	nats://localhost:4222	NATS server URL
NATS_USER	-	NATS username
NATS_PASSWORD	-	NATS password
NATS_QUEUE_GROUP	-	Queue group for load balancing
HEALTH_PORT	8080	Health check server port
OTEL_ENABLED	true	Enable OpenTelemetry
OTEL_EXPORTER_OTLP_ENDPOINT	http://localhost:4317	OTLP endpoint
OTEL_SERVICE_NAME	handler	Service name for traces

Docker

FROM ghcr.io/daviestechlabs/handler-base:latest

COPY my_handler.py /app/
CMD ["python", "/app/my_handler.py"]

Or build with audio support:

docker build --build-arg INSTALL_AUDIO=true -t my-handler .

Module Structure

handler_base/
├── __init__.py      # Public API exports
├── handler.py       # Base Handler class
├── nats_client.py   # NATS connection wrapper
├── config.py        # Pydantic Settings
├── health.py        # Health check server
├── telemetry.py     # OpenTelemetry setup
└── clients/
    ├── embeddings.py
    ├── llm.py
    ├── milvus.py
    ├── reranker.py
    ├── stt.py
    └── tts.py

Related

• voice-assistant - Voice pipeline using handler-base
• homelab-design - Architecture docs