Skip to content

trading_agent.extensions.hook_runner

Runs the user's hook scripts when events happen.

A hook is any executable file in a folder named after an event, under workspace/shared/hooks/ or under an installed package's hooks/ folder:

workspace/shared/hooks/order_placed/notify.sh workspace/shared/hooks/kill_switch/page-me.py

Each matching hook runs as its own process with the event as JSON on its standard input, the event's name in TRADING_AGENT_EVENT and the desk in TRADING_AGENT_DESK. The folder name can be any event kind, such as order, risk_decision, kill_switch, session_started, desk_stopped or error, or one of two narrower names: order_placed (an order event whose action is "placed") and risk_rejected (a risk decision that refused an order).

Hooks run outside the gateway's process, at most four at a time, each for at most 30 seconds. A hook that fails, hangs or is missing is logged and has no effect on any desk. Secrets are masked in the JSON a hook receives, and a hook gets only a few safe environment variables, as MCP servers do.

Typical usage example:

runner = HookRunner(event_bus, Path('workspace/shared/hooks'), packages, redactor) task = asyncio.create_task(runner.run())

HookRunner 

HookRunner(event_bus, hooks_directory, packages=None, redactor=None)

Starts hook processes for events.

Subscribes to the bus at once, so no event is missed.

Parameters:

Name Type Description Default
event_bus EventBus

The bus to listen to.

 required
hooks_directory Path

The user's hooks folder, holding one subfolder per event.

 required
packages InstalledPackages | None

The installed packages, whose hooks/ folders are read for every event so a new package counts at once, or None.

 None
redactor SecretRedactor | None

Masks secrets in the JSON given to hooks, or None to pass events unchanged.

 None
Source code in src/trading_agent/extensions/hook_runner.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def __init__(
    self,
    event_bus: event_bus_module.EventBus,
    hooks_directory: Path,
    packages: installed_packages_module.InstalledPackages | None = None,
    redactor: secret_redactor_module.SecretRedactor | None = None,
):
    """Subscribes to the bus at once, so no event is missed.

    Args:
        event_bus (event_bus_module.EventBus): The bus to listen to.
        hooks_directory (Path): The user's hooks folder, holding one subfolder per event.
        packages (installed_packages_module.InstalledPackages | None): The installed packages, whose hooks/ folders are read for every event so a new package counts at once, or None.
        redactor (secret_redactor_module.SecretRedactor | None): Masks secrets in the JSON given to hooks, or None to pass events unchanged.
    """
    self._hooks_directory = hooks_directory
    self._packages = packages
    self._redactor = redactor
    self._subscription = event_bus.subscribe(maximum_queued=_QUEUE_SIZE)
    self._limit = asyncio.Semaphore(_MAXIMUM_RUNNING)
    self._running: set[asyncio.Task] = set()

run async 

run()

Starts hooks for each event until cancelled, then stops the ones still running.

Source code in src/trading_agent/extensions/hook_runner.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
async def run(self) -> None:
    """Starts hooks for each event until cancelled, then stops the ones still running."""
    try:
        while True:
            published_event = await self._subscription.next_event()
            for name, program in self.programs_for(published_event):
                task = asyncio.create_task(
                    self._run_hook(name, program, published_event)
                )
                self._running.add(task)
                task.add_done_callback(self._running.discard)
    finally:
        for task in list(self._running):
            task.cancel()
        self._subscription.close()

programs_for 

programs_for(published_event)

The hook programs an event should start.

Parameters:

Name Type Description Default
published_event Event

The event.

 required

Returns:

Type Description
list[tuple[str, Path]]

Tuples (name, program), where name is the hook folder's event name and program an executable file in it.
Source code in src/trading_agent/extensions/hook_runner.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
def programs_for(
    self,
    published_event: event_module.Event,
) -> list[tuple[str, Path]]:
    """The hook programs an event should start.

    Args:
        published_event (event_module.Event): The event.

    Returns:
        list[tuple[str, Path]]: Tuples (name, program), where name is the hook folder's event name and program an executable file in it.
    """
    if published_event.kind in _SKIPPED_KINDS:
        return []
    programs = []
    for name in self._names(published_event):
        for root in self._roots():
            folder = root / name
            if not folder.is_dir():
                continue
            for path in sorted(folder.iterdir()):
                if path.name.startswith('.') or not path.is_file():
                    continue
                if os.access(path, os.X_OK):
                    programs.append((name, path))
    return programs