a2a_async

Asynchronous send-and-poll A2A communication between Agent Zero agents. The plugin tools send a message to a remote FastA2A-compatible agent and return immediately with the created task id, so the agent turn is never blocked — unlike the core a2a_chat tool, which waits for the remote answer (up to its wait_for_completion limit) and typically times out on long-running remote agents.

How it works

• a2a_send — sends message/send through helpers.fasta2a_client.AgentConnection.send_message and returns task_id, context_id and the initial state (submitted). The FastA2A server creates the task immediately and does not wait for the result. The task is registered in the pending registry.
• a2a_poll — polls tasks/get for the given task_id (or the latest pending task for an agent_url). submitted/working → a "still running" notice; completed → the full response text and artifacts (entry removed); failed/canceled/rejected → a notice and entry removal; unknown task (remote restart, error -32001) → a readable notice and entry removal.
• a2a_cancel — sends the JSON-RPC tasks/cancel method. The framework client (fasta2a.client.A2AClient) does not expose it, so the plugin sends its own POST request through httpx (plugin-local helper helpers/cancel.py), following the library's client.py payload pattern. No framework files are modified.
• [EXTRAS] reminders — between turns the agent sees the list of pending tasks (task_id, agent_url, state, age) with a hint to use a2a_poll. Mechanism: implicit @extensible hook Agent.prepare_prompt (decorator in agent.py), layout extensions/python/_functions/agent/Agent/prepare_prompt/start/. The start hook runs before the [EXTRAS] block is rendered, so the entry lands in the current turn's prompt — the same information source as the core message_loop_prompts_after → _72_include_parallel_jobs.py.

Task registry

Entries live in the agent data slot: agent.get_data/agent.set_data("_a2a_async_tasks") — the same pattern as _a2a_sessions in the core a2a_chat tool. Each entry (dataclass AsyncTaskEntry) tracks: task_id, agent_url, context_id, state, created_at, last_polled_at, poll_count.

Design decision: the entry is removed on every terminal state (completed, failed, canceled, rejected) and when the remote agent reports -32001 (task unknown). The registry only keeps tasks that still need polling; the response returns into the conversation history, so it does not need to be stored twice. Connection errors do NOT remove the entry — the remote agent may be temporarily unavailable.

Call context requirement (important)

Because the registry lives in agent context data, the tools that touch it must run top-level in the same agent context where a2a_send was called:

• a2a_poll, a2a_cancel (and follow-up a2a_send calls for the same task) must never be issued through parallel (temporary worker context) or a subagent with a different context — the registry lookup reads that context's data slot and will not find the task.
• Verified empirically: the same task_id polled through parallel returns task <id> is not in the pending registry, while the same poll top-level returns the full completed result — with the remote agent healthy (tasks/get correctly reporting completed/submitted) the whole time.
• Classic symptom: not in the pending registry while the remote agent is healthy means the call came from the wrong context, not that the tool or the remote agent is broken — repeat the call top-level in the sending agent's context.
• The [EXTRAS] reminder (a2a_async pending tasks) is reliable evidence of the correct slot: it reads the same context data key as the registry, so polling top-level right after seeing it always works.
• Fail-fast guard (since v0.1.2): a2a_send, a2a_poll and a2a_cancel detect a parallel tool worker with the framework helper helpers.parallel_tools.is_parallel_worker and return a readable error instead of performing the operation — nothing is written to the registry. Reason: every direct parallel job runs in a temporary worker context that _run_direct_tool_job (helpers/parallel_tools.py) deletes in a finally block as soon as the job finishes, so a registration made there would be lost. Finish or collect parallel jobs first, then call the A2A tools top-level.

Receiver-side limitations (FastA2A defaults)

The reference FastA2A receiver uses the default InMemoryStorage + InMemoryBroker stack, which imposes practical limits:

