trading_agent.desk_builder

Builds one desk with its tools, model backend and turn runner.

Each milestone adds parts to a desk, so this one class is where they are put together. Every backend in the desk's backend_order becomes a route of the desk's model router, most preferred first.

Typical usage example:

desk = DeskBuilder(configuration, secrets, event_bus).build(desk_configuration)

DeskBuilder

DeskBuilder(configuration, secrets, event_bus, mcp_manager=None, usage_tracker=None)

Puts a desk together from the configuration.

Creates the builder.

Parameters:

Name	Type	Description	Default
configuration	AgentConfiguration	The agent's configuration.	required
secrets	Secrets	The secrets read from the environment and .env.	required
event_bus	EventBus	The bus every desk publishes on.	required
mcp_manager	McpClientManager | None	The gateway's MCP connections, shared by every desk, or None.	None
usage_tracker	UsageTracker | None	Today's model usage, for /usage, or None.	None

Source code in src/trading_agent/desk_builder.py

def __init__(
    self,
    configuration: agent_configuration.AgentConfiguration,
    secrets: secrets_module.Secrets,
    event_bus: event_bus_module.EventBus,
    mcp_manager: mcp_client_manager.McpClientManager | None = None,
    usage_tracker: usage_tracker_module.UsageTracker | None = None,
):
    """Creates the builder.

    Args:
        configuration (agent_configuration.AgentConfiguration): The agent's configuration.
        secrets (secrets_module.Secrets): The secrets read from the environment and .env.
        event_bus (event_bus_module.EventBus): The bus every desk publishes on.
        mcp_manager (mcp_client_manager.McpClientManager | None): The gateway's MCP connections, shared by every desk, or None.
        usage_tracker (usage_tracker_module.UsageTracker | None): Today's model usage, for /usage, or None.
    """
    self._configuration = configuration
    self._secrets = secrets
    self._event_bus = event_bus
    self._mcp_manager = mcp_manager
    self._usage_tracker = usage_tracker
    self._alert_stores: dict[str, price_alert_store.PriceAlertStore] = {}
    self._desk_parts: dict[str, gateway_parts.DeskParts] = {}
    self._steering_inboxes: dict[
        str, steering_inbox_module.SteeringInbox
    ] = {}
    self._layout = workspace_layout.WorkspaceLayout(
        configuration.workspace_directory
    )
    self._proposal_store = proposal_store.ProposalStore(self._layout)
    self.user_request_store = user_request_store.UserRequestStore(
        self._layout
    )
    self.permission_grants = permission_grants.PermissionGrants(
        self._layout
    )
    self._permission_gate = permission_gate.PermissionGate(
        configuration.approvals.tools,
        configuration.approvals.expiry_minutes,
        self.user_request_store,
        self.permission_grants,
        event_bus,
    )
    self._attachment_store = attachment_store.AttachmentStore(self._layout)
    self._broker_reader = broker_reader.BrokerReader(
        configuration.tradingmachine_environment_file
    )
    self._result_formatter = result_formatter.ResultFormatter()
    self.calendar = market_calendar.MarketCalendar(
        configuration.market_calendar_directory
    )
    self.hours = exchange_hours.ExchangeHours(self.calendar)
    self._schedule_parts: dict[str, tuple] = {}
    self._order_services = self._build_order_services()
    self._skill_libraries: dict[str, skill_library.SkillLibrary] = {}
    self._packages = installed_packages.InstalledPackages(
        self._layout.packages_directory
    )
    self._subagent_library = subagent_library.SubagentLibrary(
        self._layout.agents_directory,
        self._packages.agent_roots,
    )
    self._knowledge_base = knowledge_base.KnowledgeBase(
        self._layout.knowledge_directory,
        self._layout.knowledge_index_file,
    )

build

build(desk_configuration)

Builds one desk.

Parameters:

Name	Type	Description	Default
desk_configuration	DeskConfiguration	The desk's settings.	required

Returns:

Type	Description
DeskRuntime	The desk, not yet running.

Source code in src/trading_agent/desk_builder.py

def build(
    self,
    desk_configuration: agent_configuration.DeskConfiguration,
) -> desk_runtime.DeskRuntime:
    """Builds one desk.

    Args:
        desk_configuration (agent_configuration.DeskConfiguration): The desk's settings.

    Returns:
        desk_runtime.DeskRuntime: The desk, not yet running.
    """
    desk_name = desk_configuration.name
    self._steering_inboxes[desk_name] = steering_inbox_module.SteeringInbox(
        self._event_bus,
        desk_name,
    )
    checkpoints = checkpoints_module.Checkpoints(
        self._layout.checkpoint_repository(desk_name),
        self._layout.desk_directory(desk_name),
    )
    store = session_store.SessionStore(
        self._layout.sessions_directory(desk_name)
    )
    sandbox = bash_sandbox.BashSandbox(
        self._layout.scratch_directory(desk_name),
        self._configuration.sandbox.hidden_paths,
        self._configuration.sandbox.passed_environment,
    )
    library = skill_library.SkillLibrary(
        self._layout.installed_skills_directory,
        self._layout.agent_skills_directory(desk_name),
        environment=sandbox.passed_environment(),
        package_skill_roots=self._packages.skill_roots,
    )
    self._skill_libraries[desk_name] = library
    writer = skill_writer.SkillWriter(
        library,
        checkpoints,
        self._layout.skill_ledger_file(desk_name),
    )
    jobs = cron_job_store.CronJobStore(
        self._layout.cron_directory(desk_name)
    )
    built_ins = built_in_jobs.BuiltInJobs().jobs(
        desk_configuration.heartbeat_minutes,
        desk_configuration.rollover_time,
        desk_configuration.scheduled_checks,
    )
    alerts = price_alert_store.PriceAlertStore(
        self._layout.alerts_file(desk_name)
    )
    self._alert_stores[desk_name] = alerts
    executor = self._executor(
        desk_configuration,
        checkpoints,
        sandbox,
        library,
        writer,
        jobs,
        built_ins,
        alerts,
    )
    self._desk_parts[desk_name] = gateway_parts.DeskParts(
        sessions=store,
        jobs=jobs,
        built_in_jobs=built_ins,
        alerts=alerts,
        skills=library,
        checkpoints=checkpoints,
        tools=executor.tool_registry,
    )
    runner = self._turn_runner(desk_configuration, executor, store)
    keeper = None
    if isinstance(runner, model_router.ModelRouter):
        curator = skill_curator.SkillCurator(
            library,
            skill_usage.SkillUsageReader(
                self._layout.skill_usage_file(desk_name)
            ),
            writer,
            self._configuration.skills.archive_after_days,
        )
        keeper = session_keeper.SessionKeeper(
            desk_name,
            runner,
            store,
            self._event_bus,
            curator,
        )
    _, _, _, subagents = self._schedule_parts[desk_name]
    return desk_runtime.DeskRuntime(
        desk_configuration,
        self._event_bus,
        runner,
        self._configuration.mode,
        executor.tool_registry,
        keeper,
        library,
        self._usage_tracker,
        market_status.MarketStatus(self.calendar, self.hours),
        self._steering_inboxes[desk_name],
        slash_commands.CommandParts(
            subagent_library=self._subagent_library,
            journal_reader=journal_reader.JournalReader(
                self._layout.desk_directory(desk_name) / 'journal'
            ),
            subagent_runner=subagents,
        ),
    )

desk_parts

desk_parts()

The stores and tools of every desk build() made, for the gateway's API.

Returns:

Type	Description
dict[str, DeskParts]	The parts, keyed by desk name.

Source code in src/trading_agent/desk_builder.py

def desk_parts(self) -> dict[str, gateway_parts.DeskParts]:
    """The stores and tools of every desk build() made, for the gateway's API.

    Returns:
        dict[str, gateway_parts.DeskParts]: The parts, keyed by desk name.
    """
    return dict(self._desk_parts)

shared_parts

shared_parts()

The parts every desk shares, for the gateway's API.

Returns:

Type	Description
SharedParts	The shared parts.

Source code in src/trading_agent/desk_builder.py

def shared_parts(self) -> gateway_parts.SharedParts:
    """The parts every desk shares, for the gateway's API.

    Returns:
        gateway_parts.SharedParts: The shared parts.
    """
    return gateway_parts.SharedParts(
        layout=self._layout,
        reader=self._broker_reader,
        snapshot_reader=self._order_services.snapshot_reader,
        risk_engine=self._order_services.risk_engine,
        ledger=self._order_services.ledger,
        kill_switch=self._kill_switch,
        order_book=self._order_services.order_book,
        proposals=self._proposal_store,
        subagents=self._subagent_library,
        packages=self._packages,
        calendar=self.calendar,
        attachments=self._attachment_store,
    )

market_watcher

market_watcher(desks)

The market watcher over the desks build() made.

Parameters:

Name	Type	Description	Default
desks	Mapping[str, DeskRuntime]	The desks, keyed by name.	required

Returns:

Type	Description
MarketWatcher	The watcher, sharing the desks' order book, risk engine and alert stores.

Source code in src/trading_agent/desk_builder.py

def market_watcher(
    self,
    desks: Mapping[str, desk_runtime.DeskRuntime],
) -> market_watcher.MarketWatcher:
    """The market watcher over the desks build() made.

    Args:
        desks (Mapping[str, desk_runtime.DeskRuntime]): The desks, keyed by name.

    Returns:
        market_watcher.MarketWatcher: The watcher, sharing the desks' order book, risk engine and alert stores.
    """
    return market_watcher.MarketWatcher(
        desks,
        self._alert_stores,
        self._order_services.order_book,
        self._broker_reader,
        self._order_services.risk_engine,
        self.hours,
        self._event_bus,
        self._layout.market_watcher_state_file,
    )

desk_schedule

desk_schedule(desk)

What the cron scheduler needs for a desk that build() made.

Parameters:

Name	Type	Description	Default
desk	DeskRuntime	The desk.	required

Returns:

Type	Description
DeskSchedule	Its jobs, built-in jobs, command runner and subagent runner.

Source code in src/trading_agent/desk_builder.py

def desk_schedule(
    self,
    desk: desk_runtime.DeskRuntime,
) -> cron_scheduler.DeskSchedule:
    """What the cron scheduler needs for a desk that build() made.

    Args:
        desk (desk_runtime.DeskRuntime): The desk.

    Returns:
        cron_scheduler.DeskSchedule: Its jobs, built-in jobs, command runner and subagent runner.
    """
    jobs, built_ins, sandbox, subagents = self._schedule_parts[desk.name]
    return cron_scheduler.DeskSchedule(
        desk=desk,
        store=jobs,
        built_in_jobs=built_ins,
        command_runner=command_job_runner.CommandJobRunner(
            sandbox,
            self._configuration.sandbox.command_timeout_seconds,
        ),
        subagent_runner=subagents,
    )

backend_runner

backend_runner(desk_configuration, backend_name, backend_configuration, executor, prompt, maximum_rounds, steering_inbox=None)

Builds the turn runner for one backend, according to its kind.

The desk's own runners and its subagents' runners are both built here.

Parameters:

Name	Type	Description	Default
desk_configuration	DeskConfiguration	The desk's settings.	required
backend_name	str	The backend's name in trading_agent.toml.	required
backend_configuration	BackendConfiguration	Its settings.	required
executor	ToolCallExecutor	The tool executor the runner calls tools through.	required
prompt	PromptBuilder	Builds the instructions; anything with a build() method returning a SystemPrompt.	required
maximum_rounds	int	The most model calls one turn may make.	required
steering_inbox	SteeringInbox | None	The desk's mid-turn messages, for the desk's own runners; None for a subagent's.	None

Returns:

Type	Description
TurnRunner	The runner.

Source code in src/trading_agent/desk_builder.py

def backend_runner(
    self,
    desk_configuration: agent_configuration.DeskConfiguration,
    backend_name: str,
    backend_configuration: agent_configuration.BackendConfiguration,
    executor: tool_call_executor.ToolCallExecutor,
    prompt: prompt_builder.PromptBuilder,
    maximum_rounds: int,
    steering_inbox: steering_inbox_module.SteeringInbox | None = None,
) -> turn_runner.TurnRunner:
    """Builds the turn runner for one backend, according to its kind.

    The desk's own runners and its subagents' runners are both built here.

    Args:
        desk_configuration (agent_configuration.DeskConfiguration): The desk's settings.
        backend_name (str): The backend's name in trading_agent.toml.
        backend_configuration (agent_configuration.BackendConfiguration): Its settings.
        executor (tool_call_executor.ToolCallExecutor): The tool executor the runner calls tools through.
        prompt (prompt_builder.PromptBuilder): Builds the instructions; anything with a build() method returning a SystemPrompt.
        maximum_rounds (int): The most model calls one turn may make.
        steering_inbox (steering_inbox_module.SteeringInbox | None): The desk's mid-turn messages, for the desk's own runners; None for a subagent's.

    Returns:
        turn_runner.TurnRunner: The runner.
    """
    if backend_configuration.kind == 'claude_code':
        return self._claude_code_runner(
            desk_configuration,
            backend_name,
            backend_configuration,
            executor,
            prompt,
            maximum_rounds,
            steering_inbox,
        )
    api_key = self._secrets.openai_api_key
    if backend_configuration.api_key_variable is not None:
        api_key = self._secrets.named(backend_configuration.api_key_variable)
    if backend_configuration.kind == 'deepseek':
        model_backend = deepseek_backend.DeepSeekBackend(
            backend_name,
            backend_configuration,
            api_key,
        )
    elif backend_configuration.kind == 'openai_compatible':
        model_backend = openai_compatible_backend.OpenAICompatibleBackend(
            backend_name,
            backend_configuration,
            api_key,
        )
    else:
        model_backend = ollama_backend.OllamaBackend(
            backend_name,
            backend_configuration,
        )
    return self._agent_loop_runner(
        desk_configuration,
        model_backend,
        backend_configuration,
        executor,
        prompt,
        maximum_rounds,
        steering_inbox,
    )