Skip to content

client

client

A2A client — discover and call external A2A agents.

Classes

A2AClient 

A2AClient(base_url: str, *, timeout: float = 30.0)

Client for calling external A2A-compatible agents.

Discovers agent capabilities via /.well-known/agent.json and sends tasks via /a2a/tasks.

Source code in src/openjarvis/a2a/client.py 
17
18
19
20
def __init__(self, base_url: str, *, timeout: float = 30.0) -> None:
    self._base_url = base_url.rstrip("/")
    self._timeout = timeout
    self._card: Optional[AgentCard] = None
Functions
discover 
discover() -> AgentCard

Fetch the agent card from /.well-known/agent.json.

Source code in src/openjarvis/a2a/client.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
def discover(self) -> AgentCard:
    """Fetch the agent card from /.well-known/agent.json."""
    import httpx
    resp = httpx.get(
        f"{self._base_url}/.well-known/agent.json",
        timeout=self._timeout,
    )
    resp.raise_for_status()
    data = resp.json()
    self._card = AgentCard(
        name=data.get("name", ""),
        description=data.get("description", ""),
        url=data.get("url", self._base_url),
        version=data.get("version", ""),
        capabilities=data.get("capabilities", []),
        skills=data.get("skills", []),
    )
    return self._card
send_task 
send_task(input_text: str, **kwargs: Any) -> A2ATask

Send a task to the remote agent and return the result.

Source code in src/openjarvis/a2a/client.py 
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
def send_task(self, input_text: str, **kwargs: Any) -> A2ATask:
    """Send a task to the remote agent and return the result."""
    import httpx
    request = A2ARequest(
        method="tasks/send",
        params={
            "message": {
                "role": "user",
                "parts": [{"text": input_text}],
            },
        },
    )
    resp = httpx.post(
        f"{self._base_url}/a2a/tasks",
        json=request.to_dict(),
        timeout=self._timeout,
    )
    resp.raise_for_status()
    data = resp.json()
    result = data.get("result", {})
    return A2ATask(
        task_id=result.get("id", ""),
        state=result.get("state", "unknown"),
        input_text=result.get("input", input_text),
        output_text=result.get("output", ""),
        history=result.get("history", []),
    )
get_task 
get_task(task_id: str) -> A2ATask

Get the status of a previously submitted task.

Source code in src/openjarvis/a2a/client.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
def get_task(self, task_id: str) -> A2ATask:
    """Get the status of a previously submitted task."""
    import httpx
    request = A2ARequest(
        method="tasks/get",
        params={"id": task_id},
    )
    resp = httpx.post(
        f"{self._base_url}/a2a/tasks",
        json=request.to_dict(),
        timeout=self._timeout,
    )
    resp.raise_for_status()
    data = resp.json()
    result = data.get("result", {})
    return A2ATask(
        task_id=result.get("id", task_id),
        state=result.get("state", "unknown"),
        output_text=result.get("output", ""),
    )
cancel_task 
cancel_task(task_id: str) -> A2ATask

Cancel a running task.

Source code in src/openjarvis/a2a/client.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
def cancel_task(self, task_id: str) -> A2ATask:
    """Cancel a running task."""
    import httpx
    request = A2ARequest(
        method="tasks/cancel",
        params={"id": task_id},
    )
    resp = httpx.post(
        f"{self._base_url}/a2a/tasks",
        json=request.to_dict(),
        timeout=self._timeout,
    )
    resp.raise_for_status()
    data = resp.json()
    result = data.get("result", {})
    return A2ATask(
        task_id=result.get("id", task_id),
        state=result.get("state", "canceled"),
    )