Base Classes

class interact.base.Handler

Bases: ABC

Base class for all handlers. Each handler has a role and a process method. The process method takes a Message object and a HandlerChain object as input and returns a transformed Message object.

Raises:

  • NotImplementedError – If the process method is not implemented.

  • HandlerError – If the output of process is not a Message or str.

  • UnsupportedHandlerChain – If the next object in the sequence is not a Handler or HandlerChain.

abstract async process(msg, chain)

Process a message and return a new transformed message.

Parameters:

  • msg (Message) – Message to be processed / transformed.

  • chain (HandlerChain) – Current HandlerChain object that is executing this handler.

Raises:

NotImplementedError – If the process method is not implemented.

Returns:

Transformed message.

Return type:

str | Message

role: str
class interact.base.HandlerChain(*handlers, variables={})

Bases: Sequence[Handler]

HandlerChain is a sequence of handlers that are executed in order.

Parameters:

  • handlers (tuple[Handler, ...]) – Array of handlers in the chain.

  • variables (dict) – Variables that are shared between handlers.

handlers

Array of handlers in the chain.

Type:

tuple[Handler, …]

variables

Variables that are shared between handlers.

Type:

dict

last_msg

Last message that was processed by the last handler in the chain.

Type:

Message

history

History of all messages that were processed.

Type:

History

step

Step counter during execution.

Type:

int

async run(msg: str | Message, variables: dict[str, Any] = {}, return_history: Literal[False] = False) → Message
async run(msg: str | Message, variables: dict[str, Any] = {}, return_history: Literal[True] = False) → tuple[Message, History]

Start execution of the HandlerChain.

The first message is either a string (converted to Message with sender “input”) or a Message object. Additional variables can be passed to the HandlerChain with the variables argument.

Handlers are executed in order. Each handler receives the last message as input and returns a new transformed message.

Parameters:

  • msg (str | Message, optional) – Starting message for the HandlerChain. Defaults to “”.

  • variables (dict[str, Any], optional) – Additional variables that will be shared by all handlers. Handlers can update the variables during their processing stage. Defaults to {}.

  • return_history (bool, optional) – If True, the history of all messages is returned. Defaults to False.

Returns:

Transformed message or tuple of transformed message and history.

Return type:

Message | tuple[Message, History]

class interact.base.History(*messages)

Bases: Sequence[Message]

History is a sequence of messages that were processed by a HandlerChain. .. attribute:: messages

Tuple of messages in the history.

type:

tuple[Message]

Parameters:

messages (Message)

class interact.base.Message(primary, sender='Unknown', **kwargs)

Bases: UserString

Message object that is passed between handlers in a HandlerChain. Each message object has a primary content, a sender role, and additional information.

Parameters:

  • primary (str) – The main content of the message.

  • sender (str, optional) – The role of the sender of the message. Defaults to “Unknown”.

  • kwargs (dict) – Additional information about the message.

interact.base.handler(func)

Decorator to convert any async function to a Handler object.

Parameters:

  • func (Callable[[Message, HandlerChain], Coroutine[None, None, str | Message]]) – Async function that takes a Message and HandlerChain object as input and

  • object. (returns a str or Message)

Returns:

Handler object.

Return type:

type[Handler]