Merge branch 'main' into docs/langgraph

Author: kthota-g

.github/actions/spelling/allow.txt

@@ -1,7 +1,10 @@
 AError
 ARequest
+AServer
 AStarlette
 adk
+codegen
+datamodel
 genai
 inmemory
 langgraph
@@ -10,3 +13,4 @@ oauthoidc
 opensource
 socio
 sse
+tagwords

CODE_OF_CONDUCT.md

@@ -3,7 +3,7 @@
 ## Our Pledge
 
 In the interest of fostering an open and welcoming environment, we as
-contributors and maintainers pledge to making participation in our project and
+contributors and maintainers pledge to make participation in our project and
 our community a harassment-free experience for everyone, regardless of age, body
 size, disability, ethnicity, gender identity and expression, level of
 experience, education, socio-economic status, nationality, personal appearance,

README.md

@@ -35,4 +35,4 @@ This project is licensed under the terms of the [Apache 2.0 License](LICENSE).
 
 ## Contributing
 
-See [CONTRIBUTING.md](../CONTRIBUTING.md) for contribution guidelines.
+See [CONTRIBUTING.md](CONTRIBUTING.md) for contribution guidelines.

src/a2a/server/agent_execution/base_agent_executor.py

@@ -22,6 +22,8 @@ class EventConsumer:
 
     def __init__(self, queue: EventQueue):
         self.queue = queue
+        self._timeout = 0.5
+        self._exception: BaseException | None = None
         logger.debug('EventConsumer initialized')
 
     async def consume_one(self) -> Event:
@@ -45,8 +47,15 @@ async def consume_all(self) -> AsyncGenerator[Event]:
         """Consume all the generated streaming events from the agent."""
         logger.debug('Starting to consume all events from the queue.')
         while True:
+            if self._exception:
+                raise self._exception
             try:
-                event = await self.queue.dequeue_event()
+                # We use a timeout when waiting for an event from the queue.
+                # This is required because it allows the loop to check if 
+                # `self._exception` has been set by the `agent_task_callback`. 
+                # Without the timeout, loop might hang indefinitely if no events are
+                # enqueued by the agent and the agent simply threw an exception
+                event = await asyncio.wait_for(self.queue.dequeue_event(), timeout=self._timeout)
                 logger.debug(
                     f'Dequeued event of type: {type(event)} in consume_all.'
                 )
@@ -74,5 +83,16 @@ async def consume_all(self) -> AsyncGenerator[Event]:
                     logger.debug('Stopping event consumption in consume_all.')
                     self.queue.close()
                     break
+            except asyncio.TimeoutError:
+                # continue polling until there is a final event
+                continue
             except asyncio.QueueShutDown:
                 break
+        
+                
+
+
+
+    def agent_task_callback(self, agent_task: asyncio.Task[None]):
+            if agent_task.exception() is not None:
+                self._exception = agent_task.exception()

src/a2a/server/events/event_consumer.py

@@ -138,6 +138,7 @@ async def on_message_send(
         await self._register_producer(task_id, producer_task)
 
         consumer = EventConsumer(queue)
+        producer_task.add_done_callback(consumer.agent_task_callback)
 
         interrupted = False
         try:
@@ -192,6 +193,7 @@ async def on_message_send_stream(
 
         try:
             consumer = EventConsumer(queue)
+            producer_task.add_done_callback(consumer.agent_task_callback)
             async for event in result_aggregator.consume_and_emit(consumer):
                 # Now we know we have a Task, register the queue
                 if isinstance(event, Task):

src/a2a/server/request_handlers/default_request_handler.py

@@ -39,5 +39,5 @@ async def delete(self, task_id: str) -> None:
                 logger.info('Task %s deleted successfully.', task_id)
             else:
                 logger.warning(
-                    'Attempted to delete non-existent task with id: %s', task_id
+                    'Attempted to delete nonexistent task with id: %s', task_id
                 )

src/a2a/server/tasks/inmemory_task_store.py

@@ -933,7 +933,7 @@ class SetTaskPushNotificationConfigSuccessResponse(BaseModel):
 
 class Artifact(BaseModel):
     """
-    Represents an artifact generated for a task task.
+    Represents an artifact generated for a task.
     """
 
     artifactId: str

src/a2a/types.py

@@ -15,7 +15,7 @@
 
 
 class A2AServerError(Exception):
-    """Base exception for  A2A Server errors."""
+    """Base exception for A2A Server errors."""
 
 
 class MethodNotImplementedError(A2AServerError):

src/a2a/utils/errors.py

@@ -65,7 +65,7 @@ def append_artifact_to_task(task: Task, event: TaskArtifactUpdateEvent) -> None:
             )
             task.artifacts.append(new_artifact_data)
     elif existing_artifact:
-        # Append new parts to the existing artifact's parts list
+        # Append new parts to the existing artifact's part list
         logger.debug(
             f'Appending parts to artifact id {artifact_id} for task {task.id}'
         )
@@ -74,7 +74,7 @@ def append_artifact_to_task(task: Task, event: TaskArtifactUpdateEvent) -> None:
         # We received a chunk to append, but we don't have an existing artifact.
         # we will ignore this chunk
         logger.warning(
-            f'Received append=True for non-existent artifact index {artifact_id} in task {task.id}. Ignoring chunk.'
+            f'Received append=True for nonexistent artifact index {artifact_id} in task {task.id}. Ignoring chunk.'
         )
 
 
@@ -93,7 +93,7 @@ def wrapper(self, *args, **kwargs):
             if not expression(self):
                 if not error_message:
                     message = str(expression)
-                logger.error(f'Unsuppported Operation: {error_message}')
+                logger.error(f'Unsupported Operation: {error_message}')
                 raise ServerError(
                     UnsupportedOperationError(message=error_message)
                 )