Skip to main content

routes.tool_approvals

Local tool approval endpoints served by agent-runtimes.

These endpoints support the sync approval flow without requiring an external ai-agents approval backend. A legacy route prefix is also exposed for compatibility with existing callers.

ToolApprovalCreateRequest Objects​

class ToolApprovalCreateRequest(BaseModel)

Payload to create a pending tool approval request.

ToolApprovalDecisionRequest Objects​

class ToolApprovalDecisionRequest(BaseModel)

Payload to approve or reject a pending request.

ToolApprovalRecord Objects​

class ToolApprovalRecord(BaseModel)

Stored approval record.

register_remote_approval_mapping​

def register_remote_approval_mapping(local_id: str, remote_id: str,
                                     user_jwt_token: str) -> None

Associate a local approval record with its counterpart on ai-agents.

remove_remote_approval_mapping​

def remove_remote_approval_mapping(local_id: str) -> None

Remove the remote mapping for a local approval (cleanup after decision).

register_approval_credentials​

def register_approval_credentials(local_id: str, user_jwt_token: str) -> None

Remember the JWT used to create an approval so we can always relay the eventual decision to ai-agents (and lazily forward the approval if the initial POST failed).

Also starts a persistent ai-agents listener for this JWT so decisions from any participant (e.g. the SaaS UI) are mirrored locally.

register_pending_approval_event​

def register_pending_approval_event(
        approval_id: str) -> tuple[asyncio.Event, dict[str, Any]]

Register an asyncio.Event for inline approval blocking.

Returns (event, result_dict). The caller should await event.wait() and then read result_dict["approved"] (bool) and optionally result_dict["note"] after the event is set.

signal_approval_event​

def signal_approval_event(approval_id: str,
                          approved: bool,
                          note: str | None = None) -> bool

Signal the asyncio.Event for approval_id, returning True if found.

remove_pending_approval_event​

def remove_pending_approval_event(approval_id: str) -> None

Remove the asyncio.Event entry for approval_id (cleanup after wait).

mirror_approval_to_local​

async def mirror_approval_to_local(data: dict) -> ToolApprovalRecord

Mirror an approval record from an external backend (e.g. ai-agents) into the local in-memory store so the frontend can discover it.

get_local_approval_status​

async def get_local_approval_status(approval_id: str) -> str | None

Check the status of an approval in the local in-memory store. Returns the status string or None if not found.

update_local_approval_status​

async def update_local_approval_status(approval_id: str,
                                       status: str,
                                       note: str | None = None) -> None

Update the status of a local approval record and unblock any waiter.

Called by the remote-bridge after it mirrors a decision from the datalayer-ai-agents backend. Must signal the asyncio.Event so that ToolApprovalManager.request_and_wait is unblocked.

forward_approval_to_ai_agents​

async def forward_approval_to_ai_agents(
        record: "ToolApprovalRecord",
        user_jwt_token: str | None) -> str | None

Best-effort forward a pending approval to the datalayer-ai-agents backend.

Called from ToolApprovalManager.request_and_wait so that approvals created via the inline ToolsGuardrailCapability path are visible in remote UI panels (e.g. the ToolApprovals view in datalayer/ui) that poll the ai-agents service rather than the local agent-runtimes endpoints.