Skip to content

MessageNode API Reference

Bases: ABC

Base class for all message processing nodes.

agent_pool property writable 

agent_pool: AgentPool[Any] | None

Compatibility shim: returns the pool reference.

.. deprecated:: M2 Use :attr:host_context instead. The agent_pool property will be removed in M3. host_context provides the same infrastructure fields via an immutable :class:HostContext.

agent_type property 

agent_type: str

Return the agent-type string used for persistence.

This is the persistence-domain identifier (stored in SessionData.agent_type). It differs from :data:SourceType which is the event-domain identifier used during streaming.

Subclasses may override this to provide a more specific type string (e.g. "native", "acp", "claude_code").

Returns:

Type Description
str

A string identifying the agent type for storage purposes.

connection_stats property 

connection_stats: AggregatedTalkStats

Get stats for all active connections of this node.

display_name property 

display_name: str

Get human-readable display name, falls back to name.

host_context property 

host_context: HostContext | None

Return HostContext from the pool, if available.

message_received class-attribute instance-attribute 

message_received = Signal[ChatMessage[Any]]()

Signal emitted when node receives a message.

message_sent class-attribute instance-attribute 

message_sent = Signal[ChatMessage[Any]]()

Signal emitted when node creates a message.

name property 

name: str

Get agent name.

storage property 

storage: StorageManager | None

Get storage manager from pool.

task_manager property 

task_manager: _TaskManagerShim

Deprecated: backward-compat shim for the old TaskManager API.

.. deprecated:: Use :meth:spawn_task or :attr:_pending_tasks directly.

Returns:

Type Description
_TaskManagerShim

A shim object that provides the old TaskManager method surface
_TaskManagerShim

(complete_tasks, cleanup_tasks, fire_and_forget,
_TaskManagerShim

create_task, _pending_tasks) backed by this node's
_TaskManagerShim

_pending_tasks set.

__aenter__ async 

__aenter__() -> Self

Initialize base message node.

__aexit__ async 

__aexit__(
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: TracebackType | None,
) -> None

Clean up base resources.

__init__ 

__init__(
    name: str | None = None,
    description: str | None = None,
    display_name: str | None = None,
    mcp_servers: Sequence[str | MCPServerConfig] | None = None,
    agent_pool: AgentPool[Any] | None = None,
    enable_logging: bool = True,
    event_configs: Sequence[EventConfig] | None = None,
) -> None

Initialize message node.

__rshift__ 

__rshift__(other: MessageNode[Any, Any] | ProcessorCallback[Any]) -> Talk[TResult]

__rshift__(other: Sequence[MessageNode[Any, Any] | ProcessorCallback[Any]]) -> TeamTalk[TResult]

__rshift__(
    other: (
        MessageNode[Any, Any]
        | ProcessorCallback[Any]
        | Sequence[MessageNode[Any, Any] | ProcessorCallback[Any]]
    ),
) -> Talk[Any] | TeamTalk[Any]

Connect agent to another agent or group.

Example

agent >> other_agent # Connect to single agent agent >> (agent2 & agent3) # Connect to group agent >> "other_agent" # Connect by name (needs pool)

connect_to 

connect_to(
    target: MessageNode[Any, Any] | ProcessorCallback[Any],
    *,
    queued: Literal[True],
    queue_strategy: Literal["concat"]
) -> Talk[str]

connect_to(
    target: MessageNode[Any, Any] | ProcessorCallback[Any],
    *,
    connection_type: ConnectionType = "run",
    name: str | None = None,
    priority: int = 0,
    delay: timedelta | None = None,
    queued: bool = False,
    queue_strategy: QueueStrategy = "latest",
    transform: AnyTransformFn[Any] | None = None,
    filter_condition: AsyncFilterFn | None = None,
    stop_condition: AsyncFilterFn | None = None,
    exit_condition: AsyncFilterFn | None = None
) -> Talk[TResult]

