feat: Add RESTful API Serving

.github/actions/spelling/allow.txt

@@ -53,6 +53,7 @@ lifecycles
 linting
 Llm
 lstrips
+mikeas
 mockurl
 notif
 oauthoidc
@@ -67,6 +68,7 @@ pyi
 pypistats
 pyupgrade
 pyversions
+redef
 respx
 resub
 RUF
@@ -76,5 +78,6 @@ sse
 tagwords
 taskupdate
 testuuid
+Tful
 typeerror
 vulnz

.github/workflows/unit-tests.yml

@@ -57,6 +57,6 @@ jobs:
       - name: Install dependencies
         run: uv sync --dev --extra sql --extra encryption --extra grpc --extra telemetry
       - name: Run tests and check coverage
-        run: uv run pytest --cov=a2a --cov-report=xml --cov-fail-under=89
+        run: uv run pytest --cov=a2a --cov-report term --cov-fail-under=89
       - name: Show coverage summary in log
         run: uv run coverage report

.vscode/launch.json

@@ -47,7 +47,8 @@
         "-s"
       ],
       "console": "integratedTerminal",
-      "justMyCode": true
+      "justMyCode": true,
+      "python": "${workspaceFolder}/.venv/bin/python",
     }
   ]
-}
+}

Gemini.md

@@ -0,0 +1,27 @@
+**A2A specification:** https://a2a-protocol.org/latest/specification/
+
+## Project frameworks
+- uv as package manager
+
+## How to run all tests
+1. If dependencies are not installed install them using following command
+   ```
+   uv sync --all-extras 
+   ```
+
+2. Run tests
+  ```
+  uv run pytest
+  ```
+
+## Other instructions
+1. Whenever writing python code, write types as well.
+2. After making the changes run ruff to check and fix the formatting issues
+   ```
+   uv run ruff check --fix
+   ```
+3. Run mypy type checkers to check for type errors
+   ```
+   uv run mypy
+   ```
+4. Run the unit tests to make sure that none of the unit tests are broken.

pyproject.toml

@@ -8,12 +8,14 @@ authors = [{ name = "Google LLC", email = "[email protected]" }]
 requires-python = ">=3.10"
 keywords = ["A2A", "A2A SDK", "A2A Protocol", "Agent2Agent", "Agent 2 Agent"]
 dependencies = [
-  "fastapi>=0.115.2",
+  "fastapi>=0.116.1",
   "httpx>=0.28.1",
   "httpx-sse>=0.4.0",
   "pydantic>=2.11.3",
   "sse-starlette",
-  "starlette"
+  "starlette",
+  "protobuf==5.29.5",
+  "google-api-core>=1.26.0",
 ]
 
 classifiers = [
@@ -35,7 +37,7 @@ mysql = ["sqlalchemy[asyncio,aiomysql]>=2.0.0"]
 sqlite = ["sqlalchemy[asyncio,aiosqlite]>=2.0.0"]
 sql = ["sqlalchemy[asyncio,postgresql-asyncpg,aiomysql,aiosqlite]>=2.0.0"]
 encryption = ["cryptography>=43.0.0"]
-grpc = ["grpcio>=1.60", "grpcio-tools>=1.60", "grpcio_reflection>=1.7.0", "protobuf==5.29.5", "google-api-core>=1.26.0"]
+grpc = ["grpcio>=1.60", "grpcio-tools>=1.60", "grpcio_reflection>=1.7.0"]
 telemetry = ["opentelemetry-api>=1.33.0", "opentelemetry-sdk>=1.33.0"]
 
 [project.urls]
@@ -90,6 +92,7 @@ dev = [
   "pyupgrade",
   "autoflake",
   "no_implicit_optional",
+  "trio",
 ]
 
 [[tool.uv.index]]

src/a2a/client/__init__.py

@@ -7,21 +7,24 @@
     CredentialService,
     InMemoryContextCredentialStore,
 )
-from a2a.client.client import A2ACardResolver, A2AClient
+from a2a.client.card_resolver import A2ACardResolver
+from a2a.client.client import Client, ClientConfig, ClientEvent, Consumer
+from a2a.client.client_factory import ClientFactory, minimal_agent_card
 from a2a.client.errors import (
     A2AClientError,
     A2AClientHTTPError,
     A2AClientJSONError,
     A2AClientTimeoutError,
 )
 from a2a.client.helpers import create_text_message_object
