trading_agent.gateway.routes.desk_routes

Routes for the desks: their state, messages, and pause, resume and flatten.

GET /api/desks every desk's state, models and session GET /api/desks/{desk} one desk, with its persona GET /api/desks/{desk}/context how full the model's context window is, and the desk's tools GET /api/desks/{desk}/commands the desk's slash commands and usable skills, for a command menu POST /api/desks/{desk}/messages queue a message or /command, with uploads named by id; the reply arrives on /ws/events POST /api/desks/{desk}/model try one backend first, optionally with another of its provider's models, or return to the configured order POST /api/desks/{desk}/pause stop asking models until resumed (commands still work) POST /api/desks/{desk}/resume start answering again POST /api/desks/{desk}/flatten cancel the desk's orders and close its positions

Flatten runs the desk's own flatten_desk tool, so it goes through the risk engine and the journal exactly as when the desk calls it.

Typical usage example:

application.include_router(DeskRoutes(desks, guard, parts, layout, 'dry_run').router())

MessageRequest

Bases: BaseModel

The body of a message sent to a desk.

Attributes:

Name	Type	Description
text	str	What to tell or ask the desk; may be empty when files are attached.
attachment_ids	list[str]	The uploads to send with it, from POST /api/desks/{desk}/attachments.

ModelRequest

Bases: BaseModel

The body of a model choice.

Attributes:

Name	Type	Description
backend	str | None	The backend to try first, or None for the configured order.
model	str | None	Another model of that backend's provider, from its other_models, or None for its configured model.

FlattenRequest

Bases: BaseModel

The body of a flatten request.

Attributes:

Name	Type	Description
reason	str	Why, for the journal.

DeskRoutes

DeskRoutes(desks, guard, parts=None, layout=None, mode='dry_run', attachment_store=None, backend_configurations=None)

Bases: RouteGroup

The /api/desks routes.

Creates the routes.

Parameters:

Name	Type	Description	Default
desks	Mapping[str, DeskRuntime]	The running desks, keyed by name.	required
guard	AccessGuard	Checks the access token.	required
parts	Mapping[str, DeskParts] | None	Each desk's tools, for flatten, or None when flatten is not offered.	None
layout	WorkspaceLayout | None	Where the personas are, or None to leave them out.	None
mode	str	"dry_run" or "live", reported with each desk.	'dry_run'
attachment_store	AttachmentStore | None	Where uploads are kept, or None when messages cannot carry files.	None
backend_configurations	Mapping[str, BackendConfiguration] | None	Each backend's settings, keyed by name, for the model menu, or None to leave them out.	None

Source code in src/trading_agent/gateway/routes/desk_routes.py

def __init__(
    self,
    desks: Mapping[str, desk_runtime.DeskRuntime],
    guard: access_guard.AccessGuard,
    parts: Mapping[str, gateway_parts.DeskParts] | None = None,
    layout: workspace_layout.WorkspaceLayout | None = None,
    mode: str = 'dry_run',
    attachment_store: attachment_store_module.AttachmentStore | None = None,
    backend_configurations: Mapping[str, agent_configuration.BackendConfiguration] | None = None,
):
    """Creates the routes.

    Args:
        desks (Mapping[str, desk_runtime.DeskRuntime]): The running desks, keyed by name.
        guard (access_guard.AccessGuard): Checks the access token.
        parts (Mapping[str, gateway_parts.DeskParts] | None): Each desk's tools, for flatten, or None when flatten is not offered.
        layout (workspace_layout.WorkspaceLayout | None): Where the personas are, or None to leave them out.
        mode (str): "dry_run" or "live", reported with each desk.
        attachment_store (attachment_store_module.AttachmentStore | None): Where uploads are kept, or None when messages cannot carry files.
        backend_configurations (Mapping[str, agent_configuration.BackendConfiguration] | None): Each backend's settings, keyed by name, for the model menu, or None to leave them out.
    """
    super().__init__(guard)
    self._desks = desks
    self._parts = parts or {}
    self._layout = layout
    self._mode = mode
    self._attachment_store = attachment_store
    self._backend_configurations = backend_configurations or {}

list_desks

list_desks()

Every desk's state.

Returns:

Type	Description
dict[str, Any]	The desks, in configuration order, and the mode.

Source code in src/trading_agent/gateway/routes/desk_routes.py

