chatflow.ui

This module implements utility classes for formatting/printing agent messages.

async agentopera.chatflow.ui.Console(stream: AsyncGenerator[ToolCallRequestEvent | ToolCallExecutionEvent | MemoryQueryEvent | UserInputRequestedEvent | ModelClientStreamingChunkEvent | ThoughtEvent | TextMessage | MultiModalMessage | StopMessage | ToolCallSummaryMessage | HandoffMessage | VercelMessage | T, None], *, no_inline_images: bool = False, output_stats: bool = False, user_input_manager: UserInputManager | None = None) → T[source]

Consumes the message stream from run_stream() or on_messages_stream() and renders the messages to the console. Returns the last processed TaskResult or Response.

Note

output_stats is experimental and the stats may not be accurate. It will be improved in future releases.

Parameters:

  • stream (AsyncGenerator[AgentEvent | ChatMessage | TaskResult, None] | AsyncGenerator[AgentEvent | ChatMessage | Response, None]) – Message stream to render. This can be from run_stream() or on_messages_stream().

  • no_inline_images (bool, optional) – If terminal is iTerm2 will render images inline. Use this to disable this behavior. Defaults to False.

  • output_stats (bool, optional) – (Experimental) If True, will output a summary of the messages and inline token usage info. Defaults to False.

Returns:

A TaskResult if the stream is from run_stream()

or a Response if the stream is from on_messages_stream().

Return type:

last_processed

class agentopera.chatflow.ui.UserInputManager(callback: Callable[[str], str] | Callable[[str, CancellationToken | None], Awaitable[str]])[source]

Bases: object

get_wrapped_callback() → Callable[[str, CancellationToken | None], Awaitable[str]][source]
notify_event_received(request_id: str) → None[source]