feat: Add gRPC server and client support (#162)

Introduces gRPC support at the server and client interfaces.
- Does not require changes to the AgentExecutors, simply a different
wrapper at the binary server level
- Example to follow in a2a-samples
- Will introduce spec change to enable announcement of supported
transport(s) by url for an agent in a following PR

---------

Co-authored-by: Holt Skinner <[email protected]>
Co-authored-by: Holt Skinner <[email protected]>

Author: pstephengoogle

.github/actions/spelling/allow.txt

@@ -2,20 +2,24 @@ ACard
 AClient
 AError
 AFast
+AGrpc
 ARequest
 ARun
 AServer
 AServers
+AService
 AStarlette
 EUR
 GBP
 INR
 JPY
 JSONRPCt
 Llm
+RUF
 aconnect
 adk
 agentic
+aio
 autouse
 cla
 cls
@@ -34,6 +38,7 @@ linting
 oauthoidc
 opensource
 protoc
+pyi
 pyversions
 socio
 sse

.github/actions/spelling/excludes.txt

@@ -85,7 +85,6 @@
 \.zip$
 ^\.github/actions/spelling/
 ^\.github/workflows/
-^\Qsrc/a2a/auth/__init__.py\E$
-^\Qsrc/a2a/server/request_handlers/context.py\E$
 CHANGELOG.md
 noxfile.py
+^src/a2a/grpc/

.github/linters/.jscpd.json

@@ -1,5 +1,5 @@
 {
-  "ignore": ["**/.github/**", "**/.git/**", "**/tests/**"],
+  "ignore": ["**/.github/**", "**/.git/**", "**/tests/**", "**/src/a2a/grpc/**", "**/.nox/**", "**/.venv/**"],
   "threshold": 3,
   "reporters": ["html", "markdown"]
 }

.github/linters/.ruff.toml

@@ -82,6 +82,7 @@ exclude = [
     "venv",
     "*/migrations/*",
     "noxfile.py",
+    "src/a2a/grpc/**",
 ]
 
 [lint.isort]
@@ -137,9 +138,14 @@ inline-quotes = "single"
     "SLF001",
 ]
 "types.py" = ["D", "E501", "N815"]  # Ignore docstring and annotation issues in types.py
+"proto_utils.py" = ["D102", "PLR0911"]
+"helpers.py" = ["ANN001", "ANN201", "ANN202"]
 
 [format]
-exclude = ["types.py"]
+exclude = [
+    "types.py",
+    "src/a2a/grpc/**",
+]
 docstring-code-format = true
 docstring-code-line-length = "dynamic" # Or set to 80
 quote-style = "single"

.github/workflows/linter.yaml

@@ -64,4 +64,5 @@ jobs:
           VALIDATE_GIT_COMMITLINT: false
           PYTHON_MYPY_CONFIG_FILE: .mypy.ini
           FILTER_REGEX_INCLUDE: ".*src/**/*"
+          FILTER_REGEX_EXCLUDE: ".*src/a2a/grpc/**/*"
           PYTHON_RUFF_CONFIG_FILE: .ruff.toml

buf.gen.yaml

@@ -19,9 +19,12 @@ managed:
 plugins:
   # Generate python protobuf related code
   # Generates *_pb2.py files, one for each .proto
-  - remote: buf.build/protocolbuffers/python
+  - remote: buf.build/protocolbuffers/python:v29.3
     out: src/a2a/grpc
   # Generate python service code.
   # Generates *_pb2_grpc.py
   - remote: buf.build/grpc/python
     out: src/a2a/grpc
+  # Generates *_pb2.pyi files.
+  - remote: buf.build/protocolbuffers/pyi:v29.3
+    out: src/a2a/grpc

noxfile.py

@@ -103,14 +103,17 @@ def format(session) -> None:
             }
         )
 
-        lint_paths_py = [f for f in changed_files if f.endswith('.py')]
+        lint_paths_py = [
+            f for f in changed_files if f.endswith('.py') and 'grpc/' not in f
+        ]
 
         if not lint_paths_py:
             session.log('No changed Python files to lint.')
             return
 
     session.install(
         'types-requests',
+        'types-protobuf',
         'pyupgrade',
         'autoflake',
         'ruff',

pyproject.toml

@@ -11,11 +11,15 @@ dependencies = [
     "fastapi>=0.115.12",
     "httpx>=0.28.1",
     "httpx-sse>=0.4.0",
+    "google-api-core>=1.26.0",
     "opentelemetry-api>=1.33.0",
     "opentelemetry-sdk>=1.33.0",
     "pydantic>=2.11.3",
     "sse-starlette>=2.3.3",
     "starlette>=0.46.2",
+    "grpcio>=1.60",
+    "grpcio-tools>=1.60",
+    "grpcio_reflection>=1.7.0",
 ]
 
 classifiers = [

src/a2a/client/__init__.py

@@ -6,6 +6,7 @@
     A2AClientHTTPError,
     A2AClientJSONError,
 )
+from a2a.client.grpc_client import A2AGrpcClient
 from a2a.client.helpers import create_text_message_object
 
 
@@ -15,5 +16,6 @@
     'A2AClientError',
     'A2AClientHTTPError',
     'A2AClientJSONError',
+    'A2AGrpcClient',
     'create_text_message_object',
 ]

src/a2a/client/grpc_client.py

@@ -0,0 +1,190 @@
+import logging
+
+from collections.abc import AsyncGenerator
+
+import grpc
+
+from a2a.grpc import a2a_pb2, a2a_pb2_grpc
+from a2a.types import (
+    AgentCard,
+    Message,
+    MessageSendParams,
+    Task,
+    TaskArtifactUpdateEvent,
+    TaskIdParams,
+    TaskPushNotificationConfig,
+    TaskQueryParams,
+    TaskStatusUpdateEvent,
+)
+from a2a.utils import proto_utils
+from a2a.utils.telemetry import SpanKind, trace_class
+
+
+logger = logging.getLogger(__name__)
+
+
+@trace_class(kind=SpanKind.CLIENT)
+class A2AGrpcClient:
+    """A2A Client for interacting with an A2A agent via gRPC."""
+
+    def __init__(
+        self,
+        grpc_stub: a2a_pb2_grpc.A2AServiceStub,
+        agent_card: AgentCard,
+    ):
+        """Initializes the A2AGrpcClient.
+
+        Requires an `AgentCard`
+
+        Args:
+            grpc_stub: A grpc client stub.
+            agent_card: The agent card object.
+        """
+        self.agent_card = agent_card
+        self.stub = grpc_stub
+
+    async def send_message(
+        self,
+        request: MessageSendParams,
+    ) -> Task | Message:
+        """Sends a non-streaming message request to the agent.
+
+        Args:
+            request: The `MessageSendParams` object containing the message and configuration.
+
+        Returns:
+            A `Task` or `Message` object containing the agent's response.
+        """
+        response = await self.stub.SendMessage(
+            a2a_pb2.SendMessageRequest(
+                request=proto_utils.ToProto.message(request.message),
+                configuration=proto_utils.ToProto.message_send_configuration(
+                    request.configuration
+                ),
+                metadata=proto_utils.ToProto.metadata(request.metadata),
+            )
+        )
+        if response.task:
+            return proto_utils.FromProto.task(response.task)
+        return proto_utils.FromProto.message(response.msg)
+
+    async def send_message_streaming(
+        self,
+        request: MessageSendParams,
+    ) -> AsyncGenerator[
+        Message | Task | TaskStatusUpdateEvent | TaskArtifactUpdateEvent
+    ]:
+        """Sends a streaming message request to the agent and yields responses as they arrive.
+
+        This method uses gRPC streams to receive a stream of updates from the
+        agent.
+
+        Args:
+            request: The `MessageSendParams` object containing the message and configuration.
+
+        Yields:
+            `Message` or `Task` or `TaskStatusUpdateEvent` or
+            `TaskArtifactUpdateEvent` objects as they are received in the
+            stream.
+        """
+        stream = self.stub.SendStreamingMessage(
+            a2a_pb2.SendMessageRequest(
+                request=proto_utils.ToProto.message(request.message),
+                configuration=proto_utils.ToProto.message_send_configuration(
+                    request.configuration
+                ),
+                metadata=proto_utils.ToProto.metadata(request.metadata),
+            )
+        )
+        while True:
+            response = await stream.read()
+            if response == grpc.aio.EOF:
+                break
+            if response.HasField('msg'):
+                yield proto_utils.FromProto.message(response.msg)
+            elif response.HasField('task'):
+                yield proto_utils.FromProto.task(response.task)
+            elif response.HasField('status_update'):
+                yield proto_utils.FromProto.task_status_update_event(
+                    response.status_update
+                )
+            elif response.HasField('artifact_update'):
+                yield proto_utils.FromProto.task_artifact_update_event(
+                    response.artifact_update
+                )
+
+    async def get_task(
+        self,
+        request: TaskQueryParams,
+    ) -> Task:
+        """Retrieves the current state and history of a specific task.
+
+        Args:
+            request: The `TaskQueryParams` object specifying the task ID
+
+        Returns:
+            A `Task` object containing the Task or None.
+        """
+        task = await self.stub.GetTask(
+            a2a_pb2.GetTaskRequest(name=f'tasks/{request.id}')
+        )
+        return proto_utils.FromProto.task(task)
+
+    async def cancel_task(
+        self,
+        request: TaskIdParams,
+    ) -> Task:
+        """Requests the agent to cancel a specific task.
+
+        Args:
+            request: The `TaskIdParams` object specifying the task ID.
+
+        Returns:
+            A `Task` object containing the updated Task
+        """
+        task = await self.stub.CancelTask(
+            a2a_pb2.CancelTaskRequest(name=f'tasks/{request.id}')
+        )
+        return proto_utils.FromProto.task(task)
+
+    async def set_task_callback(
+        self,
+        request: TaskPushNotificationConfig,
+    ) -> TaskPushNotificationConfig:
+        """Sets or updates the push notification configuration for a specific task.
+
+        Args:
+            request: The `TaskPushNotificationConfig` object specifying the task ID and configuration.
+
+        Returns:
+            A `TaskPushNotificationConfig` object containing the config.
+        """
+        config = await self.stub.CreateTaskPushNotification(
+            a2a_pb2.CreateTaskPushNotificationRequest(
+                parent='',
+                config_id='',
+                config=proto_utils.ToProto.task_push_notification_config(
+                    request
+                ),
+            )
+        )
+        return proto_utils.FromProto.task_push_notification_config(config)
+
+    async def get_task_callback(
+        self,
+        request: TaskIdParams,  # TODO: Update to a push id params
+    ) -> TaskPushNotificationConfig:
+        """Retrieves the push notification configuration for a specific task.
+
+        Args:
+            request: The `TaskIdParams` object specifying the task ID.
+
+        Returns:
+            A `TaskPushNotificationConfig` object containing the configuration.
+        """
+        config = await self.stub.GetTaskPushNotification(
+            a2a_pb2.GetTaskPushNotificationRequest(
+                name=f'tasks/{request.id}/pushNotification/undefined',
+            )
+        )
+        return proto_utils.FromProto.task_push_notification_config(config)