def list_desks(self) -> dict[str, Any]:
    """Every desk's state.

    Returns:
        dict[str, Any]: The desks, in configuration order, and the mode.
    """
    desks = []
    for desk in self._desks.values():
        desks.append(self._summary(desk))
    return {
        'mode': self._mode,
        'desks': desks,
    }

desk

desk(desk_name)

One desk's state and persona.

Parameters:

Name	Type	Description	Default
desk_name	str	The desk.	required

Returns:

Type	Description
dict[str, Any]	The desk's state, models, session and Persona.md text.

Source code in src/trading_agent/gateway/routes/desk_routes.py

def desk(self, desk_name: str) -> dict[str, Any]:
    """One desk's state and persona.

    Args:
        desk_name (str): The desk.

    Returns:
        dict[str, Any]: The desk's state, models, session and Persona.md text.
    """
    desk = self._find(self._desks, desk_name, 'desk')
    summary = self._summary(desk)
    summary['persona'] = None
    if self._layout is not None:
        persona = self._layout.persona_file(desk_name)
        if persona.is_file():
            summary['persona'] = persona.read_text(encoding='utf-8')
    return summary

context

context(desk_name)

How full the desk's context window is, and which tools it has.

The snapshot belongs to the backend that answered last. Claude's is measured by Claude Code at the end of each turn; the other backends' are estimated from their conversation when asked.

Parameters:

Name	Type	Description	Default
desk_name	str	The desk.	required

Returns:

Type	Description
dict[str, Any]	The desk's name, the snapshot or None when there is none yet, and its tool names.

Source code in src/trading_agent/gateway/routes/desk_routes.py

def context(self, desk_name: str) -> dict[str, Any]:
    """How full the desk's context window is, and which tools it has.

    The snapshot belongs to the backend that answered last. Claude's is measured by Claude Code at the end of each turn; the other backends' are estimated from their conversation when asked.

    Args:
        desk_name (str): The desk.

    Returns:
        dict[str, Any]: The desk's name, the snapshot or None when there is none yet, and its tool names.
    """
    desk = self._find(self._desks, desk_name, 'desk')
    usage = desk.turn_runner.context_usage()
    tools = []
    desk_parts = self._parts.get(desk_name)
    if desk_parts is not None:
        tools = desk_parts.tools.tool_names
    return {
        'desk': desk_name,
        'usage': None if usage is None else usage.to_payload(),
        'tools': tools,
    }

commands

commands(desk_name)

The desk's slash commands and usable skills.

Parameters:

Name	Type	Description	Default
desk_name	str	The desk.	required

Returns:

Type	Description
dict[str, Any]	The desk's name and its entries, each with name, usage, description, kind and runs.

Source code in src/trading_agent/gateway/routes/desk_routes.py

def commands(self, desk_name: str) -> dict[str, Any]:
    """The desk's slash commands and usable skills.

    Args:
        desk_name (str): The desk.

    Returns:
        dict[str, Any]: The desk's name and its entries, each with name, usage, description, kind and runs.
    """
    desk = self._find(self._desks, desk_name, 'desk')
    return {
        'desk': desk_name,
        'commands': desk.command_catalogue(),
    }

send_message

send_message(desk_name, message)

Queues a message for a desk and returns at once.

Parameters:

Name	Type	Description	Default
desk_name	str	The desk.	required
message	MessageRequest	The message.	required

Returns:

Type	Description
dict[str, Any]	The new trigger's identifier, to match it with later events.

Raises:

Type	Description
HTTPException	There is no such desk (404), or an attachment is not one of the desk's uploads (422).

Source code in src/trading_agent/gateway/routes/desk_routes.py

def send_message(
    self,
    desk_name: str,
    message: MessageRequest,
) -> dict[str, Any]:
    """Queues a message for a desk and returns at once.

    Args:
        desk_name (str): The desk.
        message (MessageRequest): The message.

    Returns:
        dict[str, Any]: The new trigger's identifier, to match it with later events.

    Raises:
        fastapi.HTTPException: There is no such desk (404), or an attachment is not one of the desk's uploads (422).
    """
    desk = self._find(self._desks, desk_name, 'desk')
    trigger = trigger_module.Trigger.now(
        trigger_module.TriggerSource.USER_MESSAGE,
        desk_name,
        message.text.strip(),
        self._attachments(desk_name, message.attachment_ids),
    )
    desk.submit(trigger)
    return {
        'trigger_id': trigger.trigger_id,
    }