connect_to(
    target: Sequence[MessageNode[Any, Any] | ProcessorCallback[Any]],
    *,
    queued: Literal[True],
    queue_strategy: Literal["concat"]
) -> TeamTalk[str]

connect_to(
    target: Sequence[MessageNode[Any, TResult] | ProcessorCallback[TResult]],
    *,
    connection_type: ConnectionType = "run",
    name: str | None = None,
    priority: int = 0,
    delay: timedelta | None = None,
    queued: bool = False,
    queue_strategy: QueueStrategy = "latest",
    transform: AnyTransformFn[Any] | None = None,
    filter_condition: AsyncFilterFn | None = None,
    stop_condition: AsyncFilterFn | None = None,
    exit_condition: AsyncFilterFn | None = None
) -> TeamTalk[TResult]

connect_to(
    target: Sequence[MessageNode[Any, Any] | ProcessorCallback[Any]],
    *,
    connection_type: ConnectionType = "run",
    name: str | None = None,
    priority: int = 0,
    delay: timedelta | None = None,
    queued: bool = False,
    queue_strategy: QueueStrategy = "latest",
    transform: AnyTransformFn[Any] | None = None,
    filter_condition: AsyncFilterFn | None = None,
    stop_condition: AsyncFilterFn | None = None,
    exit_condition: AsyncFilterFn | None = None
) -> TeamTalk

connect_to(
    target: (
        MessageNode[Any, Any]
        | ProcessorCallback[Any]
        | Sequence[MessageNode[Any, Any] | ProcessorCallback[Any]]
    ),
    *,
    connection_type: ConnectionType = "run",
    name: str | None = None,
    priority: int = 0,
    delay: timedelta | None = None,
    queued: bool = False,
    queue_strategy: QueueStrategy = "latest",
    transform: AnyTransformFn[Any] | None = None,
    filter_condition: AsyncFilterFn | None = None,
    stop_condition: AsyncFilterFn | None = None,
    exit_condition: AsyncFilterFn | None = None
) -> Talk[Any] | TeamTalk

Create connection(s) to target(s).

disconnect_all async 

disconnect_all() -> None

Disconnect from all nodes.

emit_agent_event async 

emit_agent_event(event: RichAgentStreamEvent[Any], source_session_id: str | None = None) -> None

Emit an agent stream event via the event manager.

Parameters:

Name Type Description Default
event RichAgentStreamEvent[Any]

The agent stream event to emit

 required
source_session_id str | None

Optional ID of the session that produced the event

 None

fire_and_forget 

fire_and_forget(coro: Coroutine[Any, Any, Any]) -> None

Run a coroutine in the background without waiting for result.

The coroutine is wrapped in a try/except that logs and swallows exceptions, preventing one non-critical task failure from propagating.

Parameters:

Name Type Description Default
coro Coroutine[Any, Any, Any]

Coroutine to run in the background.

 required

get_context 

get_context(data: Any = None, input_provider: InputProvider | None = None) -> NodeContext

Create a new context for this node.

Parameters:

Name Type Description Default
data Any

Optional custom data to attach to the context

 None
input_provider InputProvider | None

Optional input provider override

 None

Returns:

Type Description
NodeContext

A new NodeContext instance

get_message_history async 

get_message_history(
    session_id: str | None = None, limit: int | None = None
) -> list[ChatMessage[Any]]

Get message history from storage.

Parameters:

Name Type Description Default
session_id str | None

Optional session ID to query history for.

 None
limit int | None

Maximum number of messages to return.

 None

Returns:

Type Description
list[ChatMessage[Any]]

List of chat messages from the session.

get_stats abstractmethod async 

get_stats() -> MessageStats | AggregatedMessageStats

Get message statistics for this node.

log_message async 

log_message(message: ChatMessage[Any]) -> None

Handle message from chat signal.

log_session async 

