> ## Documentation Index
> Fetch the complete documentation index at: https://docs.aion.to/llms.txt
> Use this file to discover all available pages before exploring further.

# Message Mapping

> How Aion Server maps A2A requests and events to LangGraph and back.

This page describes how Aion Server adapts A2A requests to LangGraph and maps LangGraph outputs and events back
into A2A Messages, Tasks, and streaming events.

## 1. Inbound Messages

Four surfaces carry an inbound request into a graph run:

| Surface                               | Purpose                                                                       |
| ------------------------------------- | ----------------------------------------------------------------------------- |
| `state.messages`                      | Model-facing transcript for ordinary LangGraph and LangChain logic            |
| `runtime.context`                     | Aion request context, from which `thread`, `message`, and `event` are derived |
| `runtime.context.inbox`               | Raw A2A task, message, and request metadata                                   |
| `config["configurable"]["thread_id"]` | LangGraph's own checkpoint thread identifier                                  |

`thread_id` is LangGraph's persistence key, not Aion's messaging thread or context
identifier. The inbound distribution context arrives through `runtime.context`.

### 1.1 Graph Invocation

Both `SendMessage` and `SendStreamingMessage` use the same event generation flow:
`AgentExecutor.execute()` always produces an event stream via `graph.astream()`.

The difference is in how `DefaultRequestHandler.ResultAggregator` consumes this stream:

* `SendMessage` (`blocking=true`) collects all events and returns the final `Task`.
* `SendMessage` (`blocking=false`) returns after the first event, then continues processing in background
  with `status="submitted"`.
* `SendStreamingMessage` yields events as they arrive and streams them to the client via SSE.

Both methods use the same `SendMessageRequest` payload; only the response mode differs.

Older A2A examples before 0.3 used separate `invoke()` and `stream()` paths. In A2A 0.3+, execution is unified;
only the consumption strategy differs.

### 1.2 Message Ingress

When an inbound A2A `Message` arrives:

1. Append to `state.messages` (LLM-facing transcript):
   If the inbound A2A `Message` contains one or more text parts and the graph state includes a `messages`
   property, Aion Server appends a LangChain `HumanMessage` derived from the A2A text.
   The default policy concatenates all A2A text parts in order into a single `HumanMessage`.
2. Populate `AionRuntimeContext` (invocation-scoped context):
   Aion Server populates the `AionRuntimeContext` for the current invocation with `event`, `inbox`, and
   the parsed Distribution extension payload when one is present. `Thread` and `Message` are authoring
   helpers derived from that context.
3. Idempotency and dedupe:
   If the inbound A2A `messageId` has already been ingested for the current `contextId`,
   Aion Server does not append a duplicate `HumanMessage`.

For messaging media, the ordered text policy preserves provider caption before generated transcript, but model state
does not replace the protocol envelope. Authors that need files, transcript provenance, expiry, or transcription
status must inspect every part in `AionRuntimeContext.inbox.message.parts`. The file part carries the schema-marked
`MessageMediaPayload`, and its `mediaId` associates any transcript. Its `fileId` names the stable Aion File Recording;
`fileVersionId` names the exact immutable Recordable. Do not read only the first text part.

Register `AionRuntimeContext` as the graph's context schema:

```python theme={null}
from typing import TypedDict, Annotated, Optional

from aion.core.runtime import AionRuntimeContext
from aion.core.a2a import A2AOutbox
from langchain_core.messages import BaseMessage
from langgraph.graph import StateGraph, add_messages


class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    a2a_outbox: Optional[A2AOutbox]


builder = StateGraph(AgentState, context_schema=AionRuntimeContext)
```

Access inbound data inside a node via `Runtime[AionRuntimeContext]`:

```python theme={null}
from langgraph.runtime import Runtime

from aion.langgraph.authoring import Thread


async def my_node(state: AgentState, runtime: Runtime[AionRuntimeContext]) -> dict:
    context = runtime.context
    thread = Thread.from_context(context)  # normalized thread / conversation context
    message = thread.message              # normalized inbound message (or None)
    event = context.event                 # event kind: message, command, reaction, card action
    inbox = context.inbox                 # raw A2AInbox escape hatch (task, message, metadata)
    return {}
```

### 1.3 Runtime Context vs Graph State

