> ## 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.

# LangGraph Streaming API

> Custom stream-mode helper APIs for emitting A2A events from LangGraph.

This page documents the lower-level streaming helpers in
`aion-authoring-langgraph`.

Use these helpers when a LangGraph node needs direct control over emitted
messages, cards, artifacts, or reactions. For ordinary reply and post authoring,
the higher-level `Thread` and `Message` APIs are the intended fluent surface.

These helpers are intended for use with streaming requests where Aion Server
consumes `stream_mode=["values", "messages", "custom", "updates"]`.

## Event Helpers

| Helper          | Emitted A2A Event                                    | Behavior                                                       |
| --------------- | ---------------------------------------------------- | -------------------------------------------------------------- |
| `emit_artifact` | `TaskArtifactUpdateEvent`                            | Emits a pre-built artifact, including file and data artifacts. |
| `emit_card`     | `TaskStatusUpdateEvent`                              | Emits a card message with optional routing.                    |
| `emit_message`  | `TaskStatusUpdateEvent` or `TaskArtifactUpdateEvent` | Emits full messages, stream chunks, or ephemeral messages.     |
| `emit_reaction` | A2A `ReactionActionPayload`                          | Emits a reaction on an existing provider message.              |

## Functions

### `emit_artifact(...)`

```python theme={null}
emit_artifact(
    writer,
    artifact,
    *,
    routing=None,
    append=False,
    is_last_chunk=True,
    metadata=None,
)
```

Emits a pre-built `a2a.types.Artifact` during graph execution.

Use the framework-agnostic builders from `aion.core.a2a` (`url_artifact`,
`file_artifact`, `data_artifact`) to construct the artifact before emitting it.

| Parameter       | Description                                                                                             |
| --------------- | ------------------------------------------------------------------------------------------------------- |
| `writer`        | LangGraph `StreamWriter` from node signature                                                            |
| `artifact`      | Pre-built `a2a.types.Artifact` to emit                                                                  |
| `routing`       | Optional `MessageActionPayload` delivery routing target                                                 |
| `append`        | Set `true` to append to a previously sent artifact                                                      |
| `is_last_chunk` | Set `false` if more chunks are coming                                                                   |
| `metadata`      | Optional user-defined metadata merged into `Artifact.metadata`; keys starting with `aion:` are reserved |

Use cases: generated PDFs or images, chunked file streaming, external file
references, structured data payloads.

Example:

```python theme={null}
from langgraph.types import StreamWriter

from aion.core.a2a import data_artifact, file_artifact, url_artifact
from aion.langgraph.authoring import emit_artifact


def my_node(state: dict, writer: StreamWriter):
    # Remote file by URL
    emit_artifact(
        writer,
        url_artifact("https://example.com/report.pdf", mime_type="application/pdf"),
    )
    # Inline bytes
    emit_artifact(
        writer,
        file_artifact(pdf_bytes, mime_type="application/pdf", name="report"),
    )
    # Structured data
    emit_artifact(
        writer,
        data_artifact({"score": 42}, name="result"),
        metadata={"owner": "agent-x"},
    )
    return state
```

### `emit_card(...)`

```python theme={null}
emit_card(writer, card, *, routing=None, metadata=None)
```

Emits a card message during graph execution.

Produces a `TaskStatusUpdateEvent` whose message contains a card file part and
`extensions=[CardsURI]`. When `routing` is provided, a `DataPart` with
`MessageActionPayload` is also attached so the distribution delivers the card
to the correct channel.

| Parameter  | Description                                                            |
| ---------- | ---------------------------------------------------------------------- |
| `writer`   | LangGraph `StreamWriter` from node signature                           |
| `card`     | `Card` instance, either inline JSX, remote URL, or builder-created JSX |
| `routing`  | Optional `MessageActionPayload` delivery routing target                |
| `metadata` | Optional user-defined metadata forwarded to A2A `Message.metadata`     |

Example:

