trading_agent.tools.send_message_tool

The send_message tool: tells the user something without waiting for a reply.

The message is published as a message event. The web interface (milestone 18) and Telegram (milestone 19) show it; until then it is in the journal and on the gateway's event stream.

Typical usage example:

tool = SendMessageTool(event_bus, 'main')

SendMessageArguments

Bases: BaseModel

The arguments of send_message.

Attributes:

Name	Type	Description
text	str	The message.
urgent	bool	Whether it needs the user's attention now, such as a stop being hit.

SendMessageTool

SendMessageTool(event_bus, desk_name)

Bases: Tool

Publishes a message to the user.

Creates the tool.

Parameters:

Name	Type	Description	Default
event_bus	EventBus	Where the message is published.	required
desk_name	str	The desk sending it.	required

Source code in src/trading_agent/tools/send_message_tool.py

def __init__(self, event_bus: event_bus_module.EventBus, desk_name: str):
    """Creates the tool.

    Args:
        event_bus (event_bus_module.EventBus): Where the message is published.
        desk_name (str): The desk sending it.
    """
    self._event_bus = event_bus
    self._desk_name = desk_name

run async

run(arguments)

Publishes the message.

Parameters:

Name	Type	Description	Default
arguments	SendMessageArguments	The message.	required

Returns:

Type	Description
ToolResult	A confirmation.

Source code in src/trading_agent/tools/send_message_tool.py

async def run(self, arguments: SendMessageArguments) -> tool.ToolResult:
    """Publishes the message.

    Args:
        arguments (SendMessageArguments): The message.

    Returns:
        tool.ToolResult: A confirmation.
    """
    self._event_bus.publish(
        event_module.Event.now(
            event_module.EventKind.MESSAGE,
            self._desk_name,
            {
                'text': arguments.text,
                'urgent': arguments.urgent,
            },
        )
    )
    return tool.ToolResult.success('Message sent.')