log_session(
    session_id: str | None = None,
    initial_prompt: str | None = None,
    model: str | None = None,
    parent_session_id: str | None = None,
) -> None

Log conversation to storage if enabled.

Should be called at the start of run_stream() after session_id is set. For native agents, generate session_id first with uuid4(). For wrapped agents (Claude Code), set session_id from SDK session first.

Parameters:

Name Type Description Default
session_id str | None

Optional session ID for the conversation.

 None
initial_prompt str | None

Optional initial prompt to trigger title generation.

 None
model str | None

Requested model identifier for this session.

 None
parent_session_id str | None

Optional parent session ID.

 None

run async 

run(*prompts: Any, **kwargs: Any) -> ChatMessage[TResult]

Execute node with prompts via pydantic-graph single-node graph.

Builds a single-node graph and runs it to completion, wrapping the graph run with :class:SignalEmittingGraphRun so that message_received and message_sent signals are emitted at step boundaries. Subclasses may override this method to provide custom execution logic; in that case the graph-based path is bypassed.

Parameters:

Name Type Description Default
*prompts Any

Input prompts.

 ()
**kwargs Any

Additional execution arguments.

 {}

Returns:

Type Description
ChatMessage[TResult]

The resulting ChatMessage.

run_iter abstractmethod 

run_iter(*prompts: Any, **kwargs: Any) -> AsyncIterator[ChatMessage[Any]]

Yield messages during execution.

run_message async 

run_message(message: ChatMessage[Any], **kwargs: Any) -> ChatMessage[TResult]

Run with an incoming ChatMessage (e.g., from Talk routing).

Extracts content from the message, preserves session_id, and sets parent_id to track the message chain.

Parameters:

Name Type Description Default
message ChatMessage[Any]

The incoming ChatMessage to process

 required
**kwargs Any

Additional arguments passed to run()

 {}

Returns:

Type Description
ChatMessage[TResult]

Response ChatMessage with message chain tracked via parent_id

run_stream async 

run_stream(*prompts: Any, **kwargs: Any) -> AsyncIterator[RichAgentStreamEvent[TResult]]

Run with streaming output via pydantic-graph Graph.iter().

Uses :meth:Graph.iter to drive execution step-by-step, wrapping the graph run with :class:SignalEmittingGraphRun so that message_received and message_sent signals are emitted at step boundaries. For nodes that do not override :meth:run_stream (e.g. most agent subclasses), this yields the final result wrapped in a :class:StreamCompleteEvent. Agent subclasses typically override this with rich event streaming.

Parameters:

Name Type Description Default
*prompts Any

Input prompts.

 ()
**kwargs Any

Additional execution arguments.

 {}

Yields:

Type Description
AsyncIterator[RichAgentStreamEvent[TResult]]

RichAgentStreamEvent tokens during execution.

spawn_task 

spawn_task(coro_fn: Callable[..., Coroutine[Any, Any, Any]], *args: Any) -> Task[Any]

Schedule a coroutine for background execution on this node.

The created task is tracked in _pending_tasks and automatically removed when it completes. All pending tasks are awaited in __aexit__.

Parameters:

Name Type Description Default
coro_fn Callable[..., Coroutine[Any, Any, Any]]

Coroutine function to call.

 required
*args Any

Positional arguments passed to coro_fn.

 ()

Returns:

Type Description
Task[Any]

The created :class:asyncio.Task.

stop_passing_results_to 

stop_passing_results_to(other: MessageNode[Any, Any]) -> None

Stop forwarding results to another node.

to_tool 

to_tool(
    *, name: str | None = None, description: str | None = None, **kwargs: Any
) -> FunctionTool[TResult]

Convert node to a callable tool.

Parameters:

Name Type Description Default
name str | None

Optional tool name override

 None
description str | None

Optional tool description override

 None
**kwargs Any

Additional arguments for subclass customization

 {}

Returns:

Type Description
FunctionTool[TResult]

Tool instance that can be registered