vs-queue

Pluggable async message queue library for Viveka Sutra — Redis Streams, RabbitMQ, retry policies, dead letter queues, and annotation-based consumers.

---

Overview

vs-queue is the messaging backbone for all Viveka Sutra services. It provides a unified API for publishing and consuming messages across multiple queue backends. Services publish a VsMessage to a named queue and consumers process it — the queue backend is swappable via config without changing application code.

Key features:

• Unified VsBaseQueue abstraction — swap Redis for RabbitMQ without changing consumer code
• @queue_listener decorator — register consumers declaratively, like Spring's @RabbitListener
• VsQueueManager — lifecycle management for all consumers: start, stop, health check
• Built-in retry with exponential backoff
• Dead letter queue (DLQ) support
• Registry pattern — register custom queue providers by name
• Fully async — built on asyncio, compatible with FastAPI and any async framework

---

The Problem It Solves

Services that call each other directly over HTTP are tightly coupled — a slow or unavailable downstream service blocks the caller.

Without vs-queue

# Caller blocks waiting for the downstream service
response = await http_client.post("/v1/process", json=payload)
response.raise_for_status()
return response.json()

If the downstream service is slow or down, every caller fails. There is no retry, no buffering, and no way to recover without restarting.

With vs-queue

# Caller publishes and returns immediately
await queue.publish("orchestrator:tasks", VsMessage(payload={"node_id": node_id}))

The consumer processes at its own pace. If processing fails it retries automatically. If it exceeds max retries it goes to the dead letter queue for inspection. The caller never blocks.

---

Installation

pip install vs-queue

With Redis Streams support:

pip install vs-queue[redis]

With RabbitMQ support:

pip install vs-queue[rabbitmq]

Both backends:

pip install vs-queue[redis,rabbitmq]

---

Dependencies

Package	Version	Required	Purpose
pydantic	>=2.0	Yes	VsMessage schema validation
redis	>=5.0	No — install with [redis] extra	Redis Streams backend
hiredis	>=2.0	No — install with [redis] extra	Redis protocol parser (performance)
aio-pika	>=9.0	No — install with [rabbitmq] extra	RabbitMQ backend via AMQP

---

Quick Start

Publisher

import asyncio
from vs_queue import VsQueueRegistry  # auto-registers redis and rabbitmq
from vs_queue.factory.vs_queue_factory import VsQueueFactory
from vs_queue.schema.vs_message import VsMessage

async def main():
    queue = VsQueueFactory.create(
        provider="redis",
        host="localhost",
        port=6379,
    )
    await queue.connect()

    message = VsMessage(
        headers={"type": "orchestrator.task", "trace_id": "abc123"},
        payload={"node_id": "xyz", "user_message": "hello"},
    )
    await queue.publish("orchestrator:tasks", message)
    await queue.disconnect()

asyncio.run(main())

Consumer (manual)

from vs_queue.base.vs_base_consumer import VsBaseConsumer
from vs_queue.schema.vs_message import VsMessage

class OrchestratorConsumer(VsBaseConsumer):

    async def handle(self, message: VsMessage) -> None:
        node_id = message.payload["node_id"]
        print(f"Processing node: {node_id}")

    async def start(self) -> None:
        print("OrchestratorConsumer started")

    async def stop(self) -> None:
        print("OrchestratorConsumer stopped")

Consumer (annotation-based)

from vs_queue.decorator.vs_queue_listener import queue_listener
from vs_queue.base.vs_base_consumer import VsBaseConsumer
from vs_queue.schema.vs_message import VsMessage

@queue_listener(
    queue="orchestrator:tasks",
    concurrency=2,
    max_retries=3,
    retry_backoff_seconds=2.0,
    dead_letter_queue="orchestrator:dlq",
)
class OrchestratorConsumer(VsBaseConsumer):

    async def handle(self, message: VsMessage) -> None:
        node_id = message.payload["node_id"]
        # process the message
        print(f"Processing node: {node_id}")

    async def start(self) -> None:
        pass

    async def stop(self) -> None:
        pass

Starting all listeners

