Thread

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"))

See Card for all creation modes and component reference.

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

Python SDK

LangGraph

Google ADK

Import

Constructing a Thread

Properties

Methods

`reply(...)`

`post(...)`

`typing(...)`

`history(...)`

​Import

​Constructing a Thread

​Properties

​Methods

​reply(...)

​post(...)

​typing(...)

​history(...)

​Related Pages

Import

Constructing a Thread

Properties

Methods

`reply(...)`

`post(...)`

`typing(...)`

`history(...)`

Related Pages