fix: resolve Ruff linting errors in Redis event handling

Author: mjunaidca

src/a2a/server/events/redis_event_consumer.py

@@ -3,7 +3,9 @@
 import asyncio
 import logging
 
-from typing import Protocol, TYPE_CHECKING
+from typing import TYPE_CHECKING, Protocol
+
+
 if TYPE_CHECKING:
     from collections.abc import AsyncGenerator
 
@@ -24,6 +26,7 @@ def is_closed(self) -> bool:
         """Return True if the underlying queue has been closed."""
         ...
 
+
 logger = logging.getLogger(__name__)
 
 
@@ -38,6 +41,7 @@ class RedisEventConsumer:
     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)

src/a2a/server/events/redis_event_queue.py

@@ -5,28 +5,36 @@
 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
+
+from a2a.server.events.event_queue import EventQueue
+
+
 if TYPE_CHECKING:
     from a2a.server.events.event_queue import Event
 from pydantic import ValidationError
+
 from a2a.types import (
     Message,
     Task,
-    TaskStatusUpdateEvent,
     TaskArtifactUpdateEvent,
+    TaskStatusUpdateEvent,
 )
 from a2a.utils.telemetry import SpanKind, trace_class
 
+
 logger = logging.getLogger(__name__)
 
 
@@ -98,7 +106,7 @@ async def enqueue_event(self, event: Event) -> None:
         except RedisError:
             logger.exception('Failed to XADD event to redis stream')
 
-    async def dequeue_event(self, no_wait: bool = False) -> Event | Any:
+    async def dequeue_event(self, no_wait: bool = False) -> Event | Any:  # noqa: PLR0912
         """Read one event from the Redis stream respecting no_wait semantics.
 
         Returns a parsed pydantic model matching the event type.
@@ -128,18 +136,25 @@ async def dequeue_event(self, no_wait: bool = False) -> Event | Any:
             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)):
+                    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:
+                        except UnicodeDecodeError:
                             val = v
                     else:
                         val = v
                     norm[str(key)] = val
-            except Exception:
+            except Exception:  # noqa: BLE001
                 # Defensive: if normalization fails, skip this entry and continue
-                logger.debug('RedisEventQueue.dequeue_event: failed to normalize entry fields, skipping %s', entry_id)
+                logger.debug(
+                    'RedisEventQueue.dequeue_event: failed to normalize entry fields, skipping %s',
+                    entry_id,
+                )
                 continue
 
             evt_type = norm.get('type')
@@ -153,7 +168,10 @@ async def dequeue_event(self, no_wait: bool = False) -> Event | Any:
             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)
+                logger.debug(
+                    'RedisEventQueue.dequeue_event: skipping entry %s with missing payload',
+                    entry_id,
+                )
                 # continue loop to read next entry
                 continue
 
@@ -171,12 +189,17 @@ async def dequeue_event(self, no_wait: bool = False) -> Event | Any:
                 try:
                     return model.parse_obj(data)
                 except ValidationError as exc:
-                    logger.debug('Failed to parse event payload into model, returning raw data: %s', exc)
+                    logger.debug(
+                        'Failed to parse event payload into model, returning raw data: %s',
+                        exc,
+                    )
                     # Return raw data for flexibility when parsing fails
                     return data
 
             # Unknown type — return raw data for flexibility
-            logger.debug('Unknown event type: %s, returning raw payload', evt_type)
+            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
@@ -226,4 +249,6 @@ async def clear_events(self, clear_child_queues: bool = True) -> None:
         try:
             await self._redis.delete(self._stream_key)
         except RedisError:
-            logger.exception('Failed to delete redis stream during clear_events')
+            logger.exception(
+                'Failed to delete redis stream during clear_events'
+            )

src/a2a/server/events/redis_queue_manager.py

@@ -1,10 +1,12 @@
 from __future__ import annotations
 
 import logging
+
 from typing import TYPE_CHECKING, Any
 
 from a2a.server.events.queue_manager import QueueManager
 
+
 if TYPE_CHECKING:
     from a2a.server.events.event_queue import EventQueue
 
@@ -25,7 +27,9 @@ class RedisQueueManager(QueueManager):
     All coordination happens through Redis streams.
     """
 
-    def __init__(self, redis_client: Any, stream_prefix: str = 'a2a:task') -> None:
+    def __init__(
+        self, redis_client: Any, stream_prefix: str = 'a2a:task'
+    ) -> None:
         self._redis = redis_client
         self._stream_prefix = stream_prefix
 
@@ -92,7 +96,7 @@ async def close(self, task_id: str) -> None:
             if result and result[0][1].get('type') == 'CLOSE':
                 # Stream is already closed, no need to add another CLOSE entry
                 return
-        except Exception as exc:
+        except Exception as exc:  # noqa: BLE001
             # If we can't check (e.g., stream doesn't exist), proceed with closing
             logger.debug('Could not check if stream is already closed: %s', exc)
 
@@ -114,7 +118,7 @@ async def create_or_tap(self, task_id: str) -> EventQueue:
         logger.info('create_or_tap called with task_id: %s', task_id)
         logger.info('RedisEventQueue value: %s', RedisEventQueue)
         logger.info('RedisEventQueue type: %s', type(RedisEventQueue))
