Merge branch 'main' into make-fastapi-package-optional

Author: holtskinner

CHANGELOG.md

@@ -1,5 +1,25 @@
 # Changelog
 
+## [0.2.11](https://github.com/a2aproject/a2a-python/compare/v0.2.10...v0.2.11) (2025-07-07)
+
+
+### ⚠ BREAKING CHANGES
+
+* Removes `push_notifier` interface from the SDK and introduces `push_notification_config_store` and `push_notification_sender` for supporting push notifications.
+
+### Features
+
+* Add constants for Well-Known URIs ([#271](https://github.com/a2aproject/a2a-python/issues/271)) ([1c8e12e](https://github.com/a2aproject/a2a-python/commit/1c8e12e448dc7469e508fccdac06818836f5b520))
+* Adds support for List and Delete push notification configurations. ([f1b576e](https://github.com/a2aproject/a2a-python/commit/f1b576e061e7a3ab891d8368ade56c7046684c5e))
+* Adds support for more than one `push_notification_config` per task. ([f1b576e](https://github.com/a2aproject/a2a-python/commit/f1b576e061e7a3ab891d8368ade56c7046684c5e))
+* **server:** Add lock to TaskUpdater to prevent race conditions ([#279](https://github.com/a2aproject/a2a-python/issues/279)) ([1022093](https://github.com/a2aproject/a2a-python/commit/1022093110100da27f040be4b35831bf8b1fe094))
+* Support for database backend Task Store ([#259](https://github.com/a2aproject/a2a-python/issues/259)) ([7c46e70](https://github.com/a2aproject/a2a-python/commit/7c46e70b3142f3ec274c492bacbfd6e8f0204b36))
+
+
+### Code Refactoring
+
+* Removes `push_notifier` interface from the SDK and introduces `push_notification_config_store` and `push_notification_sender` for supporting push notifications. ([f1b576e](https://github.com/a2aproject/a2a-python/commit/f1b576e061e7a3ab891d8368ade56c7046684c5e))
+
 ## [0.2.10](https://github.com/a2aproject/a2a-python/compare/v0.2.9...v0.2.10) (2025-06-30)
 
 

src/a2a/server/apps/jsonrpc/fastapi_app.py

@@ -1,5 +1,7 @@
 import logging
 
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
 from typing import TYPE_CHECKING, Any
 
 
@@ -22,7 +24,7 @@
     JSONRPCApplication,
 )
 from a2a.server.request_handlers.jsonrpc_handler import RequestHandler
-from a2a.types import AgentCard
+from a2a.types import A2ARequest, AgentCard
 from a2a.utils.constants import (
     AGENT_CARD_WELL_KNOWN_PATH,
     DEFAULT_RPC_URL,
@@ -88,7 +90,22 @@ def add_routes_to_app(
             rpc_url: The URL for the A2A JSON-RPC endpoint.
             extended_agent_card_url: The URL for the authenticated extended agent card endpoint.
         """
-        app.post(rpc_url)(self._handle_requests)
+        app.post(
+            rpc_url,
+            openapi_extra={
+                'requestBody': {
+                    'content': {
+                        'application/json': {
+                            'schema': {
+                                '$ref': '#/components/schemas/A2ARequest'
+                            }
+                        }
+                    },
+                    'required': True,
+                    'description': 'A2ARequest',
+                }
+            },
+        )(self._handle_requests)
         app.get(agent_card_url)(self._handle_get_agent_card)
 
         if self.agent_card.supportsAuthenticatedExtendedCard:
@@ -114,7 +131,23 @@ def build(
         Returns:
             A configured FastAPI application instance.
         """
-        app = FastAPI(**kwargs)
+
+        @asynccontextmanager
+        async def lifespan(app: FastAPI) -> AsyncIterator[None]:
+            a2a_request_schema = A2ARequest.model_json_schema(
+                ref_template='#/components/schemas/{model}'
+            )
+            defs = a2a_request_schema.pop('$defs', {})
+            openapi_schema = app.openapi()
+            component_schemas = openapi_schema.setdefault(
+                'components', {}
+            ).setdefault('schemas', {})
+            component_schemas.update(defs)
+            component_schemas['A2ARequest'] = a2a_request_schema
+
+            yield
+
+        app = FastAPI(lifespan=lifespan, **kwargs)
 
         self.add_routes_to_app(
             app, agent_card_url, rpc_url, extended_agent_card_url

src/a2a/server/tasks/task_updater.py

@@ -1,3 +1,4 @@
+import asyncio
 import uuid
 
 from datetime import datetime, timezone
@@ -33,6 +34,14 @@ def __init__(self, event_queue: EventQueue, task_id: str, context_id: str):
         self.event_queue = event_queue
         self.task_id = task_id
         self.context_id = context_id
+        self._lock = asyncio.Lock()
+        self._terminal_state_reached = False
+        self._terminal_states = {
+            TaskState.completed,
+            TaskState.canceled,
+            TaskState.failed,
+            TaskState.rejected,
+        }
 
     async def update_status(
         self,
@@ -49,21 +58,26 @@ async def update_status(
             final: If True, indicates this is the final status update for the task.
             timestamp: Optional ISO 8601 datetime string. Defaults to current time.
         """
-        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,
-                ),
+        async with self._lock:
+            if self._terminal_state_reached:
+                raise RuntimeError(f"Task {self.task_id} is already in a terminal state.")
+            if state in self._terminal_states:
+                self._terminal_state_reached = True
+                final = True
+
+            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,
+                    ),
+                )
             )
-        )
 
     async def add_artifact(  # noqa: PLR0913
         self,

tests/server/tasks/test_task_updater.py

@@ -1,5 +1,5 @@
+import asyncio
 import uuid
-
 from unittest.mock import AsyncMock, patch
 
 import pytest
@@ -505,3 +505,31 @@ async def test_cancel_with_message(task_updater, event_queue, sample_message):
     assert event.status.state == TaskState.canceled
     assert event.final is True
     assert event.status.message == sample_message
+
+
+@pytest.mark.asyncio
+async def test_update_status_raises_error_if_terminal_state_reached(task_updater, event_queue):
+    await task_updater.complete()
+    event_queue.reset_mock()
+    with pytest.raises(RuntimeError):
+        await task_updater.start_work()
+    event_queue.enqueue_event.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_concurrent_updates_race_condition(event_queue):
+    task_updater = TaskUpdater(
+        event_queue=event_queue,
+        task_id="test-task-id",
+        context_id="test-context-id",
+    )
+    tasks = [
+        task_updater.complete(),
+        task_updater.failed(),
+    ]
+    results = await asyncio.gather(*tasks, return_exceptions=True)
+    successes = [r for r in results if not isinstance(r, Exception)]
+    failures = [r for r in results if isinstance(r, RuntimeError)]
+    assert len(successes) == 1
+    assert len(failures) == 1
+    assert event_queue.enqueue_event.call_count == 1