From a card to a stream.

Install

moon add moonbitstack/moona2a

Three packages arrive together. The root holds the wire model, the task lifecycle and the client; asgi turns a server into an HTTP application; fetch is the client's HTTP transport and the webhook notifier.

The card

A card is what a caller reads before it says anything. It names the agent, lists the interfaces it answers on, and says which capabilities are on.

let skill = @moona2a.AgentSkill::new(
  "summarise",
  "Summarise",
  "condenses a document into a short paragraph",
  ["text", "summarise"],
)
let card = @moona2a.AgentCard::new(
  "summariser",
  "An agent that condenses documents",
  "1.0.0",
  "https://summariser.example.com",
  [skill],
).capable({
  streaming: true,
  push_notifications: false,
  extended_agent_card: false,
  extensions: [],
})

The fourth argument is where the agent answers; the card writes it as one JSON-RPC interface at protocol 1.0. Printing the card is how most integration bugs are found, so it is the first example in the repository:

{
  "name": "summariser",
  "description": "An agent that condenses documents",
  "supportedInterfaces": [
    {
      "url": "https://summariser.example.com",
      "protocolBinding": "JSONRPC",
      "protocolVersion": "1.0"
    }
  ],
  "version": "1.0.0",
  "capabilities": { "streaming": true, "pushNotifications": false },
  "defaultInputModes": ["text/plain"],
  "defaultOutputModes": ["text/plain"],
  "skills": [ … ]
}

An agent, and a server

An agent is an executor: it is handed the message and a sink, and it says what happens. It does not create tasks, keep history or answer GetTask — the server does that, from the events the executor emits.

struct Echo { }

impl @moona2a.Executor for Echo with fn execute(self, ctx, out) {
  ignore(self)
  out.emit(ctx.status(@moona2a.Working))
  out.emit(ctx.artifact(@moona2a.Artifact::text("answer", "You said: " + ctx.message.text())))
  out.emit(ctx.status(@moona2a.Completed))
}

impl @moona2a.Executor for Echo with fn cancel(self, task_id) {
  ignore(self)
  ignore(task_id)
}

pub extend Echo with @moona2a.Executor::{execute, cancel}

The first event an agent emits is what opens the task, so an agent that answers with a message leaves no empty task behind. A run that raises still leaves a task in TASK_STATE_FAILED, which is the state a subscriber needs to stop waiting.

the agent finished task a2a-1793925374888-2: TASK_STATE_COMPLETED
You said: hello

Put it on an endpoint

The server is an ASGI application, so the HTTP server stays a choice the host makes. to_asgi answers two addresses: the card at its well-known path, and every method on the JSON-RPC path.

let server = @moona2a.Server::new(Echo::{ }, card)
let app = @asgi.to_asgi(server)

// app : (scope, receive, send) -> Unit, the shape an ASGI server calls.

A streaming method answers text/event-stream and one data: frame per event, written as the event happens. A call refused before the stream opens — an unknown task, a version the interface does not serve — is an ordinary JSON-RPC error body instead, because a stream carrying one error is worse than no stream.

Call an agent

A client is the same protocol from the other side. Client is the protocol; Transport is how messages travel, which is why a test can answer from a table and production can use HTTP.

let transport = @fetch.HttpTransport::new("https://summariser.example.com")
let client = @moona2a.Client::new(transport)

let card = client.discover()
match client.send_message(@moona2a.Message::user("m-1", "summarise this")) {
  @moona2a.TaskSnapshot(task) =>
    println("task \{task.id} is \{task.status.state.wire()}")
  @moona2a.TaskMessage(reply) => println(reply.text())
  _ => println("the agent answered with something else")
}

task t-1 is TASK_STATE_COMPLETED
A short paragraph.

send_message returns the event that ended the exchange: a task when the agent worked one, a message when it answered directly.

Follow a task

A streaming call takes a reply sink instead of returning a value. The sink is called once per event, in the order the agent took them, and the stream ends at a terminal state.

struct Printer { }

impl @moona2a.ReplySink for Printer with fn reply(self, message) {
  ignore(self)
  match message {
    @moonjrpc.Reply(result~, ..) => {
      let event : @moona2a.StreamEvent = @json.from_json(result) catch {
        _ => @moona2a.StreamEvent::TaskMessage(@moona2a.Message::agent("?", "?"))
      }
      match event {
        @moona2a.TaskSnapshot(task) =>
          println("task \{task.id}: \{task.status.state.wire()}")
        @moona2a.StatusChange(update) => println("status: \{update.status.state.wire()}")
        @moona2a.ArtifactChunk(update) => println("artifact: \{update.artifact.text_of()}")
        @moona2a.TaskMessage(message) => println("message: \{message.text()}")
      }
    }
    @moonjrpc.Refusal(failure~, ..) => println("refused: \{failure.message}")
    _ => ()
  }
}

task a2a-1793925-1: TASK_STATE_SUBMITTED
status: TASK_STATE_WORKING
artifact: 1. read the input
artifact: 2. write the answer
status: TASK_STATE_COMPLETED

A caller that arrives late uses SubscribeToTask instead: it is given the task as it stands, then everything that happens after it joined. Several callers can watch the same task; each reads the log from its own cursor, so one leaving affects no other.

Leave a webhook

For a task that takes minutes, a caller can leave a URL and stop asking. The agent posts each update to it, with the token and the credentials the registration names.

let config = @moona2a.TaskPushConfig::new(
  "cfg-1",
  task_id,
  "https://caller.example/hook",
).with_token("s3cret")

client.create_push_config(config) |> ignore

The notifier is a trait, so an agent that keeps its registrations in a database or delivers over something other than HTTP replaces it. @fetch.Webhook is the HTTP one: it posts application/a2a+json, adds x-a2a-notification-token when a token was set, and an authorization header for registered credentials.

When a call is refused

Refusals are the specification's own, with the code of section 5.4 and a google.rpc.ErrorInfo detail, so a client branches on the reason rather than on the sentence.

TaskNotFound                    -32001   no such task
TaskNotCancelable               -32002   it is finished, or it cannot stop
PushNotificationNotSupported    -32003   the agent takes no webhooks
UnsupportedOperation            -32004   the method, or a task that is over
ContentTypeNotSupported         -32005   a part the agent cannot read
InvalidAgentResponse            -32006   the agent answered with something else
ExtendedAgentCardNotConfigured  -32007   there is no second card
ExtensionSupportRequired        -32008   an extension the caller must honour
VersionNotSupported             -32009   a protocol version this interface does not speak

An agent that raises its own error still leaves a task: the server records it as TASK_STATE_FAILED and hands the refusal back, so a watcher is never left waiting on a task that will not move.

Rules that bite

• Send A2A-Version: 1.0. A request that names another version is refused, and so is one that names none: the specification reads a missing header as 0.3, which this interface does not serve. Patch numbers do not count.
• Ids are the server's. The caller chooses messageId; taskId and contextId are minted by the server. A message that names a task continues it; one that names a finished task is refused.
• Media types are the agent's call. Nothing is refused because its media type is missing from the card. Turn on ServerConfig.refuse_unknown_media to make the card's list a hard filter.
• History is not guaranteed. The agent decides what is worth keeping, so a client reads the status message for the agent's own words rather than assuming they are in the history.