Merge branch 'main' into sql-support

Author: kthota-g

.github/workflows/update-a2a-types.yml

@@ -68,8 +68,8 @@ jobs:
           token: ${{ secrets.A2A_BOT_PAT }}
           committer: "a2a-bot <[email protected]>"
           author: "a2a-bot <[email protected]>"
-          commit-message: "${{github.event.client_payload.message}} 🤖"
-          title: "${{github.event.client_payload.message}} 🤖"
+          commit-message: "chore: Update A2A types from specification 🤖"
+          title: "chore: Update A2A types from specification 🤖"
           body: |
             This PR updates `src/a2a/types.py` based on the latest `specification/json/a2a.json` from [google-a2a/A2A](https://github.com/google-a2a/A2A/commit/${{ github.event.client_payload.sha }}).
           branch: "auto-update-a2a-types-${{ github.event.client_payload.sha }}"

README.md

@@ -10,7 +10,7 @@
    <h2 align="center">
    <img src="https://raw.githubusercontent.com/google-a2a/A2A/refs/heads/main/docs/assets/a2a-logo-black.svg" width="256" alt="A2A Logo"/>
    </h2>
-   <h3 align="center">A Python library that helps run agentic applications as A2AServers following the <a href="https://google.github.io/A2A">Agent2Agent (A2A) Protocol</a>.</h3>
+   <h3 align="center">A Python library that helps run agentic applications as A2AServers following the <a href="https://google-a2a.github.io/A2A">Agent2Agent (A2A) Protocol</a>.</h3>
 </html>
 
 <!-- markdownlint-enable no-inline-html -->

pyproject.toml

@@ -51,10 +51,10 @@ sql = [
 ]
 
 [project.urls]
-homepage = "https://google.github.io/A2A/"
+homepage = "https://google-a2a.github.io/A2A/"
 repository = "https://github.com/google-a2a/a2a-python"
 changelog = "https://github.com/google-a2a/a2a-python/blob/main/CHANGELOG.md"
-documentation = "https://google.github.io/A2A/"
+documentation = "https://google-a2a.github.io/A2A/sdk/python/"
 
 [tool.hatch.build.targets.wheel]
 packages = ["src/a2a"]

src/a2a/client/client.py

@@ -1,26 +1,37 @@
 import json
 import logging
+
 from collections.abc import AsyncGenerator
 from typing import Any
 from uuid import uuid4
 
 import httpx
+
 from httpx_sse import SSEError, aconnect_sse
 from pydantic import ValidationError
 
 from a2a.client.errors import A2AClientHTTPError, A2AClientJSONError
-from a2a.types import (AgentCard, CancelTaskRequest, CancelTaskResponse,
-                       GetTaskPushNotificationConfigRequest,
-                       GetTaskPushNotificationConfigResponse, GetTaskRequest,
-                       GetTaskResponse, SendMessageRequest,
-                       SendMessageResponse, SendStreamingMessageRequest,
-                       SendStreamingMessageResponse,
-                       SetTaskPushNotificationConfigRequest,
-                       SetTaskPushNotificationConfigResponse)
+from a2a.types import (
+    AgentCard,
+    CancelTaskRequest,
+    CancelTaskResponse,
+    GetTaskPushNotificationConfigRequest,
+    GetTaskPushNotificationConfigResponse,
+    GetTaskRequest,
+    GetTaskResponse,
+    SendMessageRequest,
+    SendMessageResponse,
+    SendStreamingMessageRequest,
+    SendStreamingMessageResponse,
+    SetTaskPushNotificationConfigRequest,
+    SetTaskPushNotificationConfigResponse,
+)
 from a2a.utils.telemetry import SpanKind, trace_class
 
+
 logger = logging.getLogger(__name__)
 
+
 class A2ACardResolver:
     """Agent Card resolver."""
 
@@ -160,6 +171,7 @@ async def get_client_from_agent_card_url(
             agent_card_path: The path to the agent card endpoint, relative to the base URL.
             http_kwargs: Optional dictionary of keyword arguments to pass to the
                 underlying httpx.get request when fetching the agent card.
+
         Returns:
             An initialized `A2AClient` instance.
 
@@ -169,7 +181,9 @@ async def get_client_from_agent_card_url(
         """
         agent_card: AgentCard = await A2ACardResolver(
             httpx_client, base_url=base_url, agent_card_path=agent_card_path
-        ).get_agent_card(http_kwargs=http_kwargs) # Fetches public card by default
+        ).get_agent_card(
+            http_kwargs=http_kwargs
+        )  # Fetches public card by default
         return A2AClient(httpx_client=httpx_client, agent_card=agent_card)
 
     async def send_message(

src/a2a/server/apps/starlette_app.py

@@ -45,8 +45,20 @@
 
 logger = logging.getLogger(__name__)
 
-# Register Starlette User as an implementation of a2a.auth.user.User
-A2AUser.register(BaseUser)
+
+class StarletteUserProxy(A2AUser):
+    """Adapts the Starlette User class to the A2A user representation."""
+
+    def __init__(self, user: BaseUser):
+        self._user = user
+
+    @property
+    def is_authenticated(self):
+        return self._user.is_authenticated
+
+    @property
+    def user_name(self):
+        return self._user.display_name
 
 
 class CallContextBuilder(ABC):
@@ -64,7 +76,7 @@ def build(self, request: Request) -> ServerCallContext:
         user = UnauthenticatedUser()
         state = {}
         with contextlib.suppress(Exception):
-            user = request.user
+            user = StarletteUserProxy(request.user)
             state['auth'] = request.auth
         return ServerCallContext(user=user, state=state)
 
@@ -139,7 +151,7 @@ def _generate_error_response(
             log_level,
             f'Request Error (ID: {request_id}): '
             f"Code={error_resp.error.code}, Message='{error_resp.error.message}'"
-            f'{", Data=" + str(error_resp.error.data) if hasattr(error, "data") and error_resp.error.data else ""}',
+            f'{", Data=" + str(error_resp.error.data) if error_resp.error.data else ""}',
         )
         return JSONResponse(
             error_resp.model_dump(mode='json', exclude_none=True),

src/a2a/server/events/event_queue.py

@@ -14,14 +14,11 @@
 logger = logging.getLogger(__name__)
 
 
-Event = (
-    Message
-    | Task
-    | TaskStatusUpdateEvent
-    | TaskArtifactUpdateEvent
-)
+Event = Message | Task | TaskStatusUpdateEvent | TaskArtifactUpdateEvent
 """Type alias for events that can be enqueued."""
 
+DEFAULT_MAX_QUEUE_SIZE = 1024
+
 
 @trace_class(kind=SpanKind.SERVER)
 class EventQueue:
@@ -32,27 +29,37 @@ class EventQueue:
     to create child queues that receive the same events.
     """
 
-    def __init__(self) -> None:
+    def __init__(self, max_queue_size: int = DEFAULT_MAX_QUEUE_SIZE) -> None:
         """Initializes the EventQueue."""
-        self.queue: asyncio.Queue[Event] = asyncio.Queue()
+        # Make sure the `asyncio.Queue` is bounded.
+        # If it's unbounded (maxsize=0), then `queue.put()` never needs to wait,
+        # and so the streaming won't work correctly.
+        if max_queue_size <= 0:
+            raise ValueError('max_queue_size must be greater than 0')
+
+        self.queue: asyncio.Queue[Event] = asyncio.Queue(maxsize=max_queue_size)
         self._children: list[EventQueue] = []
         self._is_closed = False
         self._lock = asyncio.Lock()
         logger.debug('EventQueue initialized.')
 
-    def enqueue_event(self, event: Event):
+    async def enqueue_event(self, event: Event):
         """Enqueues an event to this queue and all its children.
 
         Args:
             event: The event object to enqueue.
         """
-        if self._is_closed:
-            logger.warning('Queue is closed. Event will not be enqueued.')
-            return
+        async with self._lock:
+            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)
+
+        # Make sure to use put instead of put_nowait to avoid blocking the event loop.
+        await self.queue.put(event)
         for child in self._children:
-            child.enqueue_event(event)
+            await child.enqueue_event(event)
 
     async def dequeue_event(self, no_wait: bool = False) -> Event:
         """Dequeues an event from the queue.

src/a2a/server/tasks/task_updater.py

@@ -1,5 +1,6 @@
 import uuid
 
+from datetime import datetime, timezone
 from typing import Any
 
 from a2a.server.events import EventQueue
@@ -33,29 +34,38 @@ def __init__(self, event_queue: EventQueue, task_id: str, context_id: str):
         self.task_id = task_id
         self.context_id = context_id
 
-    def update_status(
-        self, state: TaskState, message: Message | None = None, final=False
+    async def update_status(
+        self,
+        state: TaskState,
+        message: Message | None = None,
+        final=False,
+        timestamp: str | None = None,
     ):
         """Updates the status of the task and publishes a `TaskStatusUpdateEvent`.
 
         Args:
             state: The new state of the task.
             message: An optional message associated with the status update.
             final: If True, indicates this is the final status update for the task.
+            timestamp: Optional ISO 8601 datetime string. Defaults to current time.
         """
-        self.event_queue.enqueue_event(
+        current_timestamp = (
+            timestamp if timestamp else datetime.now(timezone.utc).isoformat()
+        )
+        await self.event_queue.enqueue_event(
             TaskStatusUpdateEvent(
                 taskId=self.task_id,
                 contextId=self.context_id,
                 final=final,
                 status=TaskStatus(
                     state=state,
                     message=message,
+                    timestamp=current_timestamp,
                 ),
             )
         )
 
-    def add_artifact(
+    async def add_artifact(
         self,
         parts: list[Part],
         artifact_id: str = str(uuid.uuid4()),
@@ -72,7 +82,7 @@ def add_artifact(
             append: Optional boolean indicating if this chunk appends to a previous one.
             last_chunk: Optional boolean indicating if this is the last chunk.
         """
-        self.event_queue.enqueue_event(
+        await self.event_queue.enqueue_event(
             TaskArtifactUpdateEvent(
                 taskId=self.task_id,
                 contextId=self.context_id,
@@ -85,32 +95,34 @@ def add_artifact(
             )
         )
 
-    def complete(self, message: Message | None = None):
+    async def complete(self, message: Message | None = None):
         """Marks the task as completed and publishes a final status update."""
-        self.update_status(
+        await self.update_status(
             TaskState.completed,
             message=message,
             final=True,
         )
 
-    def failed(self, message: Message | None = None):
+    async def failed(self, message: Message | None = None):
         """Marks the task as failed and publishes a final status update."""
-        self.update_status(TaskState.failed, message=message, final=True)
-    
-    def reject(self, message: Message | None = None):
+        await self.update_status(TaskState.failed, message=message, final=True)
+
+    async def reject(self, message: Message | None = None):
         """Marks the task as rejected and publishes a final status update."""
-        self.update_status(TaskState.rejected, message=message, final=True)
+        await self.update_status(
+            TaskState.rejected, message=message, final=True
+        )
 
-    def submit(self, message: Message | None = None):
+    async def submit(self, message: Message | None = None):
         """Marks the task as submitted and publishes a status update."""
-        self.update_status(
+        await self.update_status(
             TaskState.submitted,
             message=message,
         )
 
-    def start_work(self, message: Message | None = None):
+    async def start_work(self, message: Message | None = None):
         """Marks the task as working and publishes a status update."""
-        self.update_status(
+        await self.update_status(
             TaskState.working,
             message=message,
         )

src/a2a/types.py

@@ -100,15 +100,15 @@ class AgentSkill(BaseModel):
     """
     The set of interaction modes that the skill supports
     (if different than the default).
-    Supported mime types for input.
+    Supported media types for input.
     """
     name: str
     """
     Human readable name of the skill.
     """
     outputModes: list[str] | None = None
     """
-    Supported mime types for output.
+    Supported media types for output.
     """
     tags: list[str]
     """
@@ -583,6 +583,10 @@ class PushNotificationConfig(BaseModel):
     """
 
     authentication: PushNotificationAuthenticationInfo | None = None
+    id: str | None = None
+    """
+    Push Notification ID - created by server to support multiple callbacks
+    """
     token: str | None = None
     """
     Token unique to this task/session.
@@ -1368,11 +1372,11 @@ class AgentCard(BaseModel):
     defaultInputModes: list[str]
     """
     The set of interaction modes that the agent supports across all skills. This can be overridden per-skill.
-    Supported mime types for input.
+    Supported media types for input.
     """
     defaultOutputModes: list[str]
     """
-    Supported mime types for output.
+    Supported media types for output.
     """
     description: str
     """

src/a2a/utils/helpers.py

@@ -69,7 +69,7 @@ def append_artifact_to_task(task: Task, event: TaskArtifactUpdateEvent) -> None:
 
     # Find existing artifact by its id
     for i, art in enumerate(task.artifacts):
-        if hasattr(art, 'artifactId') and art.artifactId == artifact_id:
+        if art.artifactId == artifact_id:
             existing_artifact = art
             existing_artifact_list_index = i
             break