Graph state carries the model-facing transcript. Transport metadata, routing, and the
normalized event stay on the invocation-scoped context, so a node that needs the protocol
objects reads them off `runtime.context` instead of declaring them in the state schema:

```python theme={null}
from typing import Annotated, TypedDict

from langchain_core.messages import AnyMessage
from langgraph.graph import add_messages
from langgraph.runtime import Runtime

from aion.core.runtime import AionRuntimeContext
from aion.langgraph.authoring import Thread


class State(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]


async def node(state: State, runtime: Runtime[AionRuntimeContext]) -> dict:
    thread = Thread.from_context(runtime.context)
    inbound = thread.message

    if inbound and inbound.text == "/help":
        await thread.reply("Here is what I can do.")
        return {}

    return {}
```

`a2a_outbox` is the one protocol object that does belong in the state schema: it is a value
the graph returns rather than one it receives (Section 2.1).

## 2. Outbound Messages

### 2.1 `SendMessage` -> `graph.astream()`

Valid responses to an A2A `SendMessage` call are a `Task`.

Aion Server constructs the response using this precedence:

**(1) SDK-managed response buffer (authoritative when populated)**

The runtime maintains a request-scoped messaging buffer for the current turn.
SDK helpers and other higher-level response surfaces should write to that
buffer first. This includes streamed output that is intended to become the
durable reply.

When this buffer is non-empty, it is the authoritative source for A2A
response compilation.

**(2) `a2a_outbox`**

If the returned dictionary contains `a2a_outbox`, it must be an `A2AOutbox` instance wrapping either
a `Task` or a `Message`. Server-owned fields are enforced:

* `task_id` and `context_id` are set to current values managed by Aion Server.
* Metadata keys beginning with `aion:` or `https://docs.aion.to` are reserved for the
  platform. `aion:network` carries routing and identity, `aion:ephemeral` decides whether an
  event is persisted, and the extension URIs address extension payloads. Ask for that behaviour
  through the typed parameter that produces it — `emit_message(..., ephemeral=True)`,
  `routing=...` — rather than by writing the key.

Behavior:

* If `a2a_outbox.message` is set, append it to current task history.
* If `a2a_outbox.task` is set, treat it as a patch to the server task:
  server merges or extends history and artifacts; graph-provided metadata merges shallowly.
  Reserved keys are dropped from the patch — they reach neither the stored task nor the wire.

```python theme={null}
from a2a.types import Message, Task, Part, Role
from aion.core.a2a import A2AOutbox


# Option 1: return a Message
async def reply_node(state: AgentState) -> dict:
    message = Message(
        message_id=str(uuid.uuid4()),
        role=Role.ROLE_AGENT,
        parts=[Part(text="Done!")],
    )
    return {"a2a_outbox": A2AOutbox(message=message)}


# Option 2: return a Task (patch)
async def task_node(state: AgentState) -> dict:
    task = Task(
        id=task_id,
        context_id=context_id,
        status=TaskStatus(state=TaskState.TASK_STATE_WORKING),
        history=[...],
        artifacts=[...],
        metadata={"my_key": "my_value"},
    )
    return {"a2a_outbox": A2AOutbox(task=task)}
```

Aion Server also keeps `state.messages` in sync by appending an `AIMessage` and/or `ToolMessage`
derived from the outbound A2A payload. Linkage: `AIMessage.id = a2a.taskId`.

**(3) Framework-native fallback**

If neither the SDK-managed response buffer nor `a2a_outbox` exists, Aion
falls back to framework-native output for the current turn:

* first, accumulated streamed `messages` chunks
* then, if needed, the last agent-authored `AIMessage` in `state.messages`

The fallback path produces a `Task` with a terminal status. If a developer needs
full control over the task shape, status events, or artifact updates, they should
use `a2a_outbox` instead of relying on fallback inference.

### 2.2 Replying into the Inbound Context

`thread.reply(...)` builds its routing from the inbound event context, so the reply returns
to the conversation the request arrived from — the same DM, the same thread, the same shared
conversation — without the graph reconstructing that target. A node that needs to send
somewhere else passes an explicit target to `thread.post(...)`.

## 3. Streaming

### 3.1 `SendStreamingMessage` -> `graph.astream()`

Valid responses to an A2A `SendStreamingMessage` call are a `Task`,
`TaskStatusUpdateEvent`, or `TaskArtifactUpdateEvent` in the following
sequence:

```mermaid theme={null}
  flowchart LR
      A["SendStreamingMessage"] --> C
      C["Task(working)"] --> D["(StatusUpdateEvent | ArtifactUpdateEvent)*"]
      D --> E["Task(terminal)"]
      E --> Close

```

`aion.langgraph.server` will first dispatch a `Task` with a `"working"` status,
followed by one or more `TaskStatusUpdateEvent` or `TaskArtifactUpdateEvent`
as the stream progresses, and finally a `Task` with a terminal status.

Aion Server requests LangGraph stream updates using:
`stream_mode=["values", "messages", "custom", "updates"]`. In the following sections, we will discuss how each
Langgraph event type is mapped to A2A message stream response events.

### 3.2 Event Type: `values`

The last `values` payload in the stream represents the final output and state snapshot.
Aion Server uses it to update task state and determine the final terminal response if one has not already been sent.

Output mapping follows the same precedence as Section 2.1.

If neither the SDK-managed response buffer nor `a2a_outbox` is populated,
Aion Server may construct an A2A `Message` using accumulated streamed deltas
collected in the `"aion:stream-delta"` artifact via `messages` mode
(see Section 3.3).

### 3.3 Event Type: `messages`

`messages` stream mode yields LLM output chunks as `(message_chunk, metadata)`.
These events are not diffs to `state.messages`.

Multiple LLM invocations in a graph can produce `messages` events.

To bridge this to A2A, chunks are appended into a transitory streaming artifact:

* `artifact.name = "Stream Delta"`
* `artifact.id = "aion:stream-delta"`
* `append=false` on the first chunk, `append=true` on every later chunk
* the artifact update's own `metadata` carries the messaging extension's schema marker,
  `…#StreamDeltaPayload` — see [Streaming Delta Semantics](/sdk/python/streaming/stream-delta)
* `lastChunk=false`
* every update carries content; chunks with empty content produce no event
* the sequence ends with the next durable message or with the terminal task status

A `TaskArtifactUpdateEvent` is emitted for each chunk.
This artifact is transitory and is not persisted to the task's durable state by default.

### 3.4 Event Type: `custom`

The Aion SDK provides helper functions via LangGraph `StreamWriter` to emit custom events during graph execution.
Aion Server listens for these custom payloads and forwards them as A2A events, enforcing canonical
`taskId` and `contextId`.

Precedence rule: explicit A2A streaming events emitted via `custom` are authoritative.

Custom payloads are mapped into A2A streaming events, including status updates and artifact updates.
When helper APIs are used, Aion Server applies the same canonical ID and metadata enforcement rules.

For helper APIs, parameter semantics, and usage examples, see
[Streaming API](/sdk/langgraph/streaming-api).

### 3.5 Event Type: `updates`

Used to track the currently executing node.
Aion Server extracts the node name and updates execution context accordingly.

## 4. Summary of Responsibilities

### LangGraph Graph Author

* Register `AionRuntimeContext` as the graph's context schema via `context_schema=AionRuntimeContext`.
* Access inbound message, thread, event, and raw inbox data through `Runtime[AionRuntimeContext]` in node signatures.
* Keep `state.messages` as LangChain message types.
* Prefer SDK helpers when you want to populate the shared runtime response
  buffer directly.
* Optionally set `a2a_outbox` for full-fidelity A2A responses.
* For streaming, optionally emit A2A-native events via `custom` using SDK helper functions.

### Aion Server Adapter

* Own canonical IDs and routing metadata.
* Ensure idempotency on ingress.
* Map LangGraph output and state into A2A `Message` and `Task`.
* Stream A2A events as `StreamResponse` wrappers.

## Related Pages

* [Integration Patterns](/sdk/langgraph/guides/integration-patterns)
* [AionRuntimeContext](/sdk/langgraph/api/runtime-context)
* [Thread](/sdk/langgraph/api/thread)
* [Message](/sdk/langgraph/api/message)
* [Event Handlers](/sdk/langgraph/api/event-handlers)
* [LangGraph Streaming API](/sdk/langgraph/streaming-api)
* [Media and attachments](/docs/distributions/messaging/media-and-attachments)