trading_agent.client.message_client

Sends one message to a running gateway and prints the reply as it streams in.

It is what trading-agent send and trading-agent -p use, so a desk can be talked to from a script, a pipe, or with ! from Claude Code.

When the desk is busy, the gateway hands a plain message to the running turn instead of queueing it. The client then follows that turn until it ends, as the Telegram channel does. Questions, plans and permission requests the desk raises meanwhile are printed with their ids, to be answered in the terminal program or on the web page.

Three output formats are offered, as in claude -p:

text the reply as it streams, with one line per tool call json one JSON object at the end, with the result, tool calls and backend stream-json every event of the turn as one JSON line, then a final result line

Typical usage example:

client = MessageClient('http://127.0.0.1:8120', token) exit_code = asyncio.run(client.send('general', 'What time is it?'))

MessageClient

MessageClient(base_url, access_token, output=stdout, output_format='text')

A small client for the gateway's message route and event stream.

Creates the client.

Parameters:

Name	Type	Description	Default
base_url	str	The gateway's address, such as http://127.0.0.1:8120.	required
access_token	str	The gateway's access token.	required
output	TextIO	Where the reply is printed.	stdout
output_format	str	"text", "json" or "stream-json".	'text'

Raises:

Type	Description
ValueError	The output format is not one of the three.

Source code in src/trading_agent/client/message_client.py

def __init__(
    self,
    base_url: str,
    access_token: str,
    output: TextIO = sys.stdout,
    output_format: str = 'text',
):
    """Creates the client.

    Args:
        base_url (str): The gateway's address, such as http://127.0.0.1:8120.
        access_token (str): The gateway's access token.
        output (TextIO): Where the reply is printed.
        output_format (str): "text", "json" or "stream-json".

    Raises:
        ValueError: The output format is not one of the three.
    """
    if output_format not in _FORMATS:
        raise ValueError(f'Unknown output format: {output_format!r}')
    self._base_url = base_url.rstrip('/')
    self._access_token = access_token
    self._output = output
    self._output_format = output_format

send async

send(desk_name, text)

Sends the message and prints everything the desk does until its turn ends.

Parameters:

Name	Type	Description	Default
desk_name	str	The desk to talk to.	required
text	str	The message, or a /command.	required

Returns:

Type	Description
int	0 when the turn finished normally, 1 when it ended in an error or the gateway refused the message.

Source code in src/trading_agent/client/message_client.py

async def send(self, desk_name: str, text: str) -> int:
    """Sends the message and prints everything the desk does until its turn ends.

    Args:
        desk_name (str): The desk to talk to.
        text (str): The message, or a /command.

    Returns:
        int: 0 when the turn finished normally, 1 when it ended in an error or the gateway refused the message.
    """
    gateway = gateway_client.GatewayClient(
        self._base_url, self._access_token
    )
    async with gateway:
        try:
            socket = await websockets.connect(
                gateway.events_address(desk_name)
            )
        except (OSError, websockets.InvalidStatus) as error:
            self._problem(
                f'Could not open the event stream at {self._base_url}: {error}. Is the gateway running? Start it with `bin/trading-agent serve` or `systemctl --user start trading-agent`.'
            )
            return 1
        async with socket:
            try:
                trigger_id = await gateway.send_message(desk_name, text)
            except gateway_client.GatewayError as error:
                self._problem(f'The gateway refused the message: {error}')
                return 1
            follower = TurnFollower(
                desk_name,
                trigger_id,
                self._output,
                self._output_format,
            )
            return await asyncio.wait_for(
                follower.follow(socket),
                timeout=_REPLY_TIMEOUT_SECONDS,
            )

TurnFollower

TurnFollower(desk_name, trigger_id, output, output_format)

Reads the event stream for one message and prints what its turn does.

Creates the follower.

Parameters:

Name	Type	Description	Default
desk_name	str	The desk the message went to.	required
trigger_id	str	The message's trigger id.	required
output	TextIO	Where to print.	required
output_format	str	"text", "json" or "stream-json".	required

