trading_agent.gateway.routes.memory_routes

Routes for what the desks remember: sessions, summaries, notes, the journal and session search.

GET /api/desks/{desk}/sessions the desk's sessions, newest first GET /api/desks/{desk}/sessions/{session_id} one session's exchanges GET /api/desks/{desk}/summaries the daily summaries GET /api/desks/{desk}/notes the daily notes files GET /api/desks/{desk}/search?keywords=... search_sessions, as the desk sees it GET /api/desks/{desk}/journal?day=... the desk's journal for a day GET /api/journal?day=... the gateway's own journal, for events without a desk

Session ids and days are checked against a strict pattern before they become file names, so a request cannot reach outside the desk's folder.

Typical usage example:

application.include_router(MemoryRoutes(guard, parts, layout).router())

MemoryRoutes

MemoryRoutes(guard, parts, layout)

Bases: RouteGroup

The memory routes.

Creates the routes.

Parameters:

Name	Type	Description	Default
guard	AccessGuard	Checks the access token.	required
parts	Mapping[str, DeskParts]	Each desk's stores, keyed by desk name.	required
layout	WorkspaceLayout	Where the journals and notes are.	required

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

def __init__(
    self,
    guard: access_guard.AccessGuard,
    parts: Mapping[str, gateway_parts.DeskParts],
    layout: workspace_layout.WorkspaceLayout,
):
    """Creates the routes.

    Args:
        guard (access_guard.AccessGuard): Checks the access token.
        parts (Mapping[str, gateway_parts.DeskParts]): Each desk's stores, keyed by desk name.
        layout (workspace_layout.WorkspaceLayout): Where the journals and notes are.
    """
    super().__init__(guard)
    self._parts = parts
    self._layout = layout

sessions

sessions(desk_name)

The desk's sessions.

Parameters:

Name	Type	Description	Default
desk_name	str	The desk.	required

Returns:

Type	Description
dict[str, Any]	The current session's id and every session with its number of exchanges.

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

def sessions(self, desk_name: str) -> dict[str, Any]:
    """The desk's sessions.

    Args:
        desk_name (str): The desk.

    Returns:
        dict[str, Any]: The current session's id and every session with its number of exchanges.
    """
    store = self._find(self._parts, desk_name, 'desk').sessions
    state = store.load_state()
    sessions = []
    for session_id in store.session_ids():
        sessions.append(
            {
                'session_id': session_id,
                'exchanges': len(store.exchanges(session_id)),
            }
        )
    return {
        'current': state.session_id if state is not None else None,
        'sessions': sessions,
    }

session

session(desk_name, session_id)

One session's exchanges.

Parameters:

Name	Type	Description	Default
desk_name	str	The desk.	required
session_id	str	The session, such as 2026-10-19-1.	required

Returns:

Type	Description
dict[str, Any]	The exchanges, oldest first.

Raises:

Type	Description
HTTPException	The id is not a session id (status 400).

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

def session(self, desk_name: str, session_id: str) -> dict[str, Any]:
    """One session's exchanges.

    Args:
        desk_name (str): The desk.
        session_id (str): The session, such as 2026-10-19-1.

    Returns:
        dict[str, Any]: The exchanges, oldest first.

    Raises:
        fastapi.HTTPException: The id is not a session id (status 400).
    """
    store = self._find(self._parts, desk_name, 'desk').sessions
    if not _SESSION_PATTERN.match(session_id):
        raise fastapi.HTTPException(
            status_code=400, detail='Not a session id'
        )
    exchanges = []
    for exchange in store.exchanges(session_id):
        exchanges.append(exchange.to_json_dictionary())
    return {
        'session_id': session_id,
        'exchanges': exchanges,
    }

summaries

summaries(desk_name)

The desk's daily summaries.

Parameters:

Name	Type	Description	Default
desk_name	str	The desk.	required

Returns:

Type	Description
dict[str, Any]	The summaries, newest first.

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

def summaries(self, desk_name: str) -> dict[str, Any]:
    """The desk's daily summaries.

    Args:
        desk_name (str): The desk.

    Returns:
        dict[str, Any]: The summaries, newest first.
    """
    store = self._find(self._parts, desk_name, 'desk').sessions
    summaries = []
    for summary in store.summaries():
        summaries.append(
            {
                'date': summary.trading_date.isoformat(),
                'text': summary.text,
            }
        )
    return {
        'summaries': summaries,
    }