from vs_queue.factory.vs_queue_factory import VsQueueFactory
from vs_queue.manager.vs_queue_manager import VsQueueManager

import my_app.consumers  # noqa — import modules that contain @queue_listener classes

queue = VsQueueFactory.create(provider="redis", host="localhost", port=6379)
manager = VsQueueManager(queue)
await manager.register_listeners()  # discovers all @queue_listener classes, connects & starts immediately

# ... app runs ...
await manager.stop_all()

---

Configuration

Redis

[queue]
provider = redis
host     = localhost
port     = 6379
password = your_redis_password

RabbitMQ

[queue]
provider  = rabbitmq
host      = localhost
port      = 5672
username  = guest
password  = guest
vhost     = /

---

VsMessage

Every message published and consumed by vs-queue is a VsMessage. It is backend-agnostic — the same schema works for Redis Streams and RabbitMQ.

from vs_queue.schema.vs_message import VsMessage

message = VsMessage(
    headers={"type": "orchestrator.task", "trace_id": "abc123", "source_queue": "orchestrator:tasks"},
    payload={"node_id": "xyz", "user_id": "u1"},
)

Fields:

Field	Type	Default	Description
id	str	uuid4().hex	Unique message ID, auto-generated
timestamp	datetime	now(UTC)	Message creation time, auto-set
retry_count	int	0	Number of processing attempts. Managed by the library — do not set manually.
headers	Dict[str, str]	{}	Caller-defined metadata: type, trace_id, routing info, etc.
payload	Dict[str, Any]	None	Message body. Structure is defined by the caller.

Notes:

• id and timestamp are auto-generated — you rarely need to set them.
• retry_count is managed by the retry policy. Do not set it manually.
• headers is a flat Dict[str, str]. Nest complex metadata in payload.
• Include "source_queue" in headers if you use retry — the retry policy uses it to re-publish.

message = VsMessage(
    headers={
        "type": "task.process",
        "trace_id": "abc123",
        "source_queue": "orchestrator:tasks",
    },
    payload={
        "node_id": "n1",
        "conversation_id": "c1",
        "user_message": "what is the weather today?",
    },
)

print(message.id)           # e.g. "3f2a1b4c..."
print(message.timestamp)    # e.g. 2026-08-03T10:00:00+00:00
print(message.retry_count)  # 0

---

VsBaseQueue

Abstract base class for all queue backends. Extend this to implement a custom provider.

Constructor parameters:

Parameter	Type	Required	Description
host	str	Yes	Broker host
port	int	Yes	Broker port
credentials	Dict[str, Any]	No	Auth credentials. Keys depend on the backend.

Abstract methods:

Method	Description
connect()	Establish connection to the broker
disconnect()	Close connection cleanly
health_check()	Returns True if the broker is reachable
publish(queue, message)	Publish a VsMessage to a named queue
subscribe(queue, consumer, retry_policy)	Begin consuming from a queue using a VsBaseConsumer

---

VsBaseConsumer

Abstract base class for all consumers. Extend this and implement handle to process messages.

Abstract methods:

Method	Description
handle(message)	Called for every message received. Raise any exception to trigger retry.
start()	Called once when the consumer begins. Use for setup.
stop()	Called once on graceful shutdown. Use for cleanup.

Optional override:

Method	Description
on_error(message, error)	Called when handle raises an exception, before retry logic runs. Use for logging or alerting.

Example with error hook:

from vs_queue.base.vs_base_consumer import VsBaseConsumer
from vs_queue.schema.vs_message import VsMessage
import logging

_logger = logging.getLogger(__name__)

class TaskConsumer(VsBaseConsumer):

    async def handle(self, message: VsMessage) -> None:
        result = await process(message.payload)
        if not result:
            raise ValueError(f"Processing failed for node {message.payload['node_id']}")

    async def start(self) -> None:
        _logger.info("TaskConsumer starting")

    async def stop(self) -> None:
        _logger.info("TaskConsumer stopping")

    async def on_error(self, message: VsMessage, error: Exception) -> None:
        _logger.error(f"Message failed | id={message.id} error={error}")
        # send alert, update DB status, etc.

