LangGraph Streaming API

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(...)`

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:

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(...)`

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:

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(...)`

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:

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(...)`

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:

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

Artifact Builders — url_artifact, file_artifact, data_artifact reference
Card — Card builder API reference
Thread
Message
Event Handlers
Message Mapping

Python SDK

LangGraph

Google ADK

LangGraph Streaming API

Event Helpers

Functions

`emit_artifact(...)`

`emit_card(...)`

`emit_message(...)`

`emit_reaction(...)`

​Event Helpers

​Functions

​emit_artifact(...)

​emit_card(...)

​emit_message(...)

​emit_reaction(...)

​Related Pages

Event Helpers

Functions

`emit_artifact(...)`

`emit_card(...)`

`emit_message(...)`

`emit_reaction(...)`

Related Pages