Streaming API

The streaming helpers are the low-level alternative to using Thread and Message. They let you emit messages, cards, artifacts, and reactions directly when you already have access to the ADK event emitter. Most agents should use Thread and Message instead. Reach for these functions when you are outside an invocation method that provides a Thread, or when you need fine-grained control over routing.

Import

from aion.adk.authoring import (
    emit_message,
    emit_card,
    emit_artifact,
    emit_reaction,
)

All four functions require an EventEmitter — the callable that pushes ADK events into the current invocation stream. Retrieve it with get_adk_emitter():

from aion.adk.authoring.invocation import get_adk_emitter

emitter = get_adk_emitter()

get_adk_emitter() returns None when called outside an active invocation context. Always guard before using it.

`emit_message`

emit_message(
    emitter: EventEmitter,
    text: str,
    *,
    routing: MessageActionPayload | None = None,
    ephemeral: bool = False,
    metadata: dict | None = None,
) -> None

Emit a text message during agent execution.

Parameter	Type	Description
`emitter`	`EventEmitter`	ADK event emitter from the current invocation
`text`	`str`	Text content to emit
`routing`	`MessageActionPayload \| None`	Optional delivery routing target
`ephemeral`	`bool`	When `True`, routes to `EPHEMERAL_MESSAGE` — shown in real time, never persisted in task history
`metadata`	`dict \| None`	Optional metadata forwarded to A2A `Message.metadata`. Keys must not start with `aion:` (reserved)

from aion.adk.authoring import emit_message
from aion.adk.authoring.invocation import get_adk_emitter


emitter = get_adk_emitter()
if emitter:
    emit_message(emitter, "Processing your request...")
    emit_message(emitter, "Searching...", ephemeral=True)
    emit_message(emitter, "Done", metadata={"trace_id": "abc"})

`emit_card`

emit_card(
    emitter: EventEmitter,
    card: Card,
    *,
    routing: MessageActionPayload | None = None,
    metadata: dict | None = None,
) -> None

Emit a card message during agent execution.

Parameter	Type	Description
`emitter`	`EventEmitter`	ADK event emitter from the current invocation
`card`	`Card`	Card instance (JSX or URL)
`routing`	`MessageActionPayload \| None`	Optional delivery routing target
`metadata`	`dict \| None`	Optional metadata forwarded to A2A `Message.metadata`. Keys must not start with `aion:` (reserved)

from aion.adk.authoring import emit_card
from aion.adk.authoring.invocation import get_adk_emitter
from aion.core.agent.invocation.card import Card

emitter = get_adk_emitter()
if emitter:
    emit_card(emitter, Card(jsx="<Card><Text>Deployment approved</Text></Card>"))

See Card for all creation modes.

`emit_artifact`

async emit_artifact(
    emitter: EventEmitter,
    ctx: AionInvocationContext,
    artifact: Artifact,
    *,
    routing: MessageActionPayload | None = None,
    metadata: dict | None = None,
) -> None

Emit a pre-built A2A artifact during agent execution. The function saves the artifact via ctx.artifact_service and then emits an ADK event with EventActions(artifact_delta=...).

Parameter	Type	Description
`emitter`	`EventEmitter`	ADK event emitter from the current invocation
`ctx`	`AionInvocationContext`	ADK invocation context providing `artifact_service`
`artifact`	`a2a.types.Artifact`	Pre-built artifact to emit
`routing`	`MessageActionPayload \| None`	Optional delivery routing target
`metadata`	`dict \| None`	Optional metadata merged into A2A `Artifact.metadata`. Keys must not start with `aion:` (reserved)

from aion.adk.authoring import emit_artifact
from aion.adk.authoring.invocation import AionInvocationContext, get_adk_emitter
from aion.core.a2a import url_artifact


async def _run_async_impl(self, ctx: AionInvocationContext):
    emitter = get_adk_emitter()
    if emitter:
        await emit_artifact(
            emitter, ctx,
            url_artifact("https://example.com/report.pdf", mime_type="application/pdf"),
        )

emit_artifact does not support multi-part artifacts. If the artifact contains more than one part, the call is a no-op and a warning is logged. Build single-part artifacts when using this function.

`emit_reaction`

emit_reaction(
    emitter: EventEmitter,
    payload: ReactionActionPayload,
) -> None

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

Parameter	Type	Description
`emitter`	`EventEmitter`	ADK event emitter from the current invocation
`payload`	`ReactionActionPayload`	Reaction action to perform

from aion.adk.authoring import emit_reaction
from aion.adk.authoring.invocation import get_adk_emitter
from aion.core.a2a.extensions.messaging import ReactionActionPayload

emitter = get_adk_emitter()
if emitter:
    emit_reaction(emitter, ReactionActionPayload(
        context_id="C06ROOM123",
        message_id="1728162300.551219",
        reaction_key="thumbsup",
        operation="add",
    ))

For message-level reactions prefer message.react(...) — it reads context_id and message_id from the inbound event automatically.

Python SDK

LangGraph

Google ADK

Import

`emit_message`

`emit_card`

`emit_artifact`

`emit_reaction`

​Import

​emit_message

​emit_card

​emit_artifact

​emit_reaction

​Related Pages

Import

`emit_message`

`emit_card`

`emit_artifact`

`emit_reaction`

Related Pages