---

@queue_listener

Decorator that registers a VsBaseConsumer subclass as a listener. VsQueueManager.register_listeners() discovers all decorated classes and starts them automatically.

Parameters:

Parameter	Type	Required	Default	Description
queue	str	Yes	—	Queue name to consume from
concurrency	int	No	1	Number of parallel consumer instances to start
max_retries	int	No	3	Max retry attempts before sending to DLQ
retry_backoff_seconds	float	No	1.0	Base delay for exponential backoff between retries
dead_letter_queue	str	No	None	Queue name for messages that exceed max retries

Example — minimal:

@queue_listener(queue="tasks")
class SimpleConsumer(VsBaseConsumer):
    async def handle(self, message: VsMessage) -> None:
        print(message.payload)

    async def start(self) -> None: pass
    async def stop(self) -> None: pass

Example — full configuration:

@queue_listener(
    queue="orchestrator:tasks",
    concurrency=3,
    max_retries=5,
    retry_backoff_seconds=2.0,
    dead_letter_queue="orchestrator:dlq",
)
class OrchestratorConsumer(VsBaseConsumer):

    async def handle(self, message: VsMessage) -> None:
        trace_id = message.headers.get("trace_id")
        node_id = message.payload["node_id"]
        await run_graph(trace_id, node_id)

    async def start(self) -> None:
        await db.connect()

    async def stop(self) -> None:
        await db.disconnect()

    async def on_error(self, message: VsMessage, error: Exception) -> None:
        await notify_ops(f"Failed: {message.id} — {error}")

Important: The module containing @queue_listener classes must be imported before calling register_listeners(), otherwise the decorator never runs and the class is never registered. register_listeners() is async and safe to call only once.

import my_app.consumers  # noqa — triggers @queue_listener registration
await manager.register_listeners()

---

VsQueueManager

Manages the lifecycle of all consumers. Connects to the broker, starts consumer tasks, handles graceful shutdown.

Constructor:

Parameter	Type	Required	Default	Description
queue	VsBaseQueue	Yes	—	The queue instance
max_consumer_restarts	int	No	3	Max times a crashed consumer task is restarted before giving up

Methods:

Method	Description
register(queue_name, consumer)	Register a consumer, auto-connect if needed, start immediately. Returns a UUID.
register_listeners()	Auto-discover all @queue_listener classes and register them. Safe to call only once.
get_listeners()	Returns Dict[str, List[VsBaseConsumer]] — queue name → running consumer instances
stop_listener(queue_name, uuid=None)	Stop all consumers on a queue, or one specific consumer by UUID
stop_all()	Stop all consumers and disconnect from broker
health_check()	Returns True if connected and broker is healthy. Returns False if not yet connected.

Example — with vs-server lifecycle:

from vs_queue.factory.vs_queue_factory import VsQueueFactory
from vs_queue.manager.vs_queue_manager import VsQueueManager
from vs_server.lifecycle.vs_lifecycle import startup, shutdown
import app.consumers  # noqa

queue = VsQueueFactory.create(provider="redis", host="localhost", port=6379)
manager = VsQueueManager(queue, max_consumer_restarts=5)

@startup
async def start_consumers():
    await manager.register_listeners()

@shutdown
async def stop_consumers():
    await manager.stop_all()

Example — dynamic registration:

manager = VsQueueManager(queue)

# auto-connects on first register(), starts task immediately
for tenant_id in ["tenant_a", "tenant_b"]:
    uuid = await manager.register(f"tasks:{tenant_id}", TenantConsumer())

# stop one specific consumer
await manager.stop_listener("tasks:tenant_a", uuid)

# stop all consumers on a queue
await manager.stop_listener("tasks:tenant_b")

# inspect running consumers
listeners = manager.get_listeners()
# {"tasks:tenant_a": [<TenantConsumer uuid=...>]}

---

Retry Policy

VsRetryPolicy defines what happens when handle() raises an exception. Pass it to subscribe() on the queue directly, or let @queue_listener configure it declaratively.

Constructor:

