Skip to main content
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

HelperEmitted A2A EventBehavior
emit_artifactTaskArtifactUpdateEventEmits a pre-built artifact, including file and data artifacts.
emit_cardTaskStatusUpdateEventEmits a card message with optional routing.
emit_messageTaskStatusUpdateEvent or TaskArtifactUpdateEventEmits full messages, stream chunks, or ephemeral messages.
emit_reactionA2A ReactionActionPayloadEmits 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.
ParameterDescription
writerLangGraph StreamWriter from node signature
artifactPre-built a2a.types.Artifact to emit
routingOptional MessageActionPayload delivery routing target
appendSet true to append to a previously sent artifact
is_last_chunkSet false if more chunks are coming
metadataOptional 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.
ParameterDescription
writerLangGraph StreamWriter from node signature
cardCard instance, either inline JSX, remote URL, or builder-created JSX
routingOptional MessageActionPayload delivery routing target
metadataOptional 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.
ParameterDescription
writerLangGraph StreamWriter from node signature
messageLangChain AIMessage or AIMessageChunk
ephemeralIf true, event is sent to client but not persisted in task history
routingOptional MessageActionPayload for explicit delivery routing
metadataOptional 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.
ParameterDescription
writerLangGraph StreamWriter from node signature
payloadReactionActionPayload 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