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

samples/README.md

@@ -8,8 +8,9 @@ This directory contains runnable examples demonstrating how to build and interac
 |---|---|---|
 | `hello_world_agent.py` | **Server** | A2A agent server |
 | `cli.py` | **Client** | Interactive terminal client |
+| `text_client_cli.py` | **Client** | Simplified text-only interactive terminal client |
 
-The samples are designed to work together out of the box: the agent listens on `http://127.0.0.1:41241`, which is the default URL used by the client.
+All three samples are designed to work together out of the box: the agent listens on `http://127.0.0.1:41241`, which is the default URL used by both clients.
 ---
 
 ## `hello_world_agent.py` — Agent Server
@@ -53,6 +54,44 @@ uv run python samples/cli.py
 uv run python samples/cli.py --url http://192.168.1.10:41241 --transport GRPC
 ```
 
-Then type a message like `hello` and press Enter.
+Type `/quit` or `/exit` to stop, or press `Ctrl+C`.
+
+---
+
+## `text_client_cli.py` — Simple Text Client
+
+A stripped-down interactive client using the high-level `TextClient` abstraction. It hides all streaming and event mechanics, presenting a simple request/response interface.
+
+Ideal for understanding the **minimum code required** to call an A2A agent.
+
+**Run:**
+
+```bash
+# Connect to the local hello_world_agent (default):
+uv run python samples/text_client_cli.py
+
+# Connect to a different URL:
+uv run python samples/text_client_cli.py --url http://192.168.1.10:41241
+
+# Use a specific transport:
+uv run python samples/text_client_cli.py --transport GRPC
+```
 
 Type `/quit` or `/exit` to stop, or press `Ctrl+C`.
+
+---
+
+
+## Quick Start
+
+In two separate terminals:
+
+```bash
+# Terminal 1 — start the agent
+uv run python samples/hello_world_agent.py
+
+# Terminal 2 — start the client
+uv run python samples/cli.py
+```
+
+Then type a message like `hello` and press Enter.

samples/cli.py

@@ -62,30 +62,32 @@
 
 async def main() -> None:
     """Run the A2A terminal client."""
     parser = argparse.ArgumentParser(description='A2A Terminal Client')
     parser.add_argument(
         '--url', default='http://127.0.0.1:41241', help='Agent base URL'
     )
     parser.add_argument(
         '--transport',
         default=None,
         help='Preferred transport (JSONRPC, HTTP+JSON, GRPC)',
     )
     args = parser.parse_args()
 
     config = ClientConfig()
     if args.transport:
         config.supported_protocol_bindings = [args.transport]
+    if args.transport == 'GRPC':
+        config.grpc_channel_factory = grpc.aio.insecure_channel
 
     print(
         f'Connecting to {args.url} (preferred transport: {args.transport or "Any"})'
     )
 
     async with httpx.AsyncClient() as httpx_client:
         resolver = A2ACardResolver(httpx_client, args.url)
         card = await resolver.get_agent_card()
         print('\n✓ Agent Card Found:')
         display_agent_card(card)
 
     client = await create_client(card, client_config=config)
 

samples/text_client_cli.py

@@ -0,0 +1,70 @@
+import argparse
+import asyncio
+
+import grpc
+import httpx
+
+from a2a.client import A2ACardResolver, ClientConfig, 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'
+    )
+    parser.add_argument(
+        '--transport',
+        default=None,
+        help='Preferred transport (JSONRPC, HTTP+JSON, GRPC)',
+    )
+    args = parser.parse_args()
+
+    config = ClientConfig()
+    if args.transport:
+        config.supported_protocol_bindings = [args.transport]
+    if args.transport == 'GRPC':
+        config.grpc_channel_factory = grpc.aio.insecure_channel
+
+    print(
+        f'Connecting to {args.url} (preferred transport: {args.transport or "Any"})'
+    )
+
+    async with httpx.AsyncClient() as httpx_client:
+        resolver = A2ACardResolver(httpx_client, args.url)
+        card = await resolver.get_agent_card()
+        print('\n✓ Agent Card Found:')
+        print(f'  Name: {card.name}')
+
+    text_client = await create_text_client(card, client_config=config)
+
+    actual_transport = getattr(
+        text_client.client, '_transport', text_client.client
+    )
+    print(f'  Picked Transport: {actual_transport.__class__.__name__}')
+
+    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,114 @@
+import uuid
+
+from types import TracebackType
+
+from typing_extensions import Self
+
+from a2a.client.client import Client, ClientCallContext
+from a2a.types import Message, Part, Role, SendMessageRequest, TaskState
+from a2a.utils import get_artifact_text, get_message_text
+
+
+_TERMINAL_STATES: frozenset[TaskState] = frozenset(
+    {
+        TaskState.TASK_STATE_COMPLETED,
+        TaskState.TASK_STATE_FAILED,
+        TaskState.TASK_STATE_CANCELED,
+        TaskState.TASK_STATE_REJECTED,
+    }
+)
+
+
+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.
+    Maintains session state (context_id, task_id) automatically across calls.
+    For full Client API access, use the underlying client directly via
+    the `client` property.
+    """
+
+    def __init__(self, client: Client):
+        self._client = client
+        self._context_id: str = str(uuid.uuid4())
+        self._task_id: str | None = None
+
+    async def __aenter__(self) -> Self:
+        """Enters the async context manager."""
+        return self
+
+    async def __aexit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc_val: BaseException | None,
+        exc_tb: TracebackType | None,
+    ) -> None:
+        """Exits the async context manager and closes the client."""
+        await self.close()
+
+    @property
+    def client(self) -> Client:
+        """Returns the underlying Client instance for full API access."""
+        return self._client
+
+    def reset_session(self) -> None:
+        """Starts a new session by generating a fresh context ID and clearing the task ID."""
+        self._context_id = str(uuid.uuid4())
+        self._task_id = None
+
+    async def send_text_message(
+        self,
+        text: str,
+        *,
+        delimiter: str = ' ',
+        context: ClientCallContext | None = None,
+    ) -> str:
+        """Sends a text message and returns the aggregated text response.
+
+        Session state (context_id, task_id) is managed automatically across
+        calls. Use reset_session() to start a new conversation.
+
+        Args:
+            text: The plain-text message to send.
+            delimiter: String used to join response parts. Defaults to a
+                single space. Use '' for token-streamed responses or a
+                newline for paragraph-separated chunks.
+            context: Optional call-level context.
+        """
+        request = SendMessageRequest(
+            message=Message(
+                role=Role.ROLE_USER,
+                message_id=str(uuid.uuid4()),
+                context_id=self._context_id,
+                task_id=self._task_id,
+                parts=[Part(text=text)],
+            )
+        )
+
+        response_parts: list[str] = []
+
+        async for event in self._client.send_message(request, context=context):
+            if event.HasField('task'):
+                self._task_id = event.task.id
+            elif event.HasField('message'):
+                response_parts.append(get_message_text(event.message))
+            elif event.HasField('status_update'):
+                if not self._task_id and event.status_update.task_id:
+                    self._task_id = event.status_update.task_id
+                if event.status_update.status.state in _TERMINAL_STATES:
+                    self._task_id = None
+                if event.status_update.status.HasField('message'):
+                    response_parts.append(
+                        get_message_text(event.status_update.status.message)
+                    )
+            elif event.HasField('artifact_update'):
+                response_parts.append(
+                    get_artifact_text(event.artifact_update.artifact)
+                )
+
+        return delimiter.join(response_parts)
+
+    async def close(self) -> None:
+        """Closes the underlying client."""
+        await self._client.close()