a2a_showcase/interactive_client.py at main · Doca/a2a_showcase · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
"""Interactive A2A client — chat with any A2A agent from the terminal."""

import asyncio
import sys
from uuid import uuid4

import httpx
from rich.console import Console
from rich.panel import Panel

from a2a.client import ClientConfig, ClientFactory
from a2a.client.card_resolver import A2ACardResolver
from a2a.types import Message, Part, TextPart, TransportProtocol

console = Console()


def _make_message(text: str, context_id: str | None = None) -> Message:
    return Message(
        message_id=str(uuid4()),
        role="user",
        parts=[Part(root=TextPart(text=text))],
        **({"context_id": context_id} if context_id else {}),
    )


async def _send_and_collect(client, message):
    state = "unknown"
    response_text = ""
    context_id = None

    async for event in client.send_message(message):
        if isinstance(event, Message):
            for part in event.parts:
                p = part.root if hasattr(part, "root") else part
                if hasattr(p, "text"):
                    response_text = p.text
            context_id = event.context_id
        elif isinstance(event, tuple):
            task, update = event
            context_id = task.context_id
            if task.status:
                state = task.status.state.value if hasattr(task.status.state, "value") else str(task.status.state)
                if task.status.message and task.status.message.parts:
                    for part in task.status.message.parts:
                        p = part.root if hasattr(part, "root") else part
                        if hasattr(p, "text"):
                            response_text = p.text

    return state, response_text, context_id


async def main():
    url = sys.argv[1] if len(sys.argv) > 1 else "http://localhost:9100"

    console.print(Panel(f"Interactive A2A Client — connecting to {url}", style="bold cyan", expand=False))

    # Discover agent
    async with httpx.AsyncClient() as http_client:
        resolver = A2ACardResolver(httpx_client=http_client, base_url=url)
        card = await resolver.get_agent_card()

    console.print(f"  [green]Connected to:[/green] {card.name} — {card.description}")
    skills = ", ".join(f"{s.id} ({s.description})" for s in card.skills)
    console.print(f"  [green]Skills:[/green] {skills}")
    console.print(f"\n  Type your messages below. Press Ctrl+C to quit.\n")

    http_client = httpx.AsyncClient(timeout=600.0)
    config = ClientConfig(
        streaming=False,
        httpx_client=http_client,
        supported_transports=[TransportProtocol.http_json],
    )
    factory = ClientFactory(config)
    client = factory.create(card)

    context_id = None

    try:
        while True:
            try:
                user_input = await asyncio.get_event_loop().run_in_executor(
                    None, lambda: input("You> ")
                )
            except EOFError:
                break

            if not user_input.strip():
                continue

            message = _make_message(user_input, context_id=context_id)
            state, text, context_id = await _send_and_collect(client, message)

            console.print(f"[yellow]{card.name}[/yellow] [dim](state: {state})[/dim]")
            console.print(f"  {text}\n")

    except KeyboardInterrupt:
        console.print("\n[dim]Bye![/dim]")
    finally:
        await http_client.aclose()


if __name__ == "__main__":
    asyncio.run(main())