+from a2a.client.legacy import A2AClient
 from a2a.client.middleware import ClientCallContext, ClientCallInterceptor
 
 
 logger = logging.getLogger(__name__)
 
 try:
-    from a2a.client.grpc_client import A2AGrpcClient  # type: ignore
+    from a2a.client.legacy_grpc import A2AGrpcClient  # type: ignore
 except ImportError as e:
     _original_error = e
     logger.debug(
@@ -48,9 +51,15 @@ def __init__(self, *args, **kwargs):
     'A2AClientTimeoutError',
     'A2AGrpcClient',
     'AuthInterceptor',
+    'Client',
     'ClientCallContext',
     'ClientCallInterceptor',
+    'ClientConfig',
+    'ClientEvent',
+    'ClientFactory',
+    'Consumer',
     'CredentialService',
     'InMemoryContextCredentialStore',
     'create_text_message_object',
+    'minimal_agent_card',
 ]

src/a2a/client/base_client.py

@@ -0,0 +1,241 @@
+from collections.abc import AsyncIterator
+
+from a2a.client.client import (
+    Client,
+    ClientCallContext,
+    ClientConfig,
+    ClientEvent,
+    Consumer,
+)
+from a2a.client.client_task_manager import ClientTaskManager
+from a2a.client.errors import A2AClientInvalidStateError
+from a2a.client.middleware import ClientCallInterceptor
+from a2a.client.transports.base import ClientTransport
+from a2a.types import (
+    AgentCard,
+    GetTaskPushNotificationConfigParams,
+    Message,
+    MessageSendConfiguration,
+    MessageSendParams,
+    Task,
+    TaskArtifactUpdateEvent,
+    TaskIdParams,
+    TaskPushNotificationConfig,
+    TaskQueryParams,
+    TaskStatusUpdateEvent,
+)
+
+
+class BaseClient(Client):
+    """Base implementation of the A2A client, containing transport-independent logic."""
+
+    def __init__(
+        self,
+        card: AgentCard,
+        config: ClientConfig,
+        transport: ClientTransport,
+        consumers: list[Consumer],
+        middleware: list[ClientCallInterceptor],
+    ):
+        super().__init__(consumers, middleware)
+        self._card = card
+        self._config = config
+        self._transport = transport
+
+    async def send_message(
+        self,
+        request: Message,
+        *,
+        context: ClientCallContext | None = None,
+    ) -> AsyncIterator[ClientEvent | Message]:
+        """Sends a message to the agent.
+
+        This method handles both streaming and non-streaming (polling) interactions
+        based on the client configuration and agent capabilities. It will yield
+        events as they are received from the agent.
+
+        Args:
+            request: The message to send to the agent.
+            context: The client call context.
+
+        Yields:
+            An async iterator of `ClientEvent` or a final `Message` response.
+        """
+        config = MessageSendConfiguration(
+            accepted_output_modes=self._config.accepted_output_modes,
+            blocking=not self._config.polling,
+            push_notification_config=(
+                self._config.push_notification_configs[0]
+                if self._config.push_notification_configs
+                else None
+            ),
+        )
+        params = MessageSendParams(message=request, configuration=config)
+
+        if not self._config.streaming or not self._card.capabilities.streaming:
+            response = await self._transport.send_message(
+                params, context=context
+            )
+            result = (
+                (response, None) if isinstance(response, Task) else response
+            )
+            await self.consume(result, self._card)
+            yield result
+            return
+
+        tracker = ClientTaskManager()
+        stream = self._transport.send_message_streaming(params, context=context)
+
+        first_event = await anext(stream)
+        # The response from a server may be either exactly one Message or a
+        # series of Task updates. Separate out the first message for special
+        # case handling, which allows us to simplify further stream processing.
+        if isinstance(first_event, Message):
+            await self.consume(first_event, self._card)
+            yield first_event
+            return
+
+        yield await self._process_response(tracker, first_event)
+
+        async for event in stream:
+            yield await self._process_response(tracker, event)
+
+    async def _process_response(
+        self,
+        tracker: ClientTaskManager,
+        event: Task | Message | TaskStatusUpdateEvent | TaskArtifactUpdateEvent,
+    ) -> ClientEvent:
+        if isinstance(event, Message):
+            raise A2AClientInvalidStateError(
+                'received a streamed Message from server after first response; this is not supported'
+            )
+        await tracker.process(event)
+        task = tracker.get_task_or_raise()
+        update = None if isinstance(event, Task) else event
+        client_event = (task, update)
+        await self.consume(client_event, self._card)
+        return client_event
+
+    async def get_task(
+        self,
+        request: TaskQueryParams,
+        *,
+        context: ClientCallContext | None = None,
+    ) -> Task:
+        """Retrieves the current state and history of a specific task.
+
+        Args:
+            request: The `TaskQueryParams` object specifying the task ID.
+            context: The client call context.
+
+        Returns:
+            A `Task` object representing the current state of the task.
+        """
+        return await self._transport.get_task(request, context=context)
+
+    async def cancel_task(
+        self,
+        request: TaskIdParams,
+        *,
+        context: ClientCallContext | None = None,
+    ) -> Task:
+        """Requests the agent to cancel a specific task.
+
+        Args:
+            request: The `TaskIdParams` object specifying the task ID.
+            context: The client call context.
+
+        Returns:
+            A `Task` object containing the updated task status.
+        """
+        return await self._transport.cancel_task(request, context=context)
+
+    async def set_task_callback(
+        self,
+        request: TaskPushNotificationConfig,
+        *,
+        context: ClientCallContext | None = None,
+    ) -> TaskPushNotificationConfig:
+        """Sets or updates the push notification configuration for a specific task.
+
+        Args:
+            request: The `TaskPushNotificationConfig` object with the new configuration.
+            context: The client call context.
+
+        Returns:
+            The created or updated `TaskPushNotificationConfig` object.
+        """
+        return await self._transport.set_task_callback(request, context=context)
+
+    async def get_task_callback(
+        self,
+        request: GetTaskPushNotificationConfigParams,
+        *,
+        context: ClientCallContext | None = None,
+    ) -> TaskPushNotificationConfig:
+        """Retrieves the push notification configuration for a specific task.
+
+        Args:
+            request: The `GetTaskPushNotificationConfigParams` object specifying the task.
+            context: The client call context.
+
+        Returns:
+            A `TaskPushNotificationConfig` object containing the configuration.
+        """
+        return await self._transport.get_task_callback(request, context=context)
+
+    async def resubscribe(
+        self,
+        request: TaskIdParams,
+        *,
+        context: ClientCallContext | None = None,
+    ) -> AsyncIterator[ClientEvent]:
+        """Resubscribes to a task's event stream.
+
+        This is only available if both the client and server support streaming.
+
+        Args:
+            request: Parameters to identify the task to resubscribe to.
+            context: The client call context.
+
+        Yields:
+            An async iterator of `ClientEvent` objects.
+
+        Raises:
+            NotImplementedError: If streaming is not supported by the client or server.
+        """
+        if not self._config.streaming or not self._card.capabilities.streaming:
+            raise NotImplementedError(
+                'client and/or server do not support resubscription.'
+            )
+
+        tracker = ClientTaskManager()
+        # Note: resubscribe can only be called on an existing task. As such,
+        # we should never see Message updates, despite the typing of the service
+        # definition indicating it may be possible.
+        async for event in self._transport.resubscribe(
+            request, context=context
+        ):
+            yield await self._process_response(tracker, event)
+
+    async def get_card(
+        self, *, context: ClientCallContext | None = None
+    ) -> AgentCard:
+        """Retrieves the agent's card.
+
+        This will fetch the authenticated card if necessary and update the
+        client's internal state with the new card.
+
+        Args:
+            context: The client call context.
+
+        Returns:
+            The `AgentCard` for the agent.
+        """
+        card = await self._transport.get_card(context=context)
+        self._card = card
+        return card
+
+    async def close(self) -> None:
+        """Closes the underlying transport."""
+        await self._transport.close()