notes

notes(desk_name)

The desk's notes files.

Parameters:

Name	Type	Description	Default
desk_name	str	The desk.	required

Returns:

Type	Description
dict[str, Any]	Each notes file's name and text, newest first.

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

def notes(self, desk_name: str) -> dict[str, Any]:
    """The desk's notes files.

    Args:
        desk_name (str): The desk.

    Returns:
        dict[str, Any]: Each notes file's name and text, newest first.
    """
    self._find(self._parts, desk_name, 'desk')
    folder = self._layout.notes_directory(desk_name)
    files = []
    if folder.is_dir():
        for path in sorted(folder.glob('*.md'), reverse=True):
            files.append(
                {
                    'file': path.name,
                    'text': path.read_text(encoding='utf-8'),
                }
            )
    return {
        'notes': files,
    }

search

search(desk_name, keywords, limit=20)

Searches the desk's past, as search_sessions does.

Parameters:

Name	Type	Description	Default
desk_name	str	The desk.	required
keywords	str	The keywords.	required
limit	int	The most results.	20

Returns:

Type	Description
dict[str, Any]	The hits, best and newest first, and whether they hold every keyword.

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

def search(
    self, desk_name: str, keywords: str, limit: int = 20
) -> dict[str, Any]:
    """Searches the desk's past, as search_sessions does.

    Args:
        desk_name (str): The desk.
        keywords (str): The keywords.
        limit (int): The most results.

    Returns:
        dict[str, Any]: The hits, best and newest first, and whether they hold every keyword.
    """
    self._find(self._parts, desk_name, 'desk')
    result = session_search.SessionSearch(
        self._layout.desk_directory(desk_name)
    ).search(keywords, max(1, min(limit, 100)))
    hits = []
    for hit in result.hits:
        hits.append(
            {
                'day': hit.record.day,
                'kind': hit.record.kind,
                'snippet': hit.snippet,
            }
        )
    return {
        'matched_every_word': result.matched_every_word,
        'hits': hits,
    }

desk_journal

desk_journal(desk_name, day=None, kinds=None, limit=500)

A desk's journal for one day.

Parameters:

Name	Type	Description	Default
desk_name	str	The desk.	required
day	str | None	The India date as YYYY-MM-DD, or None for today.	None
kinds	str | None	Comma-separated event kinds to keep, or None for all.	None
limit	int	The most events, the latest kept.	500

Returns:

Type	Description
dict[str, Any]	The day and its events, oldest first.

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

def desk_journal(
    self,
    desk_name: str,
    day: str | None = None,
    kinds: str | None = None,
    limit: int = 500,
) -> dict[str, Any]:
    """A desk's journal for one day.

    Args:
        desk_name (str): The desk.
        day (str | None): The India date as YYYY-MM-DD, or None for today.
        kinds (str | None): Comma-separated event kinds to keep, or None for all.
        limit (int): The most events, the latest kept.

    Returns:
        dict[str, Any]: The day and its events, oldest first.
    """
    self._find(self._parts, desk_name, 'desk')
    folder = self._layout.desk_directory(desk_name) / 'journal'
    return self._journal(folder, day, kinds, limit)

gateway_journal

gateway_journal(day=None, kinds=None, limit=500)

The gateway's own journal for one day, such as account status and kill-switch events.

Parameters:

Name	Type	Description	Default
day	str | None	The India date as YYYY-MM-DD, or None for today.	None
kinds	str | None	Comma-separated event kinds to keep, or None for all.	None
limit	int	The most events, the latest kept.	500

Returns:

Type	Description
dict[str, Any]	The day and its events, oldest first.

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

def gateway_journal(
    self,
    day: str | None = None,
    kinds: str | None = None,
    limit: int = 500,
) -> dict[str, Any]:
    """The gateway's own journal for one day, such as account status and kill-switch events.

    Args:
        day (str | None): The India date as YYYY-MM-DD, or None for today.
        kinds (str | None): Comma-separated event kinds to keep, or None for all.
        limit (int): The most events, the latest kept.

    Returns:
        dict[str, Any]: The day and its events, oldest first.
    """
    return self._journal(self._layout.root / 'journal', day, kinds, limit)