`client.py`

65 lines
"""Minimal A2A (Agent2Agent) demo client — spec-compliant raw JSON-RPC over httpx.

Deliberately does NOT use the SDK client, so it self-documents the A2A wire protocol:
  1. DISCOVER  — GET the agent card from /.well-known/agent-card.json
  2. INVOKE    — POST a JSON-RPC 2.0 `message/send` with a text message
  3. READ      — print the agent's reply

Run (server must be up): . .venv/bin/activate && python client.py
"""
from __future__ import annotations

import json
import os
import sys
import uuid

import httpx

# Override with A2A_BASE to point at another agent (e.g. the tk bridge on :41242).
BASE = os.environ.get("A2A_BASE", "http://127.0.0.1:41241")


def discover(base: str) -> dict:
    card = httpx.get(f"{base}/.well-known/agent-card.json", timeout=10).json()
    print(f"① DISCOVERED agent card → name={card.get('name')!r} version={card.get('version')!r}")
    ifaces = card.get("supportedInterfaces") or card.get("supported_interfaces") or []
    if ifaces:
        print(f"   interface → {ifaces[0]}")
    return card


def send_message(base: str, text: str) -> dict:
    # a2a-sdk 1.1.2 uses gRPC-style method names (SendMessage) and parses params
    # as a protobuf SendMessageRequest (camelCase fields, enum-name roles).
    payload = {
        "jsonrpc": "2.0",
        "id": str(uuid.uuid4()),
        "method": "SendMessage",
        "params": {
            "message": {
                "messageId": str(uuid.uuid4()),
                "role": "ROLE_USER",
                "parts": [{"text": text}],
            }
        },
    }
    print(f"② SEND SendMessage → {text!r}")
    # The server defaults a missing A2A-Version header to legacy '0.3'; signal 1.0.
    headers = {"A2A-Version": "1.0"}
    resp = httpx.post(f"{base}/", json=payload, headers=headers, timeout=30)
    return resp.json()


def main() -> int:
    text = sys.argv[1] if len(sys.argv) > 1 else "hello a2a"
    discover(BASE)
    result = send_message(BASE, text)
    print("③ RESPONSE:")
    print(json.dumps(result, indent=2))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())