> ## 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 adapts A2A requests to Google ADK and maps ADK output back into A2A.

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

This page uses the v1 canonical JSON-RPC method names. Aion ingress may continue to accept legacy
slash-style aliases for compatibility.

## Overview

1. A distribution or client sends an A2A request into Aion Server.
2. Aion exposes the request through `ctx.aion_runtime_context.inbox`.
3. The agent yields events, optionally providing an explicit A2A outbox.
4. Aion resolves those events into a final A2A `Message` or `Task`.
5. The caller or distribution delivers the response into the original context.

```mermaid theme={null}
sequenceDiagram
participant Caller as Distribution or Client
participant Server as Aion Server
participant ADK as Google ADK Agent
participant Target as Original Context

Caller->>Server: SendMessage or SendStreamingMessage
Server->>ADK: Populate ctx.aion_runtime_context
ADK-->>Server: Yield event content or a2a_outbox
Server-->>Caller: Final A2A Message or Task
Caller-->>Target: Deliver reply in same DM, thread, or conversation
```

## 1. Inbound Messages

### 1.1 Agent Invocation

Both `SendMessage` and `SendStreamingMessage` use the same execution path: the agent's
`run_async()` is always driven as an async event stream.

* `SendMessage` (`blocking=true`) — collects all events and returns the final `Task`.
* `SendMessage` (`blocking=false`) — returns after the first event, continues processing in
  background with `status="working"`.
* `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.

### 1.2 Part Type Mapping

Inbound A2A message parts are transformed into ADK `Content` as follows:

| A2A Part         | ADK Representation                                                  |
| ---------------- | ------------------------------------------------------------------- |
| `Part(text=...)` | `types.Part(text=...)`                                              |
| `Part(raw=...)`  | `types.Part(inline_data=types.Blob(mime_type=..., data=...))`       |
| `Part(url=...)`  | `types.Part(file_data=types.FileData(mime_type=..., file_uri=...))` |
| `Part(data=...)` | `types.Part(text=json.dumps(data))`                                 |

MIME type resolution order for file parts: explicit `mime_type` attribute → guess from filename →
fallback to `application/octet-stream`.

Part conversion is additive, not authoritative. Messaging media keeps caption text, generated transcript text, a file
with `MessageMediaPayload` metadata, and normalized event data as separate ordered A2A parts. Inspect every part in
`ctx.aion_runtime_context.inbox.message.parts` when provenance matters, and associate file and transcript metadata by `mediaId`. A
file's `fileId` is the stable Aion File Recording and `fileVersionId` is the exact immutable Recordable. Do not assume
that the first text part is the whole request.

If the message contains no usable parts, the plain text input from the request is used as a fallback.

### 1.3 Accessing Inbound Context — `ctx.aion_runtime_context.inbox`

When an inbound A2A `Message` arrives, Aion Server makes it available through
`ctx.aion_runtime_context` on the invocation context:

```python theme={null}
from google.adk.agents import BaseAgent


class MyAgent(BaseAgent):
    async def _run_async_impl(self, ctx):
        inbox = ctx.aion_runtime_context.inbox
        task = inbox.task
        message = inbox.message
        metadata = inbox.metadata
```

`inbox` contains:

| Field      | Type      | Description                                                            |
| ---------- | --------- | ---------------------------------------------------------------------- |
| `task`     | `Task`    | The current A2A Task                                                   |
| `message`  | `Message` | The full inbound A2A Message, including non-text parts                 |
| `metadata` | `dict`    | `SendMessageRequest`-level metadata (distribution/network, trace info) |

### 1.4 Replying into the Inbound Context

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

```python theme={null}
from aion.adk.authoring.invocation import AionInvocationContext, Thread


async def _run_async_impl(self, ctx: AionInvocationContext):
    thread = Thread.from_context(ctx.aion_runtime_context)
    text = thread.message.text if thread.message else ""
    await thread.reply(f"Received: {text}")
```

## 2. Outbound Messages

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

Aion Server constructs the response using the following precedence:

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

The runtime maintains a request-scoped messaging buffer for the current turn.
SDK helpers and ordinary ADK event content may populate that buffer, including
partial stream output and final non-partial message content 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`

Set `a2a_outbox` in `event.actions.state_delta` to provide an explicit A2A response. It must be an
`A2AOutbox` instance wrapping either a `Message` or a `Task`:

```python theme={null}
from a2a.types import Message, Task, Part, Role
from aion.core.a2a import A2AOutbox
from google.adk.agents import BaseAgent
from google.adk.events import Event, EventActions


class MyAgent(BaseAgent):
    async def _run_async_impl(self, ctx):
        # Option 1: outbox as Message
        yield Event(
            author=self.name,
            actions=EventActions(state_delta={
                "a2a_outbox": A2AOutbox(message=Message(
                    role=Role.ROLE_AGENT,
                    parts=[Part(text="Done!")],
                ))
            })
        )

        # Option 2: outbox as Task (patch)
        yield Event(
            author=self.name,
            actions=EventActions(state_delta={
                "a2a_outbox": A2AOutbox(task=Task(
                    history=[...],
                    artifacts=[...],
                    metadata={"my_key": "my_value"},
                ))
            })
        )
```

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 to current Task history.
* If `a2a_outbox.task` is set → treat as a **patch** to the server's Task:
  server merges or extends `history` and `artifacts`; provided `metadata` merges shallowly.
  Reserved keys are dropped from the patch — they reach neither the stored Task nor the wire.

### (3) Framework-native fallback

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

* first, accumulated partial stream text
* then, if needed, the final non-partial agent-authored event content
* finally, deterministic final session/state inspection when the adapter
  exposes enough data to do so safely

> If you need to return a comprehensive A2A response (e.g., data parts, rich metadata, multiple
> artifacts), use `a2a_outbox` rather than relying on the streaming fallback.

### When to Reach for `a2a_outbox`

An ordinary final ADK event is enough for a plain text reply. Reach for `a2a_outbox` when:

* you need structured parts rather than plain text
* you want to emit a provider-neutral card payload
* you need to override the default outbound target
* you want to return a task patch instead of a single message

## 3. Summary

* Read `ctx.aion_runtime_context.inbox` to access the inbound A2A Task, Message, and metadata.
* Prefer SDK helpers or normal ADK event content when you want to populate the
  shared runtime response buffer.
* Optionally set `a2a_outbox` in `event.actions.state_delta` as an `A2AOutbox` instance for full-fidelity A2A responses.
* Yield partial events for real-time text streaming.

## Related Pages

* [Invocation Context](/sdk/google-adk/api/invocation-context)
* [Thread](/sdk/google-adk/api/thread)
* [Message](/sdk/google-adk/api/message)
* [Streaming API](/sdk/google-adk/api/streaming-api)
* [Client Events Reference](/sdk/google-adk/api/events-reference)
* [Media and attachments](/docs/distributions/messaging/media-and-attachments)