Source code in src/trading_agent/client/message_client.py

def __init__(
    self,
    desk_name: str,
    trigger_id: str,
    output: TextIO,
    output_format: str,
):
    """Creates the follower.

    Args:
        desk_name (str): The desk the message went to.
        trigger_id (str): The message's trigger id.
        output (TextIO): Where to print.
        output_format (str): "text", "json" or "stream-json".
    """
    self._desk_name = desk_name
    self._trigger_id = trigger_id
    self._followed_id = trigger_id
    self._output = output
    self._output_format = output_format
    self._streamed_any_text = False
    self._started_at = time.monotonic()
    self._answer = ''
    self._error: str | None = None
    self._backend: str | None = None
    self._tool_calls: list[dict[str, Any]] = []
    self._requests: list[dict[str, Any]] = []

follow async

follow(socket)

Prints the turn's events until it finishes.

Parameters:

Name	Type	Description	Default
socket	Any	The open event stream, with an async recv method that returns JSON text.	required

Returns:

Type	Description
int	0 for a normal finish, 1 when the turn ended in an error.

Source code in src/trading_agent/client/message_client.py

async def follow(self, socket: Any) -> int:
    """Prints the turn's events until it finishes.

    Args:
        socket (Any): The open event stream, with an async recv method that returns JSON text.

    Returns:
        int: 0 for a normal finish, 1 when the turn ended in an error.
    """
    while True:
        event = json.loads(await socket.recv())
        if self.handle(event):
            return self.finish()

handle

handle(event)

Takes one event from the stream.

Parameters:

Name	Type	Description	Default
event	dict[str, Any]	The event.	required

Returns:

Type	Description
bool	True when the followed turn has finished.

Source code in src/trading_agent/client/message_client.py

def handle(self, event: dict[str, Any]) -> bool:
    """Takes one event from the stream.

    Args:
        event (dict[str, Any]): The event.

    Returns:
        bool: True when the followed turn has finished.
    """
    kind = event.get('kind', '')
    payload = event.get('payload') or {}
    if kind == 'user_request' and event.get('desk') == self._desk_name:
        self._stream(event)
        self._show_request(payload)
        return False
    event_trigger = payload.get('trigger_id')
    if event_trigger == self._trigger_id and kind == 'trigger_received':
        self._stream(event)
        if isinstance(payload.get('steering_into'), str):
            self._note(
                'The desk is busy, so your message goes to the running turn at its next tool call.'
            )
        return False
    if event_trigger == self._trigger_id and kind == 'steering':
        self._stream(event)
        if payload.get('delivered') is True:
            self._followed_id = str(payload.get('into_trigger_id'))
            self._note('Your message reached the running turn.')
        else:
            self._note(
                'The running turn ended first, so your message is the next turn.'
            )
        return False
    if event_trigger != self._followed_id:
        return False
    self._stream(event)
    return self._take(kind, payload)

finish

finish()

Prints the closing output for the json formats and picks the exit code.

Returns:

Type	Description
int	0 for a normal finish, 1 when the turn ended in an error.

Source code in src/trading_agent/client/message_client.py

def finish(self) -> int:
    """Prints the closing output for the json formats and picks the exit code.

    Returns:
        int: 0 for a normal finish, 1 when the turn ended in an error.
    """
    exit_code = 0
    if self._error is not None:
        exit_code = 1
    if self._output_format == 'text':
        return exit_code
    request_ids = []
    for request in self._requests:
        request_ids.append(request.get('request_id'))
    result = {
        'kind': 'result',
        'desk': self._desk_name,
        'trigger_id': self._trigger_id,
        'answered_in_trigger_id': self._followed_id,
        'is_error': self._error is not None,
        'result': self._answer,
        'error': self._error,
        'backend': self._backend,
        'tool_calls': self._tool_calls,
        'requests': request_ids,
        'duration_seconds': round(time.monotonic() - self._started_at, 1),
    }
    print(
        json.dumps(result, ensure_ascii=False),
        file=self._output,
        flush=True,
    )
    return exit_code