Parameter	Type	Default	Description
max_retries	int	3	Max attempts before giving up
backoff_seconds	float	1.0	Base delay. Doubles on each retry (exponential backoff).
dead_letter_queue	str	None	Queue for messages that exceed max retries

Retry timing:

Attempt	Delay
1st retry	1s
2nd retry	2s
3rd retry	4s
Beyond max	→ DLQ

Example — manual:

from vs_queue.retry.vs_retry_policy import VsRetryPolicy

retry_policy = VsRetryPolicy(
    max_retries=5,
    backoff_seconds=2.0,
    dead_letter_queue="orchestrator:dlq",
)

await queue.subscribe("orchestrator:tasks", consumer, retry_policy=retry_policy)

Important: Include "source_queue" in the message headers so the retry policy knows where to re-publish:

message = VsMessage(
    headers={
        "source_queue": "orchestrator:tasks",
        "trace_id": "abc123",
    },
    payload={"node_id": "n1"},
)

---

Dead Letter Queue

Messages that fail after all retry attempts are published to the dead letter queue (DLQ). The DLQ is just another queue — you can attach a consumer to it for inspection, alerting, or manual replay.

Example — DLQ consumer:

@queue_listener(queue="orchestrator:dlq")
class DlqConsumer(VsBaseConsumer):

    async def handle(self, message: VsMessage) -> None:
        await alert_ops(
            f"Message permanently failed | id={message.id} retries={message.retry_count}",
            payload=message.payload,
        )

    async def start(self) -> None: pass
    async def stop(self) -> None: pass

Example — replaying a DLQ message:

dlq_message = ...  # fetch from DLQ consumer

# Reset retry count and re-publish to the original queue
dlq_message.retry_count = 0
await queue.publish("orchestrator:tasks", dlq_message)

---

Redis Streams Backend

VsRedisQueue implements the queue using Redis Streams (XADD / XREADGROUP). Redis Streams provide persistent, ordered, consumer-group-aware message delivery.

How it works:

• publish → XADD queue * field value ...
• subscribe → XREADGROUP GROUP group consumer STREAMS queue >
• On success → XACK removes the message from the pending entries list
• On failure → retry policy runs; if max retries exceeded → DLQ

Credentials:

Key	Description
password	Redis AUTH password

Example:

from vs_queue.provider.vs_redis_queue import VsRedisQueue

queue = VsRedisQueue(
    host="localhost",
    port=6379,
    credentials={"password": "secret"},
)
await queue.connect()

healthy = await queue.health_check()  # True if Redis responds to PING

Consumer group naming:

• Group name: {queue}:group
• Consumer name: {queue}:consumer

These are created automatically on first subscribe(). If the stream does not exist yet, it is created via mkstream=True.

---

RabbitMQ Backend

VsRabbitMQQueue implements the queue using AMQP via aio-pika. Messages are published to the default exchange with the queue name as the routing key. Queues are declared as durable — messages survive broker restarts.

Credentials:

Key	Default	Description
username	guest	RabbitMQ username
password	guest	RabbitMQ password
vhost	/	RabbitMQ virtual host

Example:

from vs_queue.provider.vs_rabbitmq_queue import VsRabbitMQQueue

queue = VsRabbitMQQueue(
    host="localhost",
    port=5672,
    credentials={"username": "admin", "password": "secret", "vhost": "/prod"},
)
await queue.connect()

Acknowledgement behaviour:

• On success → ack() removes the message
• On failure with retry → ack() the original, re-publish with incremented retry_count
• On failure beyond max retries → nack(requeue=False) + publish to DLQ

---

VsQueueRegistry

Stores registered queue provider classes by name. Pre-registered providers: redis, rabbitmq.

Methods:

Method	Description
register(name, queue_class)	Register a VsBaseQueue subclass under a name
get(name)	Return the class for a given name. Raises KeyError if not found.
available()	Return list of all registered provider names

Example — registering a custom provider:

from vs_queue.registry.vs_queue_registry import VsQueueRegistry
from vs_queue.base.vs_base_queue import VsBaseQueue