• Single sequential worker, no controllable queue depth: the worker consumes task operations one at a time from the broker's in-memory stream (a single _loop consumer in fasta2a/worker.py); there is no server-side queue with configurable depth. A long queue only extends the time to completed.
• No persistence across receiver restarts: tasks live in in-memory dicts, so after a receiver restart tasks/get returns -32001; the plugin reacts with a readable notice, removes the registry entry, and the work must be resent with a2a_send.
• Practical rule — one task at a time: send a single task to a given receiver and a2a_poll it before dispatching the next one. Concurrent sends to the same receiver are serialized by the single worker; under load they can be rejected or hit the 30 s client HTTP timeout instead of queueing reliably.

Installation

No external dependencies — httpx and fasta2a already ship with the Agent Zero framework runtime.

Install from Git (Settings → Plugins → Install from Git):

https://github.com/momo2002/a0-a2a_async.git

Alternatively, clone or copy the repository into usr/plugins/a2a_async/ and enable it in the Plugins UI (Settings → Plugins) or via the management API. install() in hooks.py only validates imports (idempotent).

Usage (model workflow)

1. a2a_send with agent_url and message → you get a task_id.
2. Do other work; pending tasks are visible in [EXTRAS] between turns.
3. a2a_poll with task_id (or agent_url) → the result once it is ready.
4. Optionally a2a_cancel when the task is no longer needed.

Limitations

• Same-context polling required: the pending-task registry lives in the sending agent's context data — a2a_poll/a2a_cancel called through parallel or a subagent with a different context will report not in the pending registry even though the remote agent is healthy; since v0.1.2 the tools fail fast with a readable error when called inside parallel. See Call context requirement.
• No persistence across remote restarts: FastA2A tasks live in the server storage (default InMemoryStorage); after a restart tasks/get returns -32001 — the plugin reacts with a readable notice and removes the entry. Resend with a2a_send.
• Single sequential server worker: the default receiver executes tasks one at a time from the broker's in-memory stream, with no controllable queue depth — see Receiver-side limitations.
• SSE/streaming unavailable: fasta2a 0.5.0 does not implement message/stream (NotImplementedError in task_manager.py), so polling is the only path.
• blocking: true ignored by the server: the core client sends blocking=True, but the fasta2a server only reads history_length — send-and-poll has the same semantics as a2a_chat.
• Cancel is best-effort: tasks/cancel is accepted by the server, but the final state depends on the broker; verify with a2a_poll after cancelling.
• HTTP timeout: the framework client hardcodes a 30 s per-request timeout; on slower networks a connection error is returned as a readable message.

Repository layout

a2a_async/
├── plugin.yaml
├── hooks.py
├── helpers/
│   ├── registry.py          # pending-task registry (dataclass, pure logic)
│   ├── response_extract.py  # text/artifact extraction from JSON-RPC
│   ├── cancel.py            # plugin-local JSON-RPC tasks/cancel (httpx)
│   └── parallel_guard.py    # fail-fast guard for `parallel` worker contexts
├── tools/
│   ├── a2a_send.py
│   ├── a2a_poll.py
│   └── a2a_cancel.py
├── extensions/python/_functions/agent/Agent/prepare_prompt/start/
│   └── _76_a2a_async_pending.py   # [EXTRAS] reminders
├── prompts/
│   ├── agent.system.tool.a2a_send.md
│   ├── agent.system.tool.a2a_poll.md
│   └── agent.system.tool.a2a_cancel.md
└── tests/
    ├── test_plugin.py      # assert-based unit tests (no pytest needed)
    └── e2e_local.py        # end-to-end test against a real FastA2A server

Tests

Run with the framework runtime (Docker layout):

cd /a0 && /opt/venv-a0/bin/python usr/plugins/a2a_async/tests/test_plugin.py
cd /a0 && /opt/venv-a0/bin/python usr/plugins/a2a_async/tests/e2e_local.py

License

MIT — see LICENSE.