context.channel() – ChannelSession API Reference

A ChannelSession provides message I/O, user prompts, streaming text, and progress updates through the configured channel (console/Slack/…). It also manages continuation tokens to avoid race conditions.

---

Channel Resolution & Defaults

• Channel selection priority: explicit channel arg → session override (from context.channel(channel_key)) → bus default → console:stdin.
• Events are published after alias → canonical key resolution.

---

0. Channel Setup

context.channel(channel_key)

Set up a new ChannelSession for the current node context.

Parameters:

Name	Type	Description	Default
channel_key	str | None	An optional key to specify a particular channel.	None

Returns:

Name	Type	Description
ChannelSession		An instance representing the session for the specified channel.

Notes

Supported channel key formats include:

Channel Type	Format Example	Notes
Console	console:stdin	Console input/output
Slack	slack:team/{team_id}:chan/{channel_id}	Needs additional configuration
Telegram	tg:chat/{chat_id}	Needs additional configuration
UI Session	ui:session/{session_id}	Requires AG web UI
UI Run	ui:run/{run_id}	Requires AG web UI
Webhook	webhook:{unique_identifier}	For Slack, Discord, Zapier, etc.
File-based channel	file:path/to/directory	File system based channels

context.ui_run_channel()

Creates a new ChannelSession for the current node context with session key as ui:run/<run_id>.

This method is a convenience helper for the AG UI to get the default run channel.

Returns:

Name	Type	Description
ChannelSession	ChannelSession	The channel session associated with the current run.

context.ui_session_channel()

Creates a new ChannelSession for the current node context with session key as ui:session/<session_id>.

This method is a convenience helper for the AG UI to get the default session channel.

Returns:

Name	Type	Description
ChannelSession	ChannelSession	The channel session associated with the current session.

1. Send Methods

send(event, *, channel, ...)

Publish one outbound event after channel/meta normalization.

Ensure event.channel is set when missing and merge context metadata into event.meta before publishing through the channel bus.

Examples:

Publish a pre-constructed event:

event = OutEvent(type="agent.message", text="Hello!", channel=None)
await context.channel().send(event)

Resolve the channel when event.channel is empty:

await context.channel().send(event, channel="web:chat")

Parameters:

Name	Type	Description	Default
event	OutEvent	Event to publish.	required
channel	str | None	Optional fallback channel key used only when event.channel is unset.	None

Returns:

Name	Type	Description
None		Complete when the event is published.

Notes

Existing event.channel is preserved and not overridden by channel.

send_text(text, *, meta, channel, ...)

Send one plain-text message event and optionally log it to memory.

Log the text via memory facade when enabled, then publish OutEvent(type="agent.message") with merged metadata.

Examples:

Send a message:

await context.channel().send_text("Hello, world!")

Send to a specific channel with extra metadata:

await context.channel().send_text(
    "Status update.",
    meta={"priority": "high"},
    channel="web:chat"
)

Parameters:

Name	Type	Description	Default
text	str	Message body to send.	required
meta	dict[str, Any] | None	Optional outbound event metadata.	None
channel	str | None	Optional target channel key.	None
memory_log	bool	Enable chat-memory logging for this call.	True
memory_role	Literal['user', 'assistant', 'system', 'tool']	Role used for the memory record.	'assistant'
memory_tags	list[str] | None	Optional tags for the memory record.	None
memory_data	dict[str, Any] | None	Optional structured data for the memory record.	None
memory_severity	int	Severity value for memory logging.	2
memory_signal	float | None	Optional signal value for memory logging.	None

Returns:

Name	Type	Description
None		Complete when memory logging and publish steps finish.

Notes

Set adapter-specific display hints in meta (for example name or agent_id).

send_image(url, *, alt, title, channel, ...)

Send an image attachment via send_file() with image defaults.

Build image-oriented labels and filename, then delegate to send_file using artifact_kind="image".

Examples:

Send by URL:

await context.channel().send_image(
    url="https://example.com/image.png",
    alt="Sample image"
)

Send generated bytes:

await context.channel().send_image(
    file_bytes=b"...",
    alt="Generated image",
    title="Result"
)

Parameters:

