Refactor to use TransportProtocol class from types.py

Author: holtskinner

src/a2a/client/client.py

@@ -40,6 +40,7 @@ class Channel:  # type: ignore[no-redef]
     TaskPushNotificationConfig,
     TaskQueryParams,
     TaskStatusUpdateEvent,
+    TransportProtocol,
 )
 from a2a.utils.constants import AGENT_CARD_WELL_KNOWN_PATH
 
@@ -153,11 +154,13 @@ class ClientConfig:
     grpc_channel_factory: Callable[[str], Channel] | None = None
     """Generates a grpc connection channel for a given url."""
 
-    supported_transports: list[str] = dataclasses.field(default_factory=list)
+    supported_transports: list[TransportProtocol | str] = dataclasses.field(
+        default_factory=list
+    )
     """Ordered list of transports for connecting to agent
        (in order of preference). Empty implies JSONRPC only.
 
-       This is a string type and not a Transports enum type to allow custom
+       This is a string type to allow custom
        transports to exist in closed ecosystems.
     """
 

src/a2a/client/client_factory.py

@@ -9,8 +9,12 @@
 from a2a.client.jsonrpc_client import NewJsonRpcClient
 from a2a.client.middleware import ClientCallInterceptor
 from a2a.client.rest_client import NewRestfulClient
-from a2a.types import AgentCapabilities, AgentCard, AgentInterface
-from a2a.utils import Transports
+from a2a.types import (
+    AgentCapabilities,
+    AgentCard,
+    AgentInterface,
+    TransportProtocol,
+)
 
 
 logger = logging.getLogger(__name__)
@@ -54,12 +58,12 @@ def __init__(
         self._registry: dict[str, ClientProducer] = {}
         # By default register the 3 core transports if in the config.
         # Can be overridden with custom clients via the register method.
-        if Transports.JSONRPC in self._config.supported_transports:
-            self._registry[Transports.JSONRPC] = NewJsonRpcClient
-        if Transports.RESTful in self._config.supported_transports:
-            self._registry[Transports.RESTful] = NewRestfulClient
-        if Transports.GRPC in self._config.supported_transports:
-            self._registry[Transports.GRPC] = NewGrpcClient
+        if TransportProtocol.jsonrpc in self._config.supported_transports:
+            self._registry[TransportProtocol.jsonrpc] = NewJsonRpcClient
+        if TransportProtocol.http_json in self._config.supported_transports:
+            self._registry[TransportProtocol.http_json] = NewRestfulClient
+        if TransportProtocol.grpc in self._config.supported_transports:
+            self._registry[TransportProtocol.grpc] = NewGrpcClient
 
     def register(self, label: str, generator: ClientProducer) -> None:
         """Register a new client producer for a given transport label."""
@@ -88,10 +92,12 @@ def create(
           server configuration, a `ValueError` is raised.
         """
         # Determine preferential transport
-        server_set = [card.preferred_transport or 'JSONRPC']
+        server_set = [card.preferred_transport or TransportProtocol.jsonrpc]
         if card.additional_interfaces:
             server_set.extend([x.transport for x in card.additional_interfaces])
-        client_set = self._config.supported_transports or ['JSONRPC']
+        client_set = self._config.supported_transports or [
+            TransportProtocol.jsonrpc
+        ]
         transport = None
         # Two options, use the client ordering or the server ordering.
         if self._config.use_client_preference:

src/a2a/utils/__init__.py

@@ -28,14 +28,12 @@
     completed_task,
     new_task,
 )
-from a2a.utils.transports import Transports
 
 
 __all__ = [
     'AGENT_CARD_WELL_KNOWN_PATH',
     'DEFAULT_RPC_URL',
     'EXTENDED_AGENT_CARD_PATH',
-    'Transports',
     'append_artifact_to_task',
     'are_modalities_compatible',
     'build_text_artifact',