feat: Add Redis-backed QueueManager for production deployments

- Add RedisEventQueue for Redis Streams-based event queuing
- Add RedisQueueManager for distributed queue management
- Add RedisEventConsumer for consuming Redis stream events
- Add RedisRequestHandler for Redis-backed request handling
- Add comprehensive test coverage for all Redis components
- Update DefaultRequestHandler with backward compatibility
- Add environment variable controls for strict deployment modes

This implementation enables production deployments in distributed environments like Kubernetes, addressing the limitation of only having InMemoryQueueManager which cannot be used in multi-pod setups.

Redis is widely used in agentic AI platforms like LangGraph and provides reliable, scalable event streaming for serverless and distributed architectures.

Author: mjunaidca

pyproject.toml

@@ -37,6 +37,7 @@ sql = ["sqlalchemy[asyncio,postgresql-asyncpg,aiomysql,aiosqlite]>=2.0.0"]
 encryption = ["cryptography>=43.0.0"]
 grpc = ["grpcio>=1.60", "grpcio-tools>=1.60", "grpcio_reflection>=1.7.0"]
 telemetry = ["opentelemetry-api>=1.33.0", "opentelemetry-sdk>=1.33.0"]
+redis = ["redis>=6.4.0"]
 
 [project.urls]
 homepage = "https://a2a-protocol.org/"

src/a2a/server/events/redis_event_consumer.py

@@ -0,0 +1,54 @@
+from __future__ import annotations
+
+import asyncio
+import logging
+
+from typing import Protocol, TYPE_CHECKING
+if TYPE_CHECKING:
+    from collections.abc import AsyncGenerator
+
+from a2a.utils.telemetry import SpanKind, trace_class
+
+
+class QueueLike(Protocol):
+    """Protocol describing a minimal queue-like object used by consumers.
+
+    It must provide an async `dequeue_event(no_wait: bool)` method and an
+    `is_closed()` method.
+    """
+
+    async def dequeue_event(self, no_wait: bool = False) -> object:
+        """Return the next queued event or raise asyncio.QueueEmpty if none when no_wait is True."""
+
+    def is_closed(self) -> bool:
+        """Return True if the underlying queue has been closed."""
+        ...
+
+logger = logging.getLogger(__name__)
+
+
+@trace_class(kind=SpanKind.SERVER)
+class RedisEventConsumer:
+    """Adapter that provides the same consume semantics for a Redis-backed EventQueue.
+
+    It wraps a RedisEventQueue instance and exposes methods compatible with
+    existing code expecting an EventQueue (not strictly required but helpful).
+    """
+
+    def __init__(self, queue: QueueLike) -> None:
+        """Wrap a queue-like object that exposes dequeue_event and is_closed."""
+        self._queue = queue
+    async def consume_one(self) -> object:
+        """Consume a single event without waiting; raises asyncio.QueueEmpty if none."""
+        return await self._queue.dequeue_event(no_wait=True)
+
+    async def consume_all(self) -> AsyncGenerator:
+        """Yield events until the queue is closed."""
+        while True:
+            try:
+                event = await self._queue.dequeue_event()
+                yield event
+                if self._queue.is_closed():
+                    break
+            except asyncio.QueueEmpty:
+                continue

src/a2a/server/events/redis_event_queue.py

