feat: Update to support python 3.12

- add case statement to use the asyncio.Queue.shutdown method for 3.13+
- add special handling to allow for similar semantics as
  asyncio.Queue.shutdown for 3.12

Tested on multiple samples in the a2a repo and some examples in this
repo

Author: pstephengoogle

pyproject.toml

@@ -5,7 +5,7 @@ description = "A2A Python SDK"
 readme = "README.md"
 license = { file = "LICENSE" }
 authors = [{ name = "Google LLC", email = "[email protected]" }]
-requires-python = ">=3.13"
+requires-python = ">=3.12"
 keywords = ["A2A", "A2A SDK", "A2A Protocol", "Agent2Agent"]
 dependencies = [
     "httpx>=0.28.1",
@@ -22,7 +22,7 @@ classifiers = [
   "Intended Audience :: Developers",
   "Programming Language :: Python",
   "Programming Language :: Python :: 3",
-  "Programming Language :: Python :: 3.13",
+  "Programming Language :: Python :: 3.12",
   "Operating System :: OS Independent",
   "Topic :: Software Development :: Libraries :: Python Modules",
   "License :: OSI Approved :: Apache Software License",

src/a2a/server/events/event_consumer.py

@@ -1,5 +1,6 @@
 import asyncio
 import logging
+import sys
 
 from collections.abc import AsyncGenerator
 
@@ -14,6 +15,12 @@
 from a2a.utils.errors import ServerError
 from a2a.utils.telemetry import SpanKind, trace_class
 
+# This is an alias to the execption for closed queue
+QueueClosed = asyncio.QueueEmpty
+
+# When using python 3.13 or higher, the closed queue signal is QueueShutdown
+if sys.version_info >= (3, 13):
+    QueueClosed = asyncio.QueueShutDown
 
 logger = logging.getLogger(__name__)
 
@@ -111,13 +118,16 @@ async def consume_all(self) -> AsyncGenerator[Event]:
 
                 if is_final_event:
                     logger.debug('Stopping event consumption in consume_all.')
-                    self.queue.close()
+                    await self.queue.close()
                     break
             except TimeoutError:
                 # continue polling until there is a final event
                 continue
-            except asyncio.QueueShutDown:
-                break
+            except QueueClosed:
+                # Confirm that the queue is closed, e.g. we aren't on
+                # python 3.12 and get a queue empty error on an open queue
+                if self.queue.is_closed():
+                    break
 
     def agent_task_callback(self, agent_task: asyncio.Task[None]):
         """Callback to handle exceptions from the agent's execution task.

src/a2a/server/events/event_queue.py

@@ -1,5 +1,6 @@
 import asyncio
 import logging
+import sys
 
 from a2a.types import (
     A2AError,
@@ -39,6 +40,8 @@ def __init__(self) -> None:
         """Initializes the EventQueue."""
         self.queue: asyncio.Queue[Event] = asyncio.Queue()
         self._children: list[EventQueue] = []
+        self._is_closed = False
+        self._lock = asyncio.Lock()
         logger.debug('EventQueue initialized.')
 
     def enqueue_event(self, event: Event):
@@ -47,6 +50,9 @@ def enqueue_event(self, event: Event):
         Args:
             event: The event object to enqueue.
         """
+        if self._is_closed:
+            logger.warning('Queue is closed. Event will not be enqueued.')
+            return
         logger.debug(f'Enqueuing event of type: {type(event)}')
         self.queue.put_nowait(event)
         for child in self._children:
@@ -66,6 +72,11 @@ async def dequeue_event(self, no_wait: bool = False) -> Event:
             asyncio.QueueEmpty: If `no_wait` is True and the queue is empty.
             asyncio.QueueShutDown: If the queue has been closed and is empty.
         """
+        async with self._lock:
+            if self._is_closed and self.queue.empty():
+                logger.warning('Queue is closed. Event will not be dequeued.')
+                raise asyncio.QueueEmpty('Queue is closed.')
+
         if no_wait:
             logger.debug('Attempting to dequeue event (no_wait=True).')
             event = self.queue.get_nowait()
@@ -99,13 +110,30 @@ def tap(self) -> 'EventQueue':
         self._children.append(queue)
         return queue
 
-    def close(self):
+    async def close(self):
         """Closes the queue for future push events.
 
         Once closed, `dequeue_event` will eventually raise `asyncio.QueueShutDown`
         when the queue is empty. Also closes all child queues.
         """
         logger.debug('Closing EventQueue.')
-        self.queue.shutdown()
-        for child in self._children:
-            child.close()
+        async with self._lock:
+            # If already closed, just return.
+            if self._is_closed:
+                return
+            self._is_closed = True
+        # If using python 3.13 or higher, use the shutdown method
+        if sys.version_info >= (3, 13):
+            self.queue.shutdown()
+            for child in self._children:
+                child.close()
+        # Otherwise, join the queue
+        else:
+            tasks = [asyncio.create_task(self.queue.join())]
+            for child in self._children:
+                tasks.append(asyncio.create_task(child.close()))
+            await asyncio.wait(tasks, return_when=asyncio.ALL_COMPLETED)
+
+    def is_closed(self) -> bool:
+        """Checks if the queue is closed."""
+        return self._is_closed

src/a2a/server/events/in_memory_queue_manager.py

@@ -69,7 +69,7 @@ async def close(self, task_id: str):
             if task_id not in self._task_queue:
                 raise NoTaskQueue()
             queue = self._task_queue.pop(task_id)
-            queue.close()
+            await queue.close()
 
     async def create_or_tap(self, task_id: str) -> EventQueue:
         """Creates a new event queue for a task ID if one doesn't exist, otherwise taps the existing one.

src/a2a/server/request_handlers/default_request_handler.py

@@ -147,7 +147,7 @@ async def _run_event_stream(
             queue: The event queue for the agent to publish to.
         """
         await self.agent_executor.execute(request, queue)
-        queue.close()
+        await queue.close()
 
     async def on_message_send(
         self, params: MessageSendParams