choose_model

choose_model(desk_name, request)

Tries one backend first from now on, optionally with another of its provider's models, or returns to the configured order, as /model does.

Parameters:

Name	Type	Description	Default
desk_name	str	The desk.	required
request	ModelRequest	The backend to try first and its model, or None for both.	required

Returns:

Type	Description
dict[str, Any]	The desk's state afterwards.

Raises:

Type	Description
HTTPException	The desk has no models (status 409), or no backend of that name, or the backend does not offer that model (status 400).

Source code in src/trading_agent/gateway/routes/desk_routes.py

def choose_model(
    self,
    desk_name: str,
    request: ModelRequest,
) -> dict[str, Any]:
    """Tries one backend first from now on, optionally with another of its provider's models, or returns to the configured order, as /model does.

    Args:
        desk_name (str): The desk.
        request (ModelRequest): The backend to try first and its model, or None for both.

    Returns:
        dict[str, Any]: The desk's state afterwards.

    Raises:
        fastapi.HTTPException: The desk has no models (status 409), or no backend of that name, or the backend does not offer that model (status 400).
    """
    desk = self._find(self._desks, desk_name, 'desk')
    router = desk.turn_runner
    if not isinstance(router, model_router.ModelRouter):
        raise fastapi.HTTPException(
            status_code=409, detail='This desk has no models connected'
        )
    if request.model is not None and request.backend is not None:
        configuration = self._backend_configurations.get(request.backend)
        offered = []
        if configuration is not None:
            offered = [
                configuration.model,
                *configuration.other_models,
            ]
        if request.model not in offered:
            raise fastapi.HTTPException(
                status_code=400,
                detail=f'Backend {request.backend!r} does not offer the model {request.model!r}; add it to other_models in trading_agent.toml',
            )
    try:
        router.choose(request.backend, request.model)
    except ValueError as error:
        raise fastapi.HTTPException(
            status_code=400, detail=str(error)
        ) from error
    return self._summary(desk)

pause

pause(desk_name)

Pauses a desk.

Parameters:

Name	Type	Description	Default
desk_name	str	The desk.	required

Returns:

Type	Description
dict[str, Any]	The desk's state.

Source code in src/trading_agent/gateway/routes/desk_routes.py

def pause(self, desk_name: str) -> dict[str, Any]:
    """Pauses a desk.

    Args:
        desk_name (str): The desk.

    Returns:
        dict[str, Any]: The desk's state.
    """
    desk = self._find(self._desks, desk_name, 'desk')
    desk.paused = True
    return self._summary(desk)

resume

resume(desk_name)

Resumes a desk.

Parameters:

Name	Type	Description	Default
desk_name	str	The desk.	required

Returns:

Type	Description
dict[str, Any]	The desk's state.

Source code in src/trading_agent/gateway/routes/desk_routes.py

def resume(self, desk_name: str) -> dict[str, Any]:
    """Resumes a desk.

    Args:
        desk_name (str): The desk.

    Returns:
        dict[str, Any]: The desk's state.
    """
    desk = self._find(self._desks, desk_name, 'desk')
    desk.paused = False
    return self._summary(desk)

flatten async

flatten(desk_name, request)

Cancels a desk's orders and closes its positions, through its own flatten_desk tool.

Parameters:

Name	Type	Description	Default
desk_name	str	The desk.	required
request	FlattenRequest	Why.	required

Returns:

Type	Description
dict[str, Any]	The tool's report and whether it failed.

Source code in src/trading_agent/gateway/routes/desk_routes.py

async def flatten(
    self,
    desk_name: str,
    request: FlattenRequest,
) -> dict[str, Any]:
    """Cancels a desk's orders and closes its positions, through its own flatten_desk tool.

    Args:
        desk_name (str): The desk.
        request (FlattenRequest): Why.

    Returns:
        dict[str, Any]: The tool's report and whether it failed.
    """
    parts = self._find(self._parts, desk_name, 'desk')
    result = await parts.tools.run(
        messages.ToolCall.new(
            'flatten_desk',
            {
                'reason': f'From the user, through the API: {request.reason}',
            },
        )
    )
    return {
        'text': result.text,
        'is_error': result.is_error,
    }