[Feat]: Reduced Boilerplate for A2A Python SDK Text Messaging

[msampathkumar]

Is your feature request related to a problem? Please describe.

While A2A provides robust message constructs, the process for a basic A2A Client to send and receive simple text messages currently requires excessive boilerplate. This feature dramatically reduces the complexity for common text-only exchanges, improving developer experience.

Here is a code sample from https://github.com/a2aproject/a2a-samples/blob/b5af0df12fe941ae79c52684fc1fd5357c4f88a6/samples/python/agents/helloworld/test_client.py#L58-L76

In the following code, we are sending a text message say hello. to the A2A server and print the response.

    client_factory = ClientFactory(config=ClientConfig(streaming=False))
    client = client_factory.create(_public_card)
    logger.info('\nNon-streaming A2AClient initialized.')

    parts = [Part(text='Say hello.')]
    message = Message(
        role=Role.ROLE_USER,
        parts=parts,
        message_id=uuid4().hex,
    )
    request = SendMessageRequest(message=message)

    response = client.send_message(request)

    async for chunk in response:
        print('Response:')
        task, _ = chunk
        print(task)

Problem: This is has many steps to do, methods or functions that users to know and a complex workflow to remember and follow.

Describe the solution you'd like

Request: Simplify this something like below

    client = A2AClient(
        agent_card=_public_card,				# Provide AgentCard information
        config=types.ClientConfig(streaming=False)	# Optional param - config
        )
    response = client.send_message(
        message=types.user_message(text='Say hello.')	# simple text message
        )
    print(response.text)						# provide a `.text` (property) to get access text immediately

Summary of Developer Improvements Requetss:
Simple jobs should be simple enough to do.

• Provide a simple way to pack text input message

  Eliminate boilerplate by providing a simple utility, types.user_message(text=...), to construct text input without manually instantiating Part, Message, and SendMessageRequest objects.

• Provide a simple way to get & print text output message

  Allow immediate, non-streaming access to the text output via a convenient .text property, simplifying response handling and eliminating the need for async chunk iteration in non-streaming scenarios.

Describe alternatives you've considered

No response

Additional context

No response

Code of Conduct

• I agree to follow this project's Code of Conduct

[benclarkeio]

Hey @msampathkumar, you might be interested in A2A Utils. The boilerplate code is actually worse than that.

Here’s one of most common operations in A2A: sending a message and returning the response if it’s a Message, or polling the Task until it’s in a non-terminal state.

A2A SDK:

import asyncio
from uuid import uuid4

import httpx
from a2a.client import A2AClient, A2ACardResolver
from a2a.types import (
    GetTaskRequest,
    GetTaskResponse,
    GetTaskSuccessResponse,
    JSONRPCErrorResponse,
    Message,
    MessageSendConfiguration,
    MessageSendParams,
    Part,
    Role,
    SendMessageRequest,
    SendMessageResponse,
    SendMessageSuccessResponse,
    Task,
    TaskQueryParams,
    TaskState,
    TextPart,
)

TERMINAL_OR_ACTIONABLE_STATES = {
    TaskState.completed,
    TaskState.failed,
    TaskState.canceled,
    TaskState.rejected,
    TaskState.input_required,
    TaskState.auth_required,
}

def get_result_or_raise_error(response: SendMessageResponse | GetTaskResponse) -> Task | Message:
    response_root = response.root

    if isinstance(response_root, JSONRPCErrorResponse):
        raise RuntimeError(f"JSON-RPC error: {response_root.error.message}")

    if not isinstance(response_root, (SendMessageSuccessResponse, GetTaskSuccessResponse)):
        raise TypeError(f"Unexpected response type: {type(response_root).__name__}")

    return response_root.result

async def main() -> Task | Message:
    async with httpx.AsyncClient() as httpx_client:
        # 1. Resolve the Agent Card
        resolver = A2ACardResolver(
            httpx_client=httpx_client,
            base_url="https://research.example.com",
        )
        agent_card = await resolver.get_agent_card(
            http_kwargs={"headers": {"X-API-Key": "key_123"}}
        )

        # 2. Build the Message
        send_request = SendMessageRequest(
            id=str(uuid4()),
            params=MessageSendParams(
                message=Message(
                    message_id=str(uuid4()),
                    parts=[
                        Part(root=TextPart(text="Find recent papers on quantum computing")),
                    ],
                    role=Role.user,
                ),
                configuration=MessageSendConfiguration(blocking=False),
            ),
        )

        # 3. Send the Message (non-blocking)
        client = A2AClient(httpx_client=httpx_client, agent_card=agent_card)
        response = await client.send_message(
            request=send_request,
            http_kwargs={"headers": {"X-API-Key": "key_123"}},
        )

        task_or_message = get_result_or_raise_error(response)

        if isinstance(task_or_message, Message):
            return task_or_message

        task = task_or_message

        # 4. Poll until the Task reaches a terminal state
        while task.status.state not in TERMINAL_OR_ACTIONABLE_STATES:
            await asyncio.sleep(5)
            response = await client.get_task(
                request=GetTaskRequest(
                    id=str(uuid4()),
                    params=TaskQueryParams(id=task.id),
                ),
                http_kwargs={"headers": {"X-API-Key": "key_123"}},
            )
            result = get_result_or_raise_error(response)
            if not isinstance(result, Task):
                raise TypeError(f"Expected Task response, got {type(result).__name__}")
            task = result

        return task

asyncio.run(main())

A2A Utils:

import asyncio

from a2a.types import Message, Task

from a2a_utils import A2AAgents, A2ASession, TERMINAL_OR_ACTIONABLE_STATES

async def main() -> Task | Message:
    a2a_agents = A2AAgents(
        {
            "research-bot": {
                "url": "https://research.example.com/.well-known/agent-card.json",
                "custom_headers": {"X-API-Key": "key_123"},
            },
        }
    )

    session = A2ASession(agents=a2a_agents)

    response = await session.send_message(
        "research-bot", "Find recent papers on quantum computing"
    )

    if isinstance(response, Message):
        return response

    while response.status.state not in TERMINAL_OR_ACTIONABLE_STATES:
        response = await session.get_task("research-bot", response.id)

    return response

asyncio.run(main())

In the A2A Utils example, A2AAgents automatically fetches the Agent Card, send_message sends the message as non-blocking, streams the response (falling back to polling if the agent doesn’t support streaming), and returns a Message or Task. If it times out, get_task can be used to continue streaming (or polling). Note that the A2A SDK example above is simplified, because it doesn’t check if the agent supports streaming. It also doesn’t save Tasks, Artifacts, and files locally.