trading_agent.requests.permission_gate

Holds a desk's call to a tool that needs the user's permission, and asks for it.

The tools listed in [approvals] tools are never run straight away when a desk or one of its subagents calls them. The call is saved as a permission request that lapses after [approvals] expiry_minutes, and the model is told to end its turn. If the user allows it, the harness runs exactly that call; the model never gets a way around it. A standing "Always allow" grant lets the call through at once.

The user's own actions, such as the Flatten button on the Desks page, do not pass through this gate.

Typical usage example:

gate = PermissionGate(['flatten_desk'], 5, store, grants, event_bus) if gate.needs_permission('main', call.name): result = gate.hold('main', call)

PermissionGate

PermissionGate(tool_names, expiry_minutes, store, grants, event_bus)

Decides which calls need permission, and turns them into permission requests.

Creates the gate.

Parameters:

Name	Type	Description	Default
tool_names	Collection[str]	The tools that need permission.	required
expiry_minutes	int	How long a request stays open.	required
store	UserRequestStore	Where requests are saved.	required
grants	PermissionGrants	The standing permissions.	required
event_bus	EventBus	Where requests are published for the web page and Telegram.	required

Source code in src/trading_agent/requests/permission_gate.py

def __init__(
    self,
    tool_names: Collection[str],
    expiry_minutes: int,
    store: user_request_store_module.UserRequestStore,
    grants: permission_grants_module.PermissionGrants,
    event_bus: event_bus_module.EventBus,
):
    """Creates the gate.

    Args:
        tool_names (Collection[str]): The tools that need permission.
        expiry_minutes (int): How long a request stays open.
        store (user_request_store_module.UserRequestStore): Where requests are saved.
        grants (permission_grants_module.PermissionGrants): The standing permissions.
        event_bus (event_bus_module.EventBus): Where requests are published for the web page and Telegram.
    """
    self._tool_names = set(tool_names)
    self._expiry_minutes = expiry_minutes
    self._store = store
    self._grants = grants
    self._event_bus = event_bus

needs_permission

needs_permission(desk_name, tool_name)

Says whether a call must wait for the user.

Parameters:

Name	Type	Description	Default
desk_name	str	The desk calling.	required
tool_name	str	The tool called.	required

Returns:

Type	Description
bool	True when the tool is covered and the desk has no standing permission for it.

Source code in src/trading_agent/requests/permission_gate.py

def needs_permission(self, desk_name: str, tool_name: str) -> bool:
    """Says whether a call must wait for the user.

    Args:
        desk_name (str): The desk calling.
        tool_name (str): The tool called.

    Returns:
        bool: True when the tool is covered and the desk has no standing permission for it.
    """
    if tool_name not in self._tool_names:
        return False
    return not self._grants.allows(desk_name, tool_name)

hold

hold(desk_name, call)

Saves and publishes a permission request for a call, instead of running it.

Parameters:

Name	Type	Description	Default
desk_name	str	The desk calling.	required
call	ToolCall	The call held.	required

Returns:

Type	Description
ToolResult	What the model is told: the request, when it lapses, and to end the turn.

Source code in src/trading_agent/requests/permission_gate.py

def hold(self, desk_name: str, call: messages.ToolCall) -> tool_module.ToolResult:
    """Saves and publishes a permission request for a call, instead of running it.

    Args:
        desk_name (str): The desk calling.
        call (messages.ToolCall): The call held.

    Returns:
        tool_module.ToolResult: What the model is told: the request, when it lapses, and to end the turn.
    """
    expires_at = datetime.datetime.now(datetime.UTC) + datetime.timedelta(
        minutes=self._expiry_minutes
    )
    request = user_request_module.UserRequest.new(
        desk_name,
        user_request_module.PERMISSION,
        {
            'tool_name': call.name,
            'arguments': call.arguments,
            'call_id': call.call_id,
        },
        expires_at,
    )
    self._store.save(request)
    self._event_bus.publish(
        event_module.Event.now(
            event_module.EventKind.USER_REQUEST,
            desk_name,
            request.to_json_dictionary(),
        )
    )
    lapse = expires_at.astimezone(_INDIA).strftime('%H:%M')
    return tool_module.ToolResult.success(
        f'{call.name} needs the user\'s permission, so it was not run. Request {request.request_id} was sent to them and lapses at {lapse} India time. End this turn now; do not retry {call.name} or reach the same result another way. If they allow it, the harness runs exactly this call and tells you the result in a new message.'
    )