class VsKafkaQueue(VsBaseQueue):
    async def connect(self): ...
    async def disconnect(self): ...
    async def health_check(self): ...
    async def publish(self, queue, message): ...
    async def subscribe(self, queue, consumer, retry_policy=None): ...

VsQueueRegistry.register("kafka", VsKafkaQueue)

Once registered, the factory can create it:

queue = VsQueueFactory.create(provider="kafka", host="localhost", port=9092)

---

VsQueueFactory

Creates queue instances from a registered provider name.

Methods:

Method	Signature	Description
create	create(provider, host, port, credentials) -> VsBaseQueue	Create and return a queue instance. Does not call connect().

Example:

from vs_queue.factory.vs_queue_factory import VsQueueFactory

queue = VsQueueFactory.create(
    provider="redis",
    host="localhost",
    port=6379,
    credentials={"password": "secret"},
)
await queue.connect()

---

Extending vs-queue

Custom Queue Provider

Implement VsBaseQueue and register it:

from vs_queue.base.vs_base_queue import VsBaseQueue
from vs_queue.base.vs_base_consumer import VsBaseConsumer
from vs_queue.retry.vs_retry_policy import VsRetryPolicy
from vs_queue.schema.vs_message import VsMessage
from vs_queue.registry.vs_queue_registry import VsQueueRegistry
from typing import Optional

class VsSqsQueue(VsBaseQueue):

    def __init__(self, host: str, port: int, credentials=None):
        super().__init__(host, port, credentials)
        self._client = None

    async def connect(self) -> None:
        import aiobotocore.session
        session = aiobotocore.session.get_session()
        self._client = await session.create_client(
            "sqs",
            region_name=self._credentials.get("region", "us-east-1"),
            aws_access_key_id=self._credentials.get("access_key"),
            aws_secret_access_key=self._credentials.get("secret_key"),
        ).__aenter__()

    async def disconnect(self) -> None:
        if self._client:
            await self._client.__aexit__(None, None, None)

    async def health_check(self) -> bool:
        try:
            await self._client.list_queues(MaxResults=1)
            return True
        except Exception:
            return False

    async def publish(self, queue: str, message: VsMessage) -> None:
        import json
        await self._client.send_message(
            QueueUrl=queue,
            MessageBody=json.dumps({"id": message.id, "payload": message.payload, "headers": message.headers}),
        )

    async def subscribe(self, queue: str, consumer: VsBaseConsumer, retry_policy: Optional[VsRetryPolicy] = None) -> None:
        await consumer.start()
        try:
            while True:
                response = await self._client.receive_message(QueueUrl=queue, WaitTimeSeconds=20)
                for msg in response.get("Messages", []):
                    # parse and call consumer.handle(message)
                    ...
        finally:
            await consumer.stop()

VsQueueRegistry.register("sqs", VsSqsQueue)

---

Full Example — Orchestrator Integration

Publisher (chat controller):

from vs_queue.factory.vs_queue_factory import VsQueueFactory
from vs_queue.schema.vs_message import VsMessage

queue = VsQueueFactory.create(provider="redis", host="localhost", port=6379)
await queue.connect()

await queue.publish(
    "orchestrator:tasks",
    VsMessage(
        headers={
            "type": "orchestrator.task",
            "trace_id": trace_id,
            "source_queue": "orchestrator:tasks",
        },
        payload={
            "node_id": node["node"]["id"],
            "conversation_id": node["conversation_id"],
            "user_message": request.message,
            "user_id": user_id,
        },
    ),
)

Consumer (orchestrator worker):

from vs_queue.decorator.vs_queue_listener import queue_listener
from vs_queue.base.vs_base_consumer import VsBaseConsumer
from vs_queue.schema.vs_message import VsMessage
from orchestrator_agent.graph.orchestrator_graph import OrchestratorGraph
from orchestrator_agent.graph.orchestrator_state import OrchestratorState

