LangGraph Streaming API

This page documents the lower-level helpers that already exist in aion.langgraph today. Use these helpers when you need direct control over emitted TaskStatusUpdateEvent or TaskArtifactUpdateEvent objects. For ordinary reply and post authoring, the higher-level Thread and Message APIs described elsewhere in this section are the intended fluent surface. These helpers still participate in the same request-scoped response accumulator. They do not replace the standard precedence of response buffer, a2a_outbox, and framework-native fallback. 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_file_artifact`	`TaskArtifactUpdateEvent`	Emits a file artifact with a file part.
`emit_data_artifact`	`TaskArtifactUpdateEvent`	Emits a structured artifact with a data part.
`emit_message`	`TaskStatusUpdateEvent` or `TaskArtifactUpdateEvent`	Emits message or stream updates.
`emit_task_update`	`TaskStatusUpdateEvent`	Emits one combined task update.
`emit_reaction`	A2A ReactionActionPayload	Emits a reaction on an existing provider message.

Functions

`emit_file_artifact(...)`

emit_file_artifact(
    writer,
    *,
    url=None,
    data=None,
    mime_type,
    name=None,
    artifact_id=None,
    append=False,
    is_last_chunk=True,
)

Emits a file artifact and maps to an A2A Artifact with a file part.

Parameter	Description
`writer`	LangGraph `StreamWriter` from node signature
`url`	File URL for remote files; mutually exclusive with `data`
`data`	File content as bytes; mutually exclusive with `url`
`mime_type`	MIME type such as `application/pdf` or `image/png`
`name`	Artifact name (default: `file`)
`artifact_id`	Explicit artifact ID; auto-generated if not provided
`append`	Set `true` to append to a previously sent artifact
`is_last_chunk`	Set `false` if more chunks are coming

Use cases: generated PDFs or images, chunked file streaming, external file references. Example:

from uuid import uuid4

from langgraph.types import StreamWriter

from aion.langgraph import emit_file_artifact


def my_node(state: dict, writer: StreamWriter):
    emit_file_artifact(
        writer,
        url="https://example.com/report.pdf",
        mime_type="application/pdf",
        name="report",
    )

    artifact_id = str(uuid4())
    for i, file_bytes in enumerate(file_chunks):
        emit_file_artifact(
            writer,
            data=file_bytes,
            mime_type="text/plain",
            artifact_id=artifact_id,
            append=True,
            is_last_chunk=(i == len(file_chunks) - 1),
        )

    return state

`emit_data_artifact(...)`

emit_data_artifact(writer, data, name=None, artifact_id=None, append=False, is_last_chunk=True) Emits a structured data artifact. data must be JSON-serializable.

Parameter	Description
`writer`	LangGraph `StreamWriter` from node signature
`data`	Dictionary or any JSON-serializable value
`name`	Artifact name (default: `data`)
`artifact_id`	Explicit artifact ID; auto-generated if not provided
`append`	Set `true` to append to a previously sent artifact
`is_last_chunk`	Set `false` if more chunks are coming

Use cases: analysis outputs, metrics, and structured response payloads. Example:

from langgraph.types import StreamWriter

from aion.langgraph import emit_data_artifact


def my_node(state: dict, writer: StreamWriter):
    emit_data_artifact(
        writer,
        {"status": "success", "results": ["a", "b"]},
        name="analysis",
    )
    return state

`emit_message(...)`

emit_message(writer, message, ephemeral=False, routing=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; if not provided, uses default reply target

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:

from langchain_core.messages import AIMessage, AIMessageChunk
from langgraph.types import StreamWriter

from aion.langgraph 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_task_update(...)`

emit_task_update(writer, message=None, metadata=None) Emits one combined task update containing message and or metadata. Only AIMessage is accepted for message. For chunk streaming, use emit_message().

Parameter	Description
`writer`	LangGraph `StreamWriter` from node signature
`message`	Optional full message (`AIMessage` only)
`metadata`	Optional metadata dictionary to merge into task metadata

At least one of message or metadata must be provided. Keys with the aion: prefix in metadata are ignored. Use cases: step completion updates with metadata, progress plus message in one event. Example:

from langchain_core.messages import AIMessage
from langgraph.types import StreamWriter

from aion.langgraph import emit_task_update


def my_node(state: dict, writer: StreamWriter):
    emit_task_update(
        writer,
        message=AIMessage(content="Processing complete"),
        metadata={"progress": 100},
    )
    return state

`emit_reaction(...)`

emit_reaction(writer, payload) Emits a reaction action on an existing provider message. Instructs the distribution to add or remove a reaction.

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

The ReactionActionPayload must include:

context_id: Message context/thread identifier
message_id: Message identifier to react to
reaction_key: Reaction identifier (e.g., thumbsup, heart)
operation: "add" or "remove"
display_value: Optional human-readable reaction text

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

from aion.core.types.a2a.extensions.messaging import ReactionActionPayload
from langgraph.types import StreamWriter

from aion.langgraph import emit_reaction


def my_node(state: dict, writer: StreamWriter):
    emit_reaction(
        writer,
        ReactionActionPayload(
            context_id="C06ROOM123",
            message_id="1728162300.551219",
            reaction_key="thumbsup",
            operation="add",
            display_value="Good point",
        ),
    )
    return state

​Event Helpers

​Functions

​emit_file_artifact(...)

​emit_data_artifact(...)

​emit_message(...)

​emit_task_update(...)

​emit_reaction(...)