This page describes the Thread type from aion-authoring-langgraph.
The Thread object is the main fluent surface for replying, posting, looking
up history, and emitting streaming intents without forcing authors to build
raw A2A messages by hand.
Properties
| Property | Purpose |
|---|
context_id | Current messaging thread or context identifier |
parent_context_id | Parent thread or context identifier when one exists |
network | Originating network or distribution identifier |
default_reply_target | Canonical reply target derived from the inbound event |
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) | Create an explicit outbound post action |
history(limit=20, offset=None) | Request recent conversation history through the control plane |
typing(content, *, metadata=None) | Emit a stream-only typing or progress intent |
reply(...) and post(...) accept the same content categories:
reply(...)
reply(...) is intended for the most common case: answer in the same place
the inbound message arrived.
from langchain_core.messages import AIMessage, AIMessageChunk
from aion.core.a2a import data_artifact, file_artifact, url_artifact
from aion.core.agent.invocation.card import Card
async def node(state: State, runtime: Runtime[AionRuntimeContext]) -> dict:
thread = Thread.from_context(runtime.context)
# Plain text
await thread.reply("Thanks. I'm looking into that now.")
# LangChain message types
await thread.reply(AIMessage(content="Done."))
await thread.reply(AIMessageChunk(content="Streaming chunk..."))
# Async iterator — streamed live, accumulated as a durable reply
async def generate():
for word in ["Processing", " your", " request..."]:
yield word
await thread.reply(generate())
# Card — see /sdk/python/messaging/card for all creation modes
await thread.reply(Card("Report ready").add(Text("All checks passed.")))
# Artifacts — see /sdk/python/messaging/artifacts for the full reference
await thread.reply(url_artifact("https://example.com/report.pdf", mime_type="application/pdf", name="report"))
await thread.reply(file_artifact(pdf_bytes, mime_type="application/pdf", name="report"))
await thread.reply(data_artifact({"score": 0.95}, name="result"))
return {}
post(...)
post(...) is intended for explicit outbound actions that should remain
distinct from the default reply target.
async def node(state: State, runtime: Runtime[AionRuntimeContext]) -> dict:
thread = Thread.from_context(runtime.context)
await thread.post("Posting a follow-up note to a different channel.")
return {}
reply(...), post(...) should remain explicit in the response buffer
so cross-target delivery is never inferred from transcript order alone.
history(...)
history(...) is intended to ask the control plane for more context than was
included on the inbound turn.
async def node(state: State, runtime: Runtime[AionRuntimeContext]) -> dict:
thread = Thread.from_context(runtime.context)
recent = await thread.history(limit=5)
return {}
history() is not yet fully implemented and currently returns an empty list.
typing(...)
typing(content, *, metadata=None) is intended for stream-only live intents.
content is a required string — the transient text shown to the user while the
agent is working.
async def node(state: State, runtime: Runtime[AionRuntimeContext]) -> dict:
thread = Thread.from_context(runtime.context)
await thread.typing("Thinking...")
return {}
Related Pages