Name	Type	Description	Default
url	str | None	Optional HTTP(S) URL or local path for the image source.	None
file_bytes	bytes | None	Optional image bytes. Preferred when both are provided.	None
alt	str	Alternate text used for fallback titling/filename.	'image'
title	str | None	Optional display title.	None
channel	str | None	Optional target channel key.	None
artifact_labels	dict[str, Any] | None	Optional extra artifact labels.	None
memory_log	bool	Enable chat-memory logging for this call.	True
memory_role	Literal['user', 'assistant', 'system', 'tool']	Role used for the memory record.	'assistant'
memory_tags	list[str] | None	Optional tags for the memory record.	None
memory_data	dict[str, Any] | None	Optional structured data for the memory record.	None
memory_severity	int	Severity value for memory logging.	2
memory_signal	float | None	Optional signal value for memory logging.	None

Returns:

Name	Type	Description
None		Complete when delegated file send finishes.

Notes

artifact_labels are merged with {"renderer": "image"}.

send_file(url, *, file_bytes, filename, title, channel, ...)

Send a file attachment message, optionally persisting it as an artifact.

Prefer artifact-backed payloads for file_bytes or local-path inputs. Fall back to a URL-only payload when artifact persistence is unavailable.

Examples:

Send by URL:

await context.channel().send_file(
    url="https://example.com/report.pdf",
    filename="report.pdf",
    title="Monthly Report"
)

Send from bytes:

await context.channel().send_file(
    file_bytes=b"binarydata...",
    filename="data.bin",
    title="Raw Data"
)

Parameters:

Name	Type	Description	Default
url	str | None	Optional source URL or local filesystem path.	None
file_bytes	bytes | None	Optional raw file bytes.	None
filename	str	Display filename for the chat attachment.	'file.bin'
title	str | None	Optional message text. Defaults to filename.	None
channel	str | None	Optional target channel key.	None
artifact_kind	str	Artifact kind when writing to artifact store.	'file'
artifact_labels	dict[str, Any] | None	Optional labels attached to persisted artifacts.	None
memory_log	bool	Enable chat-memory logging for this call.	True
memory_role	Literal['user', 'assistant', 'system', 'tool']	Role used for the memory record.	'assistant'
memory_tags	list[str] | None	Optional tags for the memory record.	None
memory_data	dict[str, Any] | None	Optional structured data for the memory record.	None
memory_severity	int	Severity value for memory logging.	2
memory_signal	float | None	Optional signal value for memory logging.	None

Returns:

Name	Type	Description
None		Complete when the outbound message event is published.

Notes

When both file_bytes and url are provided, bytes are attempted first.

send_buttons(text, buttons, *, meta, channel, ...)

Send a button prompt event and optionally log prompt text to memory.

Publish OutEvent(type="link.buttons") with button definitions and merged metadata for the resolved channel.

Examples:

Send a yes/no prompt:

from aethergraph import Button
await context.channel().send_buttons(
    "Choose an option:",
    [Button(label="Yes", value="yes"), Button(label="No", value="no")]
)

Send with custom metadata:

await context.channel().send_buttons(
    "Select your role:",
    [Button(label="Admin", value="admin"), Button(label="User", value="user")],
    meta={"priority": "high"},
    channel="web:chat"
)

Parameters:

Name	Type	Description	Default
text	str	Prompt text displayed above the buttons.	required
buttons	list[Button]	Button definitions to send.	required
meta	dict[str, Any] | None	Optional outbound event metadata.	None
channel	str | None	Optional target channel key.	None
memory_log	bool	Enable chat-memory logging for this call.	True
memory_role	Literal['user', 'assistant', 'system', 'tool']	Role used for the memory record.	'assistant'
memory_tags	list[str] | None	Optional tags for the memory record.	None
memory_data	dict[str, Any] | None	Optional structured data for the memory record.	None
memory_severity	int	Severity value for memory logging.	2
memory_signal	float | None	Optional signal value for memory logging.	None

Returns:

Name	Type	Description
None		Complete when memory logging and publish steps finish.

Notes

Button rendering/interaction semantics depend on the active adapter.

send_rich(text, *, rich, meta, channel, ...)

Send a message with an optional rich payload and optional memory logging.

Publish OutEvent(type="agent.message") with both text and rich fields, and record chat memory when enabled.

Examples:

Send one rich block:

await context.channel().send_rich(
    text="Here is the loss curve:",
    rich={
        "kind": "plot",
        "title": "Training loss",
        "payload": {"engine": "vega-lite", "spec": loss_vega_spec},
    },
)

