refactor: unify transport constants usage

Fixes #705

Author: ishymko

src/a2a/client/client_factory.py

@@ -20,9 +20,7 @@
     AgentInterface,
 )
 from a2a.utils.constants import (
-    TRANSPORT_GRPC,
-    TRANSPORT_HTTP_JSON,
-    TRANSPORT_JSONRPC,
+    TransportProtocol,
 )
 
 
@@ -74,9 +72,9 @@ def __init__(
 
     def _register_defaults(self, supported: list[str]) -> None:
         # Empty support list implies JSON-RPC only.
-        if TRANSPORT_JSONRPC in supported or not supported:
+        if TransportProtocol.JSONRPC in supported or not supported:
             self.register(
-                TRANSPORT_JSONRPC,
+                TransportProtocol.JSONRPC,
                 lambda card, url, config, interceptors: JsonRpcTransport(
                     config.httpx_client or httpx.AsyncClient(),
                     card,
@@ -85,9 +83,9 @@ def _register_defaults(self, supported: list[str]) -> None:
                     config.extensions or None,
                 ),
             )
-        if TRANSPORT_HTTP_JSON in supported:
+        if TransportProtocol.HTTP_JSON in supported:
             self.register(
-                TRANSPORT_HTTP_JSON,
+                TransportProtocol.HTTP_JSON,
                 lambda card, url, config, interceptors: RestTransport(
                     config.httpx_client or httpx.AsyncClient(),
                     card,
@@ -96,14 +94,14 @@ def _register_defaults(self, supported: list[str]) -> None:
                     config.extensions or None,
                 ),
             )
-        if TRANSPORT_GRPC in supported:
+        if TransportProtocol.GRPC in supported:
             if GrpcTransport is None:
                 raise ImportError(
                     'To use GrpcClient, its dependencies must be installed. '
                     'You can install them with \'pip install "a2a-sdk[grpc]"\''
                 )
             self.register(
-                TRANSPORT_GRPC,
+                TransportProtocol.GRPC,
                 GrpcTransport.create,
             )
 
@@ -207,7 +205,7 @@ def create(
           server configuration, a `ValueError` is raised.
         """
         client_set = self._config.supported_protocol_bindings or [
-            TRANSPORT_JSONRPC
+            TransportProtocol.JSONRPC
         ]
         transport_protocol = None
         transport_url = None

src/a2a/client/transports/rest.py

@@ -35,8 +35,7 @@
     TaskPushNotificationConfig,
 )
 from a2a.utils.constants import (
-    TRANSPORT_HTTP_JSON,
-    TRANSPORT_JSONRPC,
+    TransportProtocol,
 )
 from a2a.utils.telemetry import SpanKind, trace_class
 
@@ -62,15 +61,15 @@ def __init__(
         elif agent_card:
             for interface in agent_card.supported_interfaces:
                 if interface.protocol_binding in (
-                    TRANSPORT_HTTP_JSON,
-                    TRANSPORT_JSONRPC,
+                    TransportProtocol.HTTP_JSON,
+                    TransportProtocol.JSONRPC,
                 ):
                     self.url = interface.url
                     break
             else:
                 raise ValueError(
-                    f'AgentCard does not support {TRANSPORT_HTTP_JSON} '
-                    f'or {TRANSPORT_JSONRPC}'
+                    f'AgentCard does not support {TransportProtocol.HTTP_JSON} '
+                    f'or {TransportProtocol.JSONRPC}'
                 )
         else:
             raise ValueError('Must provide either agent_card or url')

src/a2a/utils/__init__.py

@@ -12,9 +12,6 @@
     DEFAULT_RPC_URL,
     EXTENDED_AGENT_CARD_PATH,
     PREV_AGENT_CARD_WELL_KNOWN_PATH,
-    TRANSPORT_GRPC,
-    TRANSPORT_HTTP_JSON,
-    TRANSPORT_JSONRPC,
     TransportProtocol,
 )
 from a2a.utils.helpers import (
@@ -45,9 +42,6 @@
     'DEFAULT_RPC_URL',
     'EXTENDED_AGENT_CARD_PATH',
     'PREV_AGENT_CARD_WELL_KNOWN_PATH',
-    'TRANSPORT_GRPC',
-    'TRANSPORT_HTTP_JSON',
-    'TRANSPORT_JSONRPC',
     'TransportProtocol',
     'append_artifact_to_task',
     'are_modalities_compatible',

src/a2a/utils/constants.py

@@ -1,5 +1,8 @@
 """Constants for well-known URIs used throughout the A2A Python SDK."""
 
+from enum import Enum
+
+
 AGENT_CARD_WELL_KNOWN_PATH = '/.well-known/agent-card.json'
 PREV_AGENT_CARD_WELL_KNOWN_PATH = '/.well-known/agent.json'
 EXTENDED_AGENT_CARD_PATH = '/agent/authenticatedExtendedCard'
@@ -8,19 +11,12 @@
 """Default page size for the `tasks/list` method."""
 
 
-# Transport protocol constants
-# These match the protocol binding values used in AgentCard
-TRANSPORT_JSONRPC = 'JSONRPC'
-TRANSPORT_HTTP_JSON = 'HTTP+JSON'
-TRANSPORT_GRPC = 'GRPC'
-
-
-class TransportProtocol:
+class TransportProtocol(str, Enum):
     """Transport protocol string constants."""
 
-    jsonrpc = TRANSPORT_JSONRPC
-    http_json = TRANSPORT_HTTP_JSON
-    grpc = TRANSPORT_GRPC
+    JSONRPC = 'JSONRPC'
+    HTTP_JSON = 'HTTP+JSON'
+    GRPC = 'GRPC'
 
 
 DEFAULT_MAX_CONTENT_LENGTH = 10 * 1024 * 1024  # 10MB

tests/client/test_auth_middleware.py

@@ -178,7 +178,7 @@ async def test_client_with_simple_interceptor() -> None:
     interceptor = HeaderInterceptor('X-Test-Header', 'Test-Value-123')
     card = AgentCard(
         supported_interfaces=[
-            AgentInterface(url=url, protocol_binding=TransportProtocol.jsonrpc)
+            AgentInterface(url=url, protocol_binding=TransportProtocol.JSONRPC)
         ],
         name='testbot',
         description='test bot',
@@ -192,7 +192,7 @@ async def test_client_with_simple_interceptor() -> None:
     async with httpx.AsyncClient() as http_client:
         config = ClientConfig(
             httpx_client=http_client,
-            supported_protocol_bindings=[TransportProtocol.jsonrpc],
+            supported_protocol_bindings=[TransportProtocol.JSONRPC],
         )
         factory = ClientFactory(config)
         client = factory.create(card, interceptors=[interceptor])
@@ -310,7 +310,7 @@ async def test_auth_interceptor_variants(
     agent_card = AgentCard(
         supported_interfaces=[
             AgentInterface(
-                url=test_case.url, protocol_binding=TransportProtocol.jsonrpc
+                url=test_case.url, protocol_binding=TransportProtocol.JSONRPC
             )
         ],
         name=f'{test_case.scheme_name}bot',
@@ -333,7 +333,7 @@ async def test_auth_interceptor_variants(
     async with httpx.AsyncClient() as http_client:
         config = ClientConfig(
             httpx_client=http_client,
-            supported_protocol_bindings=[TransportProtocol.jsonrpc],
+            supported_protocol_bindings=[TransportProtocol.JSONRPC],
         )
         factory = ClientFactory(config)
         client = factory.create(agent_card, interceptors=[auth_interceptor])
@@ -362,7 +362,7 @@ async def test_auth_interceptor_skips_when_scheme_not_in_security_schemes(
         supported_interfaces=[
             AgentInterface(
                 url='http://agent.com/rpc',
-                protocol_binding=TransportProtocol.jsonrpc,
+                protocol_binding=TransportProtocol.JSONRPC,
             )
         ],
         name='missingbot',

tests/client/test_client_factory.py

@@ -25,7 +25,7 @@ def base_agent_card() -> AgentCard:
         description='An agent for testing.',
         supported_interfaces=[
             AgentInterface(
-                protocol_binding=TransportProtocol.jsonrpc,
+                protocol_binding=TransportProtocol.JSONRPC,
                 url='http://primary-url.com',
             )
         ],
@@ -42,8 +42,8 @@ def test_client_factory_selects_preferred_transport(base_agent_card: AgentCard):
     config = ClientConfig(
         httpx_client=httpx.AsyncClient(),
         supported_protocol_bindings=[
-            TransportProtocol.jsonrpc,
-            TransportProtocol.http_json,
+            TransportProtocol.JSONRPC,
+            TransportProtocol.HTTP_JSON,
         ],
         extensions=['https://example.com/test-ext/v0'],
     )
@@ -61,16 +61,16 @@ def test_client_factory_selects_secondary_transport_url(
     """Verify that the factory selects the correct URL for a secondary transport."""
     base_agent_card.supported_interfaces.append(
         AgentInterface(
-            protocol_binding=TransportProtocol.http_json,
+            protocol_binding=TransportProtocol.HTTP_JSON,
             url='http://secondary-url.com',
         )
     )
     # Client prefers REST, which is available as a secondary transport
     config = ClientConfig(
         httpx_client=httpx.AsyncClient(),
         supported_protocol_bindings=[
-            TransportProtocol.http_json,
-            TransportProtocol.jsonrpc,
+            TransportProtocol.HTTP_JSON,
+            TransportProtocol.JSONRPC,
         ],
         use_client_preference=True,
         extensions=['https://example.com/test-ext/v0'],
@@ -89,22 +89,22 @@ def test_client_factory_server_preference(base_agent_card: AgentCard):
     base_agent_card.supported_interfaces.insert(
         0,
         AgentInterface(
-            protocol_binding=TransportProtocol.http_json,
+            protocol_binding=TransportProtocol.HTTP_JSON,
             url='http://primary-url.com',
         ),
     )
     base_agent_card.supported_interfaces.append(
         AgentInterface(
-            protocol_binding=TransportProtocol.jsonrpc,
+            protocol_binding=TransportProtocol.JSONRPC,
             url='http://secondary-url.com',
         )
     )
     # Client supports both, but server prefers REST
     config = ClientConfig(
         httpx_client=httpx.AsyncClient(),
         supported_protocol_bindings=[
-            TransportProtocol.jsonrpc,
-            TransportProtocol.http_json,
+            TransportProtocol.JSONRPC,
+            TransportProtocol.HTTP_JSON,
         ],
     )
     factory = ClientFactory(config)
@@ -258,7 +258,8 @@ def custom_transport_producer(*args, **kwargs):
         base_agent_card,
         client_config=config,
         extra_transports=typing.cast(
-            dict[str, TransportProducer], {'custom': custom_transport_producer}
+            'dict[str, TransportProducer]',
+            {'custom': custom_transport_producer},
         ),
     )
 

tests/client/transports/test_rest_client.py

@@ -3,8 +3,8 @@
 
 import httpx
 import pytest
-from google.protobuf import json_format
 
+from google.protobuf import json_format
 from httpx_sse import EventSource, ServerSentEvent
 
 from a2a.client import create_text_message_object
@@ -15,10 +15,9 @@
     AgentCapabilities,
     AgentCard,
     AgentInterface,
-    Role,
     SendMessageRequest,
 )
-from a2a.utils.constants import TRANSPORT_HTTP_JSON
+from a2a.utils.constants import TransportProtocol
 
 
 @pytest.fixture
@@ -31,7 +30,7 @@ def mock_agent_card() -> MagicMock:
     mock = MagicMock(spec=AgentCard, url='http://agent.example.com/api')
     mock.supported_interfaces = [
         AgentInterface(
-            protocol_binding=TRANSPORT_HTTP_JSON,
+            protocol_binding=TransportProtocol.HTTP_JSON,
             url='http://agent.example.com/api',
         )
     ]
@@ -276,7 +275,7 @@ async def test_get_card_with_extended_card_support_with_extensions(
             capabilities=AgentCapabilities(extended_agent_card=True),
         )
         interface = agent_card.supported_interfaces.add()
-        interface.protocol_binding = TRANSPORT_HTTP_JSON
+        interface.protocol_binding = TransportProtocol.HTTP_JSON
         interface.url = 'http://agent.example.com/api'
 
         client = RestTransport(

tests/e2e/push_notifications/test_default_push_notification_support.py

@@ -105,7 +105,7 @@ async def test_notification_triggering_with_in_message_config_e2e(
     token = uuid.uuid4().hex
     a2a_client = ClientFactory(
         ClientConfig(
-            supported_protocol_bindings=[TransportProtocol.http_json],
+            supported_protocol_bindings=[TransportProtocol.HTTP_JSON],
             push_notification_configs=[
                 PushNotificationConfig(
                     id='in-message-config',
@@ -114,7 +114,7 @@ async def test_notification_triggering_with_in_message_config_e2e(
                 )
             ],
         )
-    ).create(minimal_agent_card(agent_server, [TransportProtocol.http_json]))
+    ).create(minimal_agent_card(agent_server, [TransportProtocol.HTTP_JSON]))
 
     # Send a message and extract the returned task.
     responses = [
@@ -157,9 +157,9 @@ async def test_notification_triggering_after_config_change_e2e(
     # Configure an A2A client without a push notification config.
     a2a_client = ClientFactory(
         ClientConfig(
-            supported_protocol_bindings=[TransportProtocol.http_json],
+            supported_protocol_bindings=[TransportProtocol.HTTP_JSON],
         )
-    ).create(minimal_agent_card(agent_server, [TransportProtocol.http_json]))
+    ).create(minimal_agent_card(agent_server, [TransportProtocol.HTTP_JSON]))
 
     # Send a message and extract the returned task.
     responses = [

tests/integration/test_client_server_integration.py

@@ -19,11 +19,7 @@
 from a2a.types import a2a_pb2_grpc
 from a2a.server.apps import A2AFastAPIApplication, A2ARESTFastAPIApplication
 from a2a.server.request_handlers import GrpcHandler, RequestHandler
-from a2a.utils.constants import (
-    TRANSPORT_HTTP_JSON,
-    TRANSPORT_GRPC,
-    TRANSPORT_JSONRPC,
-)
+from a2a.utils.constants import TransportProtocol
 from a2a.utils.signing import (
     create_agent_card_signer,
     create_signature_verifier,
@@ -156,10 +152,10 @@ def agent_card() -> AgentCard:
         default_output_modes=['text/plain'],
         supported_interfaces=[
             AgentInterface(
-                protocol_binding=TRANSPORT_HTTP_JSON,
+                protocol_binding=TransportProtocol.HTTP_JSON,
                 url='http://testserver',
             ),
-            AgentInterface(protocol_binding='grpc', url='localhost:50051'),
+            AgentInterface(protocol_binding=TransportProtocol.GRPC, url='localhost:50051'),
         ],
     )
 

tests/integration/test_end_to_end.py

@@ -33,7 +33,7 @@
     TaskState,
     a2a_pb2_grpc,
 )
-from a2a.utils import TRANSPORT_GRPC, TRANSPORT_HTTP_JSON, TRANSPORT_JSONRPC
+from a2a.utils import TransportProtocol
 
 
 class MockAgentExecutor(AgentExecutor):
@@ -68,15 +68,15 @@ def agent_card() -> AgentCard:
         default_output_modes=['text/plain'],
         supported_interfaces=[
             AgentInterface(
-                protocol_binding=TRANSPORT_HTTP_JSON,
+                protocol_binding=TransportProtocol.HTTP_JSON,
                 url='http://testserver',
             ),
             AgentInterface(
-                protocol_binding=TRANSPORT_JSONRPC,
+                protocol_binding=TransportProtocol.JSONRPC,
                 url='http://testserver',
             ),
             AgentInterface(
-                protocol_binding=TRANSPORT_GRPC,
+                protocol_binding=TransportProtocol.GRPC,
                 url='localhost:50051',
             ),
         ],
@@ -149,7 +149,7 @@ async def grpc_setup(
 
     # Update the gRPC interface dynamically based on the assigned port
     for interface in grpc_agent_card.supported_interfaces:
-        if interface.protocol_binding == TRANSPORT_GRPC:
+        if interface.protocol_binding == TransportProtocol.GRPC:
             interface.url = server_address
             break
     else: