This page describes the Thread type from aion-authoring-adk.
The Thread object is the main fluent surface for replying, posting, and
emitting streaming intents without forcing authors to build raw A2A messages
by hand.
Import
from aion.adk.authoring.invocation import Thread, AionInvocationContext
Constructing a Thread
Create a Thread from the Aion runtime context available on the ADK invocation
context. Inside _run_async_impl, type ctx as AionInvocationContext to get
IDE support for the Aion-specific fields:
from aion.adk.authoring.invocation import Thread, AionInvocationContext
async def _run_async_impl(self, ctx: AionInvocationContext):
thread = Thread.from_context(ctx.aion_runtime_context)
await thread.reply("Hello!")
ctx.aion_runtime_context is an AionRuntimeContext injected by Aion Server
into every ADK invocation. It is None when the agent runs outside of Aion.
See Invocation Context for the full ctx field reference.
Properties
| Property | Type | Purpose |
|---|
context_id | str | None | Current messaging thread or context identifier |
parent_context_id | str | None | Parent thread or context identifier when one exists |
network | str | Originating network or distribution identifier |
default_reply_target | str | None | Canonical reply target derived from the inbound event |
message | Message | None | Inbound Message bound to this thread, or None for task-only turns |
Methods
| Method | Purpose |
|---|
from_context(context) | Class method. Create a Thread from an AionRuntimeContext |
reply(content, *, metadata=None) | Add a durable reply to the current thread |
post(content, *, target=None, metadata=None) | Post a message with explicit routing control |
typing(content, *, metadata=None) | Emit an ephemeral typing or progress indicator |
history(limit=20, offset=None) | Request recent conversation history |
reply(...)
reply(...) is the most common way to respond. It automatically routes the
reply back to the same place the inbound message arrived.
from aion.adk.authoring.invocation import Thread, AionInvocationContext
async def _run_async_impl(self, ctx: AionInvocationContext):
thread = Thread.from_context(ctx.aion_runtime_context)
await thread.reply("Thanks, I'm looking into that now.")
content can be:
- a
str — sent as a text message
- a
Card — sent as a provider-neutral card document
- an
a2a.types.Artifact — saved via the artifact service and emitted as a TaskArtifactUpdateEvent
- an
Event — forwarded directly to the ADK event stream
- an async iterator of
str chunks — streamed as partial events, then
emitted as one durable final event
When content is an async iterator, the SDK emits partial events during
streaming and accumulates a durable final reply in the same invocation.
from aion.adk.authoring.invocation import Thread, AionInvocationContext
from aion.core.agent.invocation.card import Card
async def _run_async_impl(self, ctx: AionInvocationContext):
thread = Thread.from_context(ctx.aion_runtime_context)
await thread.reply(Card("Deployment approved"))
post(...)
post(...) gives explicit control over where the message is delivered. Use it
when you want to route output to a different target than the default reply.
from aion.adk.authoring.invocation import Thread, AionInvocationContext
async def _run_async_impl(self, ctx: AionInvocationContext):
thread = Thread.from_context(ctx.aion_runtime_context)
await thread.post("Follow-up sent to a different context.")
| Parameter | Type | Description |
|---|
content | str | Card | Artifact | Event | async iterator | Message content |
target | MessageActionPayload | None | Explicit delivery target; uses default reply target when omitted |
metadata | dict | None | Optional metadata forwarded to A2A Message.metadata. Keys must not start with aion: (reserved for service use) |
post(...) accepts the same content categories as reply(...).
typing(...)
typing(...) emits an ephemeral event that is shown in real time and never
persisted in task history. Use it for progress indicators or intermediate status
messages.
from aion.adk.authoring.invocation import Thread, AionInvocationContext
async def _run_async_impl(self, ctx: AionInvocationContext):
thread = Thread.from_context(ctx.aion_runtime_context)
await thread.typing("Searching the knowledge base...")
# ... do work ...
await thread.reply("Here is what I found.")
| Parameter | Type | Description |
|---|
content | str | Ephemeral status text displayed during processing |
metadata | dict | None | Optional metadata forwarded to A2A Message.metadata. Keys must not start with aion: (reserved for service use) |
history(...)
history(...) requests recent conversation history from the control plane.
from aion.adk.authoring.invocation import Thread, AionInvocationContext
async def _run_async_impl(self, ctx: AionInvocationContext):
thread = Thread.from_context(ctx.aion_runtime_context)
recent = await thread.history(limit=5)
| Parameter | Type | Description |
|---|
limit | int | Maximum number of messages to return (default: 20) |
offset | int | None | Number of messages to skip for pagination |
history() is not yet fully implemented and currently returns an empty list.
Related Pages
