feat: add a TextClient class for a simplified text-based communication

samples/cli.py

@@ -11,25 +11,22 @@
 
 from a2a.client import A2ACardResolver, ClientConfig, create_client
 from a2a.types import Message, Part, Role, SendMessageRequest, TaskState
+from a2a.utils.message import get_message_text
 
 
 async def _handle_stream(
     stream: Any, current_task_id: str | None
 ) -> str | None:
-    async for event, task in stream:
-        if not task:
-            continue
+    async for event in stream:
         if not current_task_id:
-            current_task_id = task.id
-
+            current_task_id = event.task.id
         if event:
             if event.HasField('status_update'):
                 state_name = TaskState.Name(event.status_update.status.state)
                 print(f'TaskStatusUpdate [state={state_name}]:', end=' ')
                 if event.status_update.status.HasField('message'):
-                    for part in event.status_update.status.message.parts:
-                        if part.text:
-                            print(part.text, end=' ')
+                    message = event.status_update.status.message
+                    print(get_message_text(message, delimiter=' '))
                 print()
 
                 if (

samples/text_client_cli.py

@@ -0,0 +1,51 @@
+import argparse
+import asyncio
+
+import grpc
+import httpx
+
+from a2a.client import A2ACardResolver, create_text_client
+
+
+async def main() -> None:
+    """Run the simple A2A terminal client using TextClient."""
+    parser = argparse.ArgumentParser(description='A2A Simple Text Client')
+    parser.add_argument(
+        '--url', default='http://127.0.0.1:41241', help='Agent base URL'
+    )
+    args = parser.parse_args()
+
+    print(f'Connecting to {args.url}')
+
+    async with httpx.AsyncClient() as httpx_client:
+        resolver = A2ACardResolver(httpx_client, args.url)
+        card = await resolver.get_agent_card()
+        print(f'\n✓ Agent Card Found: {card.name}')
+
+    text_client = await create_text_client(card)
+
+    print('\nConnected! Send a message or type /quit to exit.')
+
+    while True:
+        try:
+            loop = asyncio.get_running_loop()
+            user_input = await loop.run_in_executor(None, input, 'You: ')
+        except KeyboardInterrupt:
+            break
+
+        if user_input.lower() in ('/quit', '/exit'):
+            break
+        if not user_input.strip():
+            continue
+
+        try:
+            response = await text_client.send_text_message(user_input)
+            print(f'Agent: {response}')
+        except (httpx.RequestError, grpc.RpcError) as e:
+            print(f'Error communicating with agent: {e}')
+
+    await text_client.close()
+
+
+if __name__ == '__main__':
+    asyncio.run(main())

src/a2a/client/__init__.py

@@ -15,6 +15,7 @@
 from a2a.client.client_factory import (
     ClientFactory,
     create_client,
+    create_text_client,
     minimal_agent_card,
 )
 from a2a.client.errors import (
@@ -24,6 +25,7 @@
 )
 from a2a.client.helpers import create_text_message_object
 from a2a.client.interceptors import ClientCallInterceptor
+from a2a.client.text_client import TextClient
 
 
 __all__ = [
@@ -40,7 +42,9 @@
     'ClientFactory',
     'CredentialService',
     'InMemoryContextCredentialStore',
+    'TextClient',
     'create_client',
+    'create_text_client',
     'create_text_message_object',
     'minimal_agent_card',
 ]

src/a2a/client/client_factory.py

@@ -12,6 +12,7 @@
 from a2a.client.base_client import BaseClient
 from a2a.client.card_resolver import A2ACardResolver
 from a2a.client.client import Client, ClientConfig
+from a2a.client.text_client import TextClient
 from a2a.client.transports.base import ClientTransport
 from a2a.client.transports.jsonrpc import JsonRpcTransport
 from a2a.client.transports.rest import RestTransport
@@ -406,6 +407,47 @@
     return factory.create(agent, interceptors)
 
 
+async def create_text_client(  # noqa: PLR0913
+    agent: str | AgentCard,
+    client_config: ClientConfig | None = None,
+    interceptors: list[ClientCallInterceptor] | None = None,
+    relative_card_path: str | None = None,
+    resolver_http_kwargs: dict[str, Any] | None = None,
+    signature_verifier: Callable[[AgentCard], None] | None = None,
+) -> TextClient:
+    """Create a `TextClient` for an agent from a URL or `AgentCard`.
+
+    Convenience function that constructs a `ClientFactory` internally.
+    For reusing a factory across multiple agents or registering custom
+    transports, use `ClientFactory` directly instead.
+
+    Args:
+      agent: The base URL of the agent, or an `AgentCard` to use
+        directly.
+      client_config: Optional `ClientConfig`. A default config is
+        created if not provided.
+      interceptors: A list of interceptors to use for each request.
+      relative_card_path: The relative path when resolving the agent
+        card. Only used when `agent` is a URL.
+      resolver_http_kwargs: Dictionary of arguments to provide to the
+        httpx client when resolving the agent card.
+      signature_verifier: A callable used to verify the agent card's
+        signatures.
+
+    Returns:
+      A `TextClient` wrapping the constructed `Client`.
+    """
+    client = await create_client(
+        agent=agent,
+        client_config=client_config,
+        interceptors=interceptors,
+        relative_card_path=relative_card_path,
+        resolver_http_kwargs=resolver_http_kwargs,
+        signature_verifier=signature_verifier,
+    )
+    return TextClient(client)
+
+
 def minimal_agent_card(
     url: str, transports: list[str] | None = None
 ) -> AgentCard:

src/a2a/client/text_client.py

@@ -0,0 +1,64 @@
+import uuid
+
+from a2a.client.client import Client, ClientCallContext
+from a2a.types import Message, Part, Role, SendMessageRequest
+
+
+class TextClient:
+    """A facade around Client that simplifies text-based communication.
+
+    Wraps an underlying Client instance and exposes a simplified interface
+    for sending plain-text messages and receiving aggregated text responses.
+    For full Client API access, use the underlying client directly via
+    the `client` property.
+    """
+
+    def __init__(self, client: Client):
+        self._client = client
+
+    @property
+    def client(self) -> Client:
+        """Returns the underlying Client instance for full API access."""
+        return self._client
+
+    async def send_text_message(
+        self,
+        text: str,
+        *,
+        context: ClientCallContext | None = None,
+    ) -> str:
+        """Sends a text message and returns the aggregated text response."""
+        request = SendMessageRequest(
+            message=Message(
+                role=Role.ROLE_USER,
+                message_id=str(uuid.uuid4()),
+                parts=[Part(text=text)],
+            )
+        )
+
+        response_parts: list[str] = []
+
+        async for event in self._client.send_message(request, context=context):
+            if event.HasField('message'):
+                response_parts.extend(
+                    part.text for part in event.message.parts if part.text
+                )
+            elif event.HasField('status_update'):
+                if event.status_update.status.HasField('message'):
+                    response_parts.extend(
+                        part.text
+                        for part in event.status_update.status.message.parts
+                        if part.text
+                    )
+            elif event.HasField('artifact_update'):
+                response_parts.extend(
+                    part.text
+                    for part in event.artifact_update.artifact.parts
+                    if part.text
+                )
+
+        return ' '.join(response_parts)
+
+    async def close(self) -> None:
+        """Closes the underlying client."""
+        await self._client.close()

tests/client/test_text_client.py

@@ -0,0 +1,111 @@
+from unittest.mock import AsyncMock
+
+import pytest
+
+from a2a.client import (
+    Client,
+    ClientConfig,
+    ClientCallContext,
+    create_text_client,
+    minimal_agent_card,
+    TextClient,
+)
+from a2a.types import Part, StreamResponse
+
+
+@pytest.fixture
+def mock_client() -> AsyncMock:
+    return AsyncMock(spec=Client)
+
+
+@pytest.fixture
+def text_client(mock_client: AsyncMock) -> TextClient:
+    return TextClient(mock_client)
+
+
+def test_client_property(
+    text_client: TextClient, mock_client: AsyncMock
+) -> None:
+    assert text_client.client is mock_client
+
+
+@pytest.mark.asyncio
+async def test_create_client_and_wrap() -> None:
+    # Create a minimal card
+    card = minimal_agent_card(url='http://test.com', transports=['JSONRPC'])
+
+    config = ClientConfig(supported_protocol_bindings=['JSONRPC'])
+
+    text_client = await create_text_client(card, client_config=config)
+
+    assert isinstance(text_client, TextClient)
+    assert isinstance(text_client.client, Client)
+
+    # Clean up
+    await text_client.close()
+
+
+@pytest.mark.asyncio
+async def test_send_text_message(
+    text_client: TextClient, mock_client: AsyncMock
+) -> None:
+    async def create_stream(*args, **kwargs):
+        # Event 0: task (ignored)
+        resp0 = StreamResponse()
+        resp0.task.id = 'task-1'
+        yield resp0
+
+        # Event 1: direct message
+        resp1 = StreamResponse()
+        resp1.message.parts.append(Part(text='Hello'))
+        yield resp1
+
+        # Event 2: status update without message
+        resp2 = StreamResponse()
+        resp2.status_update.status.state = 1
+        yield resp2
+
+        # Event 3: status update with message
+        resp3 = StreamResponse()
+        resp3.status_update.status.message.parts.append(Part(text='Processing'))
+        yield resp3
+
+        # Event 4: artifact update
+        resp4 = StreamResponse()
+        resp4.artifact_update.artifact.parts.append(Part(text='World!'))
+        yield resp4
+
+    mock_client.send_message.return_value = create_stream()
+
+    response = await text_client.send_text_message('Hi')
+
+    assert response == 'Hello Processing World!'
+    mock_client.send_message.assert_called_once()
+    # Verify request construction
+    args, _ = mock_client.send_message.call_args
+    request = args[0]
+    assert request.message.parts[0].text == 'Hi'
+
+
+@pytest.mark.asyncio
+async def test_send_text_message_forwards_context(
+    text_client: TextClient, mock_client: AsyncMock
+) -> None:
+
+    async def empty_stream(*args, **kwargs):
+        return
+        yield
+
+    mock_client.send_message.return_value = empty_stream()
+    context = ClientCallContext()
+
+    await text_client.send_text_message('Hi', context=context)
+
+    _, kwargs = mock_client.send_message.call_args
+    assert kwargs['context'] is context
+
+
+@pytest.mark.asyncio
+async def test_close(text_client: TextClient, mock_client: AsyncMock) -> None:
+    await text_client.close()
+    mock_client.close.assert_awaited_once()