```python theme={null}
from langgraph.types import StreamWriter

from aion.core.agent.invocation.card import (
    Actions,
    Button,
    Card,
    Divider,
    Field,
    Fields,
    Text,
)
from aion.langgraph.authoring import emit_card


def my_node(state: dict, writer: StreamWriter):
    # Inline JSX
    emit_card(writer, Card(jsx="<Card><Text>Hello</Text></Card>"))

    # Remote URL
    emit_card(writer, Card(url="https://cdn.example.com/cards/welcome.jsx"))

    # Builder — compose from typed components
    card = (
        Card("Deployment approved")
        .add(Text("Production rollout is ready."))
        .add(Fields().add(Field("Env", "prod")).add(Field("Run", "#42")))
        .add(Divider())
        .add(Actions().add(Button("Approve", id="approve", style="primary")))
    )
    emit_card(writer, card, metadata={"template": "deploy-approval"})

    return state
```

### `emit_message(...)`

```python theme={null}
emit_message(writer, message, ephemeral=False, routing=None, metadata=None)
```

Emits a programmatic message during graph execution. Supports full messages and
streaming chunks.

| Parameter   | Description                                                                                                 |
| ----------- | ----------------------------------------------------------------------------------------------------------- |
| `writer`    | LangGraph `StreamWriter` from node signature                                                                |
| `message`   | LangChain `AIMessage` or `AIMessageChunk`                                                                   |
| `ephemeral` | If `true`, event is sent to client but not persisted in task history                                        |
| `routing`   | Optional `MessageActionPayload` for explicit delivery routing                                               |
| `metadata`  | Optional user-defined metadata forwarded to A2A `Message.metadata`; keys starting with `aion:` are reserved |

`ephemeral=False` (default):

* `AIMessage` -> `TaskStatusUpdateEvent(working, message=...)`; persisted in history.
* `AIMessageChunk` -> `TaskArtifactUpdateEvent(STREAM_DELTA)`; streamed and not persisted.

`ephemeral=True`:

* `AIMessage` or `AIMessageChunk` -> `TaskArtifactUpdateEvent(EPHEMERAL_MESSAGE)`.
* Emitted to client and filtered out by task store.
* Does not change durable response fallback behavior.

Use cases: progress notifications, thinking indicators, and transient status
events.

Example:

```python theme={null}
from langchain_core.messages import AIMessage, AIMessageChunk
from langgraph.types import StreamWriter

from aion.langgraph.authoring import emit_message


def my_node(state: dict, writer: StreamWriter):
    emit_message(writer, AIMessageChunk(content="Processing..."))
    emit_message(
        writer,
        AIMessage(content="Searching knowledge base..."),
        ephemeral=True,
    )
    return state
```

### `emit_reaction(...)`

```python theme={null}
emit_reaction(writer, payload)
```

Emits a reaction action on an existing provider message. Instructs the
distribution to add or remove a reaction.

When reacting to the current inbound message, prefer `Message.react(...)`; it
derives the context and message identifiers from the runtime context.

| Parameter | Description                                   |
| --------- | --------------------------------------------- |
| `writer`  | LangGraph `StreamWriter` from node signature  |
| `payload` | `ReactionActionPayload` with reaction details |

The `ReactionActionPayload` must include:

* `context_id`: Provider-native conversation or thread identifier — take from `context.event.payload.context_id`
* `message_id`: Provider-native message identifier to react to — take from `context.event.payload.message_id`
* `reaction_key`: Reaction identifier, such as `thumbsup` or `heart`
* `operation`: `"add"` or `"remove"`
* `display_value`: Optional human-readable reaction text

Use cases: automated reactions to messages, conditional emoji responses, approval workflows.

Example:

```python theme={null}
from aion.core.a2a.extensions.messaging import ReactionActionPayload
from aion.core.runtime import AionRuntimeContext
from aion.langgraph.authoring.invocation import emit_reaction
from langgraph.runtime import Runtime
from langgraph.types import StreamWriter


def my_node(state: dict, writer: StreamWriter, runtime: Runtime[AionRuntimeContext]):
    payload = runtime.context.event.payload
    emit_reaction(
        writer,
        ReactionActionPayload(
            context_id=payload.context_id,
            message_id=payload.message_id,
            reaction_key="thumbsup",
            operation="add",
            display_value="Good point",
        ),
    )
    return state
```

## Related Pages

* [Artifact Builders](/sdk/python/messaging/artifacts) — `url_artifact`, `file_artifact`, `data_artifact` reference
* [Card](/sdk/python/messaging/card) — Card builder API reference
* [Thread](/sdk/langgraph/api/thread)
* [Message](/sdk/langgraph/api/message)
* [Event Handlers](/sdk/langgraph/api/event-handlers)
* [Message Mapping](/sdk/langgraph/message-mapping)