Send multiple rich blocks:

await context.channel().send_rich(
    text="Training summary:",
    rich={"blocks": [{"kind": "plot"}, {"kind": "metrics"}]},
)

Parameters:

Name	Type	Description	Default
text	str | None	Optional plain text content.	None
rich	dict[str, Any] | None	Optional structured payload for UI-aware adapters.	None
meta	dict[str, Any] | None	Optional outbound event metadata.	None
channel	str | None	Optional target channel key.	None
memory_log	bool	Enable chat-memory logging for this call.	True
memory_role	Literal['user', 'assistant', 'system', 'tool']	Role used for the memory record.	'assistant'
memory_tags	list[str] | None	Optional tags for the memory record.	None
memory_data	dict[str, Any] | None	Optional structured data for the memory record.	None
memory_severity	int	Severity value for memory logging.	2
memory_signal	float | None	Optional signal value for memory logging.	None

Returns:

Name	Type	Description
None		Complete when memory logging and publish steps finish.

Notes

If text is None, memory logging records "[rich message]".

send_phase(phase, status, *, label, detail, code, channel, key_suffix)

Emit a phase-style progress update event.

Build a rich payload with phase metadata and publish it as agent.progress.update with a stable upsert key for this run/node.

Examples:

Send a pending phase update:

await context.channel().send_phase("routing", "pending")

Send an active phase update with details:

await context.channel().send_phase(
    "planning",
    "active",
    label="Planning Phase",
    detail="Calculating optimal routes",
)

Parameters:

Name	Type	Description	Default
phase	str	Logical phase identifier.	required
status	Literal['pending', 'active', 'done', 'failed', 'skipped']	Phase status value. One of "pending", "active", "done", "failed", or "skipped".	required
label	str | None	Optional display label. Defaults to phase.title().	None
detail	str | None	Optional detail text. Defaults to an empty string.	None
code	str | None	Optional code string in the rich payload.	None
channel	str | None	Optional target channel key.	None
key_suffix	str | None	Optional upsert-key suffix. Defaults to f"phase:{phase}".	None

Returns:

Name	Type	Description
None	None	Complete when the progress update is published.

Notes

• Payload shape is UI-oriented and adapters may render or ignore it differently.
• The AG UI will continue showing the phase block until a new message arrives or "done"/"failed"/"skipped" status is received for the same phase (identified by upsert key).

For the capability and tutorial of AG UI on rich content rendering sent by send_rich(), see Rich Card page for instruction.

2. Ask Methods

Ask method automatic implemented continuation so that the agent will resume running when triggered from external channels.

ask_text(prompt, *, timeout_s, silent, channel)

Prompt for user text and return normalized reply text.

Optionally log prompt/reply to memory, then use the continuation flow to await kind="user_input" and return payload["text"] as str.

Examples:

Ask a question:

reply = await context.channel().ask_text("What is your name?")

Use timeout and silent mode:

reply = await context.channel().ask_text(
    "Enter your feedback.",
    timeout_s=120,
    silent=True
)

Parameters:

Name	Type	Description	Default
prompt	str | None	Prompt text. None means wait without showing a prompt.	required
timeout_s	int	Continuation deadline in seconds.	3600
silent	bool	Back-compat flag forwarded in continuation payload.	False
channel	str | None	Optional target channel key.	None
memory_log_prompt	bool	Enable memory logging for the prompt.	True
memory_log_reply	bool	Enable memory logging for non-empty reply text.	True
memory_tags	list[str] | None	Optional tags applied to memory entries.	None

Returns:

Name	Type	Description
str	str	User text reply, or "" when absent.

Notes

Reply memory logging occurs only when returned text is non-empty.

wait_text(*, timeout_s, silent, channel)

Wait for a single text response from the user in a normalized format.

This method prompts the user for input (with no explicit prompt), waits for a reply, and returns the text. It automatically handles context metadata, timeout, and channel resolution.

Examples:

Basic usage to wait for user input:

reply = await context.channel().wait_text()

Waiting with a custom timeout and specific channel:

reply = await context.channel().wait_text(
    timeout_s=120,
    channel="web:chat"
)

Parameters:

