wip

Author: guglielmo-san

samples/cli.py

@@ -10,9 +10,9 @@
 import httpx
 
 from a2a.client import A2ACardResolver, ClientConfig, create_client
-from a2a.types import Message, Part, Role, SendMessageRequest, TaskState
 from a2a.helpers import get_artifact_text, get_message_text
 from a2a.helpers.agent_card import display_agent_card
+from a2a.types import Message, Part, Role, SendMessageRequest, TaskState
 
 
 async def _handle_stream(

src/a2a/helpers/__init__.py

@@ -1,7 +1,7 @@
 """Helper functions for the A2A Python SDK."""
 
 from a2a.helpers.agent_card import display_agent_card
-from a2a.helpers.types import (
+from a2a.helpers.proto_helpers import (
     get_artifact_text,
     get_message_text,
     get_stream_response_text,
@@ -16,6 +16,7 @@
     new_text_status_update_event,
 )
 
+
 __all__ = [
     'display_agent_card',
     'get_artifact_text',

src/a2a/helpers/types.py src/a2a/helpers/proto_helpers.pysrc/a2a/helpers/types.py renamed to src/a2a/helpers/proto_helpers.py

@@ -1,10 +1,8 @@
 """Unified helper functions for creating and handling A2A types."""
 
 import uuid
-from collections.abc import Sequence
-from typing import Any
 
-from google.protobuf.json_format import MessageToDict
+from collections.abc import Sequence
 
 from a2a.types.a2a_pb2 import (
     Artifact,
@@ -121,7 +119,6 @@ def new_task(
     history: list[Message] | None = None,
 ) -> Task:
     """Creates a Task object with a specified status."""
-
     if history is None:
         history = []
     if artifacts is None:
@@ -169,7 +166,7 @@ def new_text_status_update_event(
     )
 
 
-def new_text_artifact_update_event(
+def new_text_artifact_update_event(  # noqa: PLR0913
     task_id: str,
     context_id: str,
     name: str,
@@ -189,17 +186,23 @@ def new_text_artifact_update_event(
     )
 
 
-def get_stream_response_text(response: StreamResponse, delimiter: str = '\n') -> str:
+def get_stream_response_text(
+    response: StreamResponse, delimiter: str = '\n'
+) -> str:
     """Extracts text content from a StreamResponse."""
     if response.HasField('message'):
         return get_message_text(response.message, delimiter)
-    elif response.HasField('task'):
-        texts = [get_artifact_text(a, delimiter) for a in response.task.artifacts]
+    if response.HasField('task'):
+        texts = [
+            get_artifact_text(a, delimiter) for a in response.task.artifacts
+        ]
         return delimiter.join(t for t in texts if t)
-    elif response.HasField('status_update'):
+    if response.HasField('status_update'):
         if response.status_update.status.HasField('message'):
-            return get_message_text(response.status_update.status.message, delimiter)
+            return get_message_text(
+                response.status_update.status.message, delimiter
+            )
         return ''
-    elif response.HasField('artifact_update'):
+    if response.HasField('artifact_update'):
         return get_artifact_text(response.artifact_update.artifact, delimiter)
     return ''

src/a2a/server/agent_execution/context.py

@@ -1,5 +1,6 @@
 from typing import Any
 
+from a2a.helpers.proto_helpers import get_message_text
 from a2a.server.context import ServerCallContext
 from a2a.server.id_generator import (
     IDGenerator,
@@ -12,7 +13,6 @@
     SendMessageRequest,
     Task,
 )
-from a2a.helpers.types import get_message_text
 from a2a.utils.errors import InvalidParamsError
 
 

src/a2a/utils/__init__.py

@@ -6,7 +6,6 @@
     DEFAULT_RPC_URL,
     TransportProtocol,
 )
-
 from a2a.utils.proto_utils import to_stream_response
 
 

tests/client/test_client_helpers.py

@@ -4,7 +4,7 @@
 from google.protobuf.json_format import MessageToDict
 
 from a2a.client.helpers import parse_agent_card
-from a2a.helpers.types import new_text_message
+from a2a.helpers.proto_helpers import new_text_message
 from a2a.server.request_handlers.response_helpers import agent_card_to_dict
 from a2a.types.a2a_pb2 import (
     APIKeySecurityScheme,

tests/client/transports/test_grpc_client.py

@@ -35,7 +35,7 @@
     TaskStatus,
     TaskStatusUpdateEvent,
 )
-from a2a.helpers.types import get_text_parts
+from a2a.helpers.proto_helpers import get_text_parts
 
 
 @pytest.fixture

tests/client/transports/test_rest_client.py

@@ -8,7 +8,7 @@
 from google.protobuf.timestamp_pb2 import Timestamp
 from httpx_sse import EventSource, ServerSentEvent
 
-from a2a.helpers.types import new_text_message
+from a2a.helpers.proto_helpers import new_text_message
 from a2a.client.client import ClientCallContext
 from a2a.client.errors import A2AClientError
 from a2a.client.transports.rest import RestTransport
@@ -120,9 +120,7 @@ async def test_rest_mapped_errors(
             agent_card=mock_agent_card,
             url='http://agent.example.com/api',
         )
-        params = SendMessageRequest(
-            message=new_text_message(text='Hello')
-        )
+        params = SendMessageRequest(message=new_text_message(text='Hello'))
 
         mock_build_request = MagicMock(
             return_value=AsyncMock(spec=httpx.Request)
@@ -172,9 +170,7 @@ async def test_send_message_with_timeout_context(
             agent_card=mock_agent_card,
             url='http://agent.example.com/api',
         )
-        params = SendMessageRequest(
-            message=new_text_message(text='Hello')
-        )
+        params = SendMessageRequest(message=new_text_message(text='Hello'))
         context = ClientCallContext(timeout=10.0)
 
         mock_build_request = MagicMock(
@@ -246,9 +242,7 @@ async def test_send_message_with_default_extensions(
             agent_card=mock_agent_card,
             url='http://agent.example.com/api',
         )
-        params = SendMessageRequest(
-            message=new_text_message(text='Hello')
-        )
+        params = SendMessageRequest(message=new_text_message(text='Hello'))
 
         # Mock the build_request method to capture its inputs
         mock_build_request = MagicMock(

tests/e2e/push_notifications/agent_app.py

@@ -24,7 +24,7 @@
     Message,
     Task,
 )
-from a2a.helpers.types import (
+from a2a.helpers.proto_helpers import (
     new_text_message,
     new_task_from_user_message,
 )
@@ -90,9 +90,7 @@ async def invoke(
         # Flow with user input required: "How are you?" -> "Good! How are you?" -> "Good" -> "Amazing".
         elif text_message == 'How are you?':
             await updater.requires_input(
-                new_text_message(
-                    'Good! How are you?', task.context_id, task.id
-                )
+                new_text_message('Good! How are you?', task.context_id, task.id)
             )
         elif text_message == 'Good':
             await updater.complete(