feat: Add Kafka as a transport

README.md

@@ -45,6 +45,12 @@ To install with gRPC support:
 uv add "a2a-sdk[grpc]"
 ```
 
+To install with Kafka transport support:
+
+```bash
+uv add "a2a-sdk[kafka]"
+```
+
 To install with OpenTelemetry tracing support:
 
 ```bash
@@ -87,6 +93,12 @@ To install with gRPC support:
 pip install "a2a-sdk[grpc]"
 ```
 
+To install with Kafka transport support:
+
+```bash
+pip install "a2a-sdk[kafka]"
+```
+
 To install with OpenTelemetry tracing support:
 
 ```bash

pyproject.toml

@@ -36,6 +36,7 @@ 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"]
+kafka = ["aiokafka>=0.11.0"]
 telemetry = ["opentelemetry-api>=1.33.0", "opentelemetry-sdk>=1.33.0"]
 
 [project.urls]

src/a2a/client/client_factory.py

@@ -1,3 +1,3 @@
 from __future__ import annotations
 
 import logging
@@ -25,6 +25,11 @@
 except ImportError:
     GrpcTransport = None  # type: ignore # pyright: ignore
 
+try:
+    from a2a.client.transports.kafka import KafkaClientTransport
+except ImportError:
+    KafkaClientTransport = None  # type: ignore # pyright: ignore
+
 
 logger = logging.getLogger(__name__)
 
@@ -97,6 +102,32 @@
                 TransportProtocol.grpc,
                 GrpcTransport.create,
             )
+        if TransportProtocol.kafka in supported:
+            if KafkaClientTransport is None:
+                raise ImportError(
+                    'To use KafkaClient, its dependencies must be installed. '
+                    'You can install them with \'pip install "a2a-sdk[kafka]"\''
+                )
+            self.register(
+                TransportProtocol.kafka,
+                self._create_kafka_transport,
+            )
+
+    def _create_kafka_transport(
+        self,
+        card: AgentCard,
+        url: str,
+        config: ClientConfig,
+        interceptors: list[ClientCallInterceptor],
+    ) -> ClientTransport:
+        """Create a Kafka transport that will auto-start when first used."""
+        # Create the transport using the existing create method
+        transport = KafkaClientTransport.create(card, url, config, interceptors)
+
+        # Mark the transport for auto-start when first used
+        transport._auto_start = True
+
+        return transport
 
     def register(self, label: str, generator: TransportProducer) -> None:
         """Register a new transport producer for a given transport label."""

src/a2a/client/transports/__init__.py

@@ -10,10 +10,16 @@
 except ImportError:
     GrpcTransport = None  # type: ignore
 
+try:
+    from a2a.client.transports.kafka import KafkaClientTransport
+except ImportError:
+    KafkaClientTransport = None  # type: ignore
+
 
 __all__ = [
     'ClientTransport',
     'GrpcTransport',
     'JsonRpcTransport',
+    'KafkaClientTransport',
     'RestTransport',
 ]