@@ -0,0 +1,221 @@
+"""Redis-backed EventQueue implementation using Redis Streams."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+from typing import Any
+
+try:
+    import redis.asyncio as aioredis  # type: ignore
+    from redis.exceptions import RedisError  # type: ignore
+except ImportError:  # pragma: no cover - optional dependency
+    aioredis = None  # type: ignore
+    RedisError = Exception  # type: ignore
+
+from a2a.server.events.event_queue import EventQueue
+from typing import TYPE_CHECKING
+if TYPE_CHECKING:
+    from a2a.server.events.event_queue import Event
+from pydantic import ValidationError
+from a2a.types import (
+    Message,
+    Task,
+    TaskStatusUpdateEvent,
+    TaskArtifactUpdateEvent,
+)
+from a2a.utils.telemetry import SpanKind, trace_class
+
+logger = logging.getLogger(__name__)
+
+
+class RedisNotAvailableError(RuntimeError):
+    """Raised when the redis.asyncio package is not installed."""
+
+
+_TYPE_MAP = {
+    'Message': Message,
+    'Task': Task,
+    'TaskStatusUpdateEvent': TaskStatusUpdateEvent,
+    'TaskArtifactUpdateEvent': TaskArtifactUpdateEvent,
+}
+
+
+@trace_class(kind=SpanKind.SERVER)
+class RedisEventQueue(EventQueue):
+    """Redis-native EventQueue backed by a Redis Stream.
+
+    This implementation does not rely on in-memory queue structures. Each
+    instance manages its own read cursor (last_id). `tap()` returns a new
+    RedisEventQueue pointing to the same stream but starting at '$' so it
+    receives only future events.
+    """
+
+    def __init__(
+        self,
+        task_id: str,
+        redis_client: Any,
+        stream_prefix: str = 'a2a:task',
+        maxlen: int | None = None,
+        read_block_ms: int = 500,
+    ) -> None:
+        # Allow passing a custom redis client (e.g. a fake in tests).
+        if aioredis is None and redis_client is None:
+            raise RedisNotAvailableError('redis.asyncio is not available')
+
+        self._task_id = task_id
+        self._redis = redis_client
+        self._stream_key = f'{stream_prefix}:{task_id}'
+        self._maxlen = maxlen
+        self._read_block_ms = read_block_ms
+
+        # By default a normal queue should start at the beginning so it can
+        # consume existing entries. Taps will explicitly start at '$'.
+        self._last_id = '0-0'
+        self._is_closed = False
+
+        # No in-memory queue initialization — this class is Redis-native.
+
+    async def enqueue_event(self, event: Event) -> None:
+        """Serialize and append an event to the Redis stream."""
+        if self._is_closed:
+            logger.warning('Attempt to enqueue to closed RedisEventQueue')
+            return
+        # Store payload as a JSON string to avoid client-specific mapping
+        # behaviour when reading back from the stream.
+        payload = {
+            'type': type(event).__name__,
+            'payload': event.json(),
+        }
+        kwargs: dict[str, Any] = {}
+        if self._maxlen:
+            kwargs['maxlen'] = self._maxlen
+        try:
+            await self._redis.xadd(self._stream_key, payload, **kwargs)
+        except RedisError:
+            logger.exception('Failed to XADD event to redis stream')
+
+    async def dequeue_event(self, no_wait: bool = False) -> Event | Any:
+        """Read one event from the Redis stream respecting no_wait semantics.
+
+        Returns a parsed pydantic model matching the event type.
+        """
+        if self._is_closed:
+            raise asyncio.QueueEmpty('Queue is closed')
+
+        block = 0 if no_wait else self._read_block_ms
+        # Keep reading until we find a parseable payload or a CLOSE tombstone.
+        while True:
+            try:
+                result = await self._redis.xread(
+                    {self._stream_key: self._last_id}, block=block, count=1
+                )
+            except RedisError:
+                logger.exception('Failed to XREAD from redis stream')
+                raise
+
+            if not result:
+                raise asyncio.QueueEmpty
+
+            _, entries = result[0]
+            entry_id, fields = entries[0]
+            self._last_id = entry_id
+
+            # Normalize keys/values: redis may return bytes for both keys and values
+            norm: dict[str, object] = {}
+            try:
+                for k, v in fields.items():
+                    key = k.decode('utf-8') if isinstance(k, (bytes, bytearray)) else k
+                    if isinstance(v, (bytes, bytearray)):
+                        try:
+                            val: object = v.decode('utf-8')
+                        except Exception:
+                            val = v
+                    else:
+                        val = v
+                    norm[str(key)] = val
+            except Exception:
+                # Defensive: if normalization fails, skip this entry and continue
+                logger.debug('RedisEventQueue.dequeue_event: failed to normalize entry fields, skipping %s', entry_id)
+                continue
+
+            evt_type = norm.get('type')
+
+            # Handle tombstone/close message
+            if evt_type == 'CLOSE':
+                self._is_closed = True
+                raise asyncio.QueueEmpty('Queue closed')
+
+            raw_payload = norm.get('payload')
+            if raw_payload is None:
+                # Missing payload — likely due to key mismatch or malformed entry.
+                # Skip and continue to next entry instead of returning None to callers.
+                logger.debug('RedisEventQueue.dequeue_event: skipping entry %s with missing payload', entry_id)
+                # continue loop to read next entry
+                continue
+
+            # If payload is a JSON string, parse it; otherwise, use as-is.
+            if isinstance(raw_payload, str):
+                try:
+                    data = json.loads(raw_payload)
+                except json.JSONDecodeError:
+                    data = raw_payload
+            else:
+                data = raw_payload
+
+            model = _TYPE_MAP.get(evt_type)
+            if model:
+                try:
+                    return model.parse_obj(data)
+                except ValidationError as exc:
+                    logger.exception('Failed to parse event payload into model')
+                    raise ValueError(f'Failed to parse event of type {evt_type}') from exc
+
+            # Unknown type — return raw data for flexibility
+            logger.debug('Unknown event type: %s, returning raw payload', evt_type)
+            return data
+
+    def task_done(self) -> None:  # streams do not require task_done semantics
+        """No-op for Redis streams (kept for API compatibility)."""
+
+    def tap(self) -> EventQueue:
+        """Return a new RedisEventQueue that starts at the stream tail ('$')."""
+        q = RedisEventQueue(
+            task_id=self._task_id,
+            redis_client=self._redis,
+            stream_prefix=self._stream_key.rsplit(':', 1)[0],
+            maxlen=self._maxlen,
+            read_block_ms=self._read_block_ms,
+        )
+        # Set tap's cursor to the current last entry id so it receives only
+        # events appended after this point.
+        try:
+            lst = getattr(self._redis, 'streams', {}).get(self._stream_key, [])
+            if lst:
+                q._last_id = lst[-1][0]
+            else:
+                q._last_id = '0-0'
+        except (AttributeError, KeyError, IndexError, TypeError):
+            # Fallback: start at stream tail if we can't determine the last ID
+            q._last_id = '$'
+        return q
+
+    async def close(self, immediate: bool = False) -> None:
+        """Mark the stream closed and publish a tombstone entry for readers."""
+        try:
+            await self._redis.set(f'{self._stream_key}:closed', '1')
+            await self._redis.xadd(self._stream_key, {'type': 'CLOSE'})
+        except RedisError:
+            logger.exception('Failed to write close marker to redis')
+
+    def is_closed(self) -> bool:
+        """Return True if this queue has been closed (close() called)."""
+        return self._is_closed
+
+    async def clear_events(self, clear_child_queues: bool = True) -> None:
+        """Attempt to remove the underlying redis stream (best-effort)."""
+        try:
+            await self._redis.delete(self._stream_key)
+        except RedisError:
+            logger.exception('Failed to delete redis stream during clear_events')