Name	Type	Description	Default
timeout_s	int	Maximum time in seconds to wait for a response (default: 3600).	3600
channel	str | None	Optional explicit channel key to override the default or session-bound channel.	None
memory_log_reply	bool	Whether to log the user's reply to memory (default: True).	True
memory_tags	list[str] | None	Optional list of tags to associate with the memory log entry.	None

Returns:

Name	Type	Description
str	str	The user's text response, or an empty string if no input was received.

ask_approval(prompt, options, *, timeout_s, channel)

Prompt for a button-style approval choice and return normalized result.

Send an approval continuation with button labels, wait for the user's selected choice, then derive approved from whether the first option was selected (case-insensitive).

Examples:

Use default approve/reject options:

result = await context.channel().ask_approval("Do you approve this action?")
# result: { "approved": True/False, "choice": "Approve"/"Reject" }

Use custom options:

result = await context.channel().ask_approval(
    "Proceed with deployment?",
    options=["Yes", "No", "Defer"],
    timeout_s=120
)

Parameters:

Name	Type	Description	Default
prompt	str	Prompt text shown to the user.	required
options	Iterable[str]	Ordered button labels. The first option maps to approved.	('Approve', 'Reject')
timeout_s	int	Continuation deadline in seconds.	3600
channel	str | None	Optional target channel key.	None
memory_log_prompt	bool	Enable memory logging for the prompt.	True
memory_log_reply	bool	Enable memory logging for a selected choice.	True
memory_tags	list[str] | None	Optional tags applied to memory entries.	None

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: {"approved": bool, "choice": Any}.

Notes

If no choice is returned, or options is empty, approved is False.

ask_files(prompt, *, accept, multiple, timeout_s, channel)

Prompt for file upload and return normalized text/files payload.

Send a user_files continuation prompt, wait for response, and return a dictionary with text plus a list-valued files field.

Examples:

Ask for one or more files:

result = await context.channel().ask_files(
    prompt="Please upload your report."
)
# result: { "text": "...", "files": [FileRef(...), ...] }

Provide type hints for upload UI:

result = await context.channel().ask_files(
    prompt="Upload images for review.",
    accept=["image/png", ".jpg"],
    multiple=True
)

Parameters:

Name	Type	Description	Default
prompt	str	Prompt text shown with the file picker.	required
accept	list[str] | None	Optional MIME/extension hints for the adapter UI.	None
multiple	bool	Allow multiple selections when True.	True
timeout_s	int	Continuation deadline in seconds.	3600
channel	str | None	Optional target channel key.	None
memory_log_prompt	bool	Enable memory logging for the prompt.	True
memory_log_reply	bool	Enable memory logging for non-empty reply text.	True
memory_tags	list[str] | None	Optional tags applied to memory entries.	None

Returns:

Name	Type	Description
dict	dict	{"text": str, "files": list}.

Notes

accept is advisory and adapter-dependent, not strict server-side validation.

3. Streaming

stream(channel)

Create a stream sender context for incremental text output.

Yield a _StreamSender bound to the resolved channel. The sender emits agent.stream.start/delta/end events when start(), delta(), and end() are called.

Examples:

Stream text deltas:

async with context.channel().stream() as s:
    await s.delta("Hello, ")
    await s.delta("world!")
    await s.end()

Stream to a specific channel:

async with context.channel().stream(channel="web:chat") as s:
    await s.delta("Generating results...")
    await s.end(full_text="Results complete.", memory_tags=["llm"])

Parameters:

Name	Type	Description	Default
channel	str | None	Optional target channel key for this stream context.	None

Returns:

Type	Description
AsyncIterator[_StreamSender]	AsyncIterator[_StreamSender]: Context yielding the stream sender.

Notes

Caller is responsible for calling end(). No auto-end is performed.

4. Utilities

get_latest_uploads(*, clear)

Read latest uploaded files from this channel inbox.

Load files from the ephemeral KV inbox key for the resolved channel and optionally clear the inbox after retrieval.

Examples:

Fetch and clear inbox files:

files = await context.channel().get_latest_uploads()

Fetch without clearing:

files = await context.channel().get_latest_uploads(clear=False)

Parameters:

Name	Type	Description	Default
clear	bool	Pop files when True; read without removal when False.	True

Returns:

Type	Description
list[FileRef]	list[FileRef]: Uploaded files for this channel inbox.

Notes

This reads inbox uploads regardless of whether they came from ask_files().