-        
+
         if RedisEventQueue is None:
             logger.error('RedisEventQueue is None - import failed!')
             raise RuntimeError(

src/a2a/server/request_handlers/default_request_handler.py

@@ -94,11 +94,13 @@ def __init__(  # noqa: PLR0913
         """
         self.agent_executor = agent_executor
         self.task_store = task_store
-        
+
         # Handle queue_manager with deprecation warning for backward compatibility
         if queue_manager is None:
             # Allow disabling fallback via environment variable for strict production deployments
-            disable_fallback = os.getenv('A2A_DISABLE_QUEUE_MANAGER_FALLBACK', '').lower() in ('true', '1', 'yes')
+            disable_fallback = os.getenv(
+                'A2A_DISABLE_QUEUE_MANAGER_FALLBACK', ''
+            ).lower() in ('true', '1', 'yes')
 
             if disable_fallback:
                 raise ValueError(
@@ -110,12 +112,12 @@ def __init__(  # noqa: PLR0913
                 'Using default InMemoryQueueManager. This will be removed in a future version. '
                 'Please explicitly pass a QueueManager instance to ensure proper production deployment.',
                 DeprecationWarning,
-                stacklevel=2
+                stacklevel=2,
             )
             self._queue_manager = InMemoryQueueManager()
         else:
             self._queue_manager = queue_manager
-            
+
         self._push_config_store = push_config_store
         self._push_sender = push_sender
         self._request_context_builder = (

src/a2a/server/request_handlers/redis_request_handler.py

@@ -2,8 +2,10 @@
 
 from typing import Any
 
-from a2a.server.request_handlers.default_request_handler import DefaultRequestHandler
-from  a2a.server.events.redis_queue_manager import RedisQueueManager
+from a2a.server.events.redis_queue_manager import RedisQueueManager
+from a2a.server.request_handlers.default_request_handler import (
+    DefaultRequestHandler,
+)
 
 
 def create_redis_request_handler(
@@ -19,5 +21,12 @@ def create_redis_request_handler(
     provided `redis_client` and passes it into `DefaultRequestHandler` so the
     rest of the application can remain unchanged.
     """
-    queue_manager = RedisQueueManager(redis_client=redis_client, stream_prefix=stream_prefix)
-    return DefaultRequestHandler(agent_executor=agent_executor, task_store=task_store, queue_manager=queue_manager, **kwargs)
+    queue_manager = RedisQueueManager(
+        redis_client=redis_client, stream_prefix=stream_prefix
+    )
+    return DefaultRequestHandler(
+        agent_executor=agent_executor,
+        task_store=task_store,
+        queue_manager=queue_manager,
+        **kwargs,
+    )

src/a2a/utils/stream_write/redis_stream_writer.py

@@ -8,6 +8,7 @@
 import logging
 
 from datetime import datetime, timezone
+from types import TracebackType
 from typing import Any
 
 
@@ -58,19 +59,26 @@ async def disconnect(self) -> None:
             self._connected = False
             logger.info('Disconnected from Redis')
 
-    async def __aenter__(self):
+    async def __aenter__(self) -> 'RedisStreamInjector':
+        """Enter the async context manager."""
         await self.connect()
         return self
 
-    async def __aexit__(self, exc_type, exc_val, exc_tb):
+    async def __aexit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc_val: BaseException | None,
+        exc_tb: TracebackType | None,
+    ) -> None:
+        """Exit the async context manager."""
         await self.disconnect()
 
     def _get_stream_key(self, task_id: str) -> str:
         """Get the Redis stream key for a task."""
         if not task_id:
             raise ValueError('task_id cannot be empty')
         stream_key = f'a2a:task:{task_id}'
-        logger.debug(f'Generated stream key: {stream_key}')
+        logger.debug('Generated stream key: %s', stream_key)
         return stream_key
 
     def _serialize_event(
@@ -105,10 +113,7 @@ async def stream_message(
         if not context_id:
             raise ValueError('context_id cannot be empty')
 
-        if isinstance(message, dict):
-            data = message
-        else:
-            data = message.model_dump()
+        data = message if isinstance(message, dict) else message.model_dump()
 
         event_data = self._serialize_event('Message', data)
         return await self._append_to_stream(task_id, event_data)
@@ -227,7 +232,7 @@ async def get_latest_event(self, task_id: str) -> dict[str, Any] | None:
             if result:
                 entry_id, fields = result[0]
                 return {'id': entry_id, **fields}
-        except Exception as e:
+        except Exception as e:  # noqa: BLE001
             logger.warning(
                 'Failed to get latest event',
                 extra={'task_id': task_id, 'error': str(e)},
@@ -247,7 +252,7 @@ async def get_events_since(self, task_id: str, start_id: str = '0') -> list:
         try:
             result = await self._client.xrange(stream_key, start_id, '+')
             return [{'id': entry_id, **fields} for entry_id, fields in result]
-        except Exception as e:
+        except Exception as e:  # noqa: BLE001
             logger.warning(
                 'Failed to get events',
                 extra={'task_id': task_id, 'error': str(e)},