Messages and Streaming Calls

A2A messages use a JSON-RPC 2.0 envelope. MOI adds the Agent selector at the top level, while params.message contains the user message. Synchronous and streaming calls share the same message shape; only the method and response transport differ.

Build a user message

This request selects the explore Agent and sends one text Part:

{
  "agent_code": "explore",
  "jsonrpc": "2.0",
  "id": "req_01J0A",
  "method": "message/send",
  "params": {
    "message": {
      "kind": "message",
      "role": "user",
      "messageId": "msg_01J0A",
      "parts": [
        {
          "kind": "text",
          "text": "Summarize quarterly sales by region"
        }
      ]
    }
  }
}

Each ID has a distinct purpose:

  • The JSON-RPC id correlates a request and response. Generate a unique value for each call.

  • messageId identifies the user message. Do not casually change it when retrying the same logical message.

  • The Task id returned by the server identifies one execution.

  • contextId identifies conversation context that a later message can reuse.

To continue an earlier context, put the returned contextId on the next message:

{
  "kind": "message",
  "role": "user",
  "messageId": "msg_01J0B",
  "contextId": "ctx_123",
  "parts": [
    {
      "kind": "text",
      "text": "Keep only the East and South regions, sorted by sales descending"
    }
  ]
}

Do not substitute the JSON-RPC id or Task id for contextId. Whether a concrete Agent also needs the previous Task id is Agent-specific; unless its contract says otherwise, send only the context fields returned by the service.

Non-streaming calls

message/send returns a JSON-RPC document. A successful response normally has an A2A Task in result, but a client should check for error first and then dispatch on result.kind.

import os
import uuid
import requests

base_url = os.environ["MOI_BASE_URL"].rstrip("/")
payload = {
    "agent_code": "explore",
    "jsonrpc": "2.0",
    "id": f"req_{uuid.uuid4().hex}",
    "method": "message/send",
    "params": {
        "message": {
            "kind": "message",
            "role": "user",
            "messageId": f"msg_{uuid.uuid4().hex}",
            "parts": [{"kind": "text", "text": "List the five highest-selling products this month"}],
        }
    },
}

response = requests.post(
    f"{base_url}/newmoi/agents/a2a",
    json=payload,
    headers={"moi-key": os.environ["MOI_API_KEY"]},
    timeout=120,
)
response.raise_for_status()
document = response.json()

if "error" in document:
    raise RuntimeError(document["error"])

result = document["result"]
task_id = result.get("id") if result.get("kind") == "task" else None
context_id = result.get("contextId")

Do not treat HTTP 200 as a completion signal. The response may only have created a Task that is still working; confirm completion from Task state or streaming events.

Streaming calls

Change the method to message/stream while keeping the rest of the body. The response becomes text/event-stream:

event: message
id: 12
data: {"jsonrpc":"2.0","id":"req_01J0A","result":{"kind":"status-update",...}}

An event can carry several result kinds:

  • The first successful event normally contains the Task, allowing the client to bind to a stable Task id early.

  • status-update changes Task state.

  • artifact-update carries text, structured data, or another Agent artifact.

  • Some terminal paths can return an A2A Message directly; a parser must not assume every event is a Task update.

The stream can also contain comment frames such as : heartbeat. An SSE client must ignore comments and blank separators rather than trying to decode them as JSON.

This minimal Python reader preserves the event sequence and supports multiline data:

import json
import requests

last_event_id = None
event_type = "message"
data_lines = []

with requests.post(
    f"{base_url}/newmoi/agents/a2a",
    json={**payload, "method": "message/stream"},
    headers={
        "moi-key": os.environ["MOI_API_KEY"],
        "Accept": "text/event-stream",
    },
    stream=True,
    timeout=(10, None),
) as response:
    response.raise_for_status()

    for line in response.iter_lines(decode_unicode=True):
        if line == "":
            if data_lines:
                event = json.loads("\n".join(data_lines))
                handle_a2a_event(event_type, last_event_id, event)
            event_type, data_lines = "message", []
            continue
        if line.startswith(":"):
            continue
        if line.startswith("event:"):
            event_type = line[6:].strip()
        elif line.startswith("id:"):
            last_event_id = line[3:].strip()
        elif line.startswith("data:"):
            data_lines.append(line[5:].lstrip())

handle_a2a_event is the application’s projection function. Dispatch on result.kind, update Task state, append Artifacts, and persist last_event_id with the business result. Never render an unknown Data Part as plain text without validation.

Timeouts, disconnects, and retries

  • Configure connection and read timeouts separately. A long-running Task should not inherit the short read timeout used for ordinary JSON calls.

  • If a stream ends before a terminal state, do not synthesize completed or failed. Call tasks/resubscribe with the last saved sequence.

  • Automatically resend message/stream only when you know the request did not reach the server or your application has a stable message-ID/idempotency strategy. Otherwise, a retry can create a duplicate Task.

  • Treat JSON-RPC error, non-2xx HTTP responses, and SSE read failures as separate failure classes in logs and recovery logic.