@queue_listener(
    queue="orchestrator:tasks",
    concurrency=2,
    max_retries=3,
    retry_backoff_seconds=2.0,
    dead_letter_queue="orchestrator:dlq",
)
class OrchestratorTaskConsumer(VsBaseConsumer):

    def __init__(self):
        super().__init__()
        self._graph = OrchestratorGraph()

    async def handle(self, message: VsMessage) -> None:
        payload = message.payload
        result = await self._graph.invoke(OrchestratorState(
            trace_id=message.headers.get("trace_id", ""),
            error=None,
            current_message=payload["user_message"],
            user_id=payload["user_id"],
            message=payload["user_message"],
            resolved_context=None,
            response=None,
        ))

    async def start(self) -> None:
        pass

    async def stop(self) -> None:
        pass

    async def on_error(self, message: VsMessage, error: Exception) -> None:
        import logging
        logging.getLogger(__name__).error(
            f"Graph execution failed | node_id={message.payload.get('node_id')} error={error}"
        )

App startup:

from vs_queue.factory.vs_queue_factory import VsQueueFactory
from vs_queue.manager.vs_queue_manager import VsQueueManager
from vs_server.lifecycle.vs_lifecycle import startup, shutdown
import orchestrator_agent.consumers  # noqa — registers @queue_listener classes

queue = VsQueueFactory.create(provider="redis", host="localhost", port=6379)
manager = VsQueueManager(queue, max_consumer_restarts=5)

@startup
async def start_consumers():
    await manager.register_listeners()

@shutdown
async def stop_consumers():
    await manager.stop_all()

---

Class Reference

---

VsMessage

Standard message envelope. Every message published and consumed by vs-queue is a VsMessage.

Field	Type	Default	Description
id	str	uuid4().hex	Auto-generated unique ID
timestamp	datetime	now(UTC)	Auto-set creation time
retry_count	int	0	Managed by retry policy
headers	Dict[str, str]	{}	Caller metadata
payload	Dict[str, Any]	None	Message body

---

VsBaseQueue

Abstract base. Extend to implement a custom queue provider.

Method	Description
connect()	Establish broker connection
disconnect()	Close connection cleanly
health_check()	Returns True if broker is reachable
publish(queue, message)	Publish a message
subscribe(queue, consumer, retry_policy)	Start consuming

---

VsBaseConsumer

Abstract base. Extend and implement handle to process messages.

Attribute:

Attribute	Type	Description
uuid	Optional[str]	Set automatically at registration time. Use it with stop_listener() to stop a specific instance.

Methods:

Method	Required	Description
handle(message)	Yes	Process a message. Raise to trigger retry.
start()	Yes	Setup before consuming begins
stop()	Yes	Cleanup on shutdown — called on graceful cancel
on_error(message, error)	No	Called on handle failure before retry

---

VsQueueManager

Lifecycle manager for all consumers.

Method	Description
register(queue_name, consumer)	Register a consumer, auto-connect, start immediately. Returns UUID.
register_listeners()	Auto-discover @queue_listener classes. Safe to call only once.
get_listeners()	Returns Dict[str, List[VsBaseConsumer]]
stop_listener(queue_name, uuid=None)	Stop one or all consumers on a queue
stop_all()	Stop all consumers and disconnect
health_check()	Returns True if connected and broker is healthy

---

VsRetryPolicy

Defines retry and DLQ behaviour on consumer failure.

Parameter	Default	Description
max_retries	3	Max attempts
backoff_seconds	1.0	Base exponential backoff delay
dead_letter_queue	None	DLQ name

---

VsQueueRegistry

Registry of queue provider classes.

Method	Description
register(name, queue_class)	Register a provider
get(name)	Get a provider class by name
available()	List all registered provider names

---

VsQueueFactory

Creates queue instances from the registry.

Method	Description
create(provider, host, port, credentials)	Instantiate a queue. Call connect() before use.

---

@queue_listener

Decorator. Registers a VsBaseConsumer subclass for auto-discovery by VsQueueManager.

Parameter	Default	Description
queue	—	Queue name
concurrency	1	Parallel consumer instances
max_retries	3	Max retries on failure
retry_backoff_seconds	1.0	Exponential backoff base delay
dead_letter_queue	None	DLQ name

---

Running Tests

./run_tests.sh