executor.py

Go to the documentation of this file.

# Copyright (c) 2025-present Aquin Labs Private Limited. All Rights Reserved.
# This file is part of the Aquin Engine. Unauthorized copying, modification,
# distribution, or use of this file, via any medium, is strictly prohibited.
# Proprietary and confidential. See LICENSE for terms.
 
"""
Single execute_tool dispatcher for Step 35.
 
    execute_tool(name, args, ctx) -> dict
        bridge.call(name, args, ctx)       -> result
        card_mapper.to_card(name, result)  -> card | None
        attach card on result for agent / dispatch_with_sync callers
 
Sync (tool.start / tool.result) is owned by agent.py and sync_dispatch.py —
not here — so we never double-push or overwrite cards with a second result.
"""
from __future__ import annotations
 
from typing import Any
 
 
def execute_tool(name: str, args: dict[str, Any], ctx: dict[str, Any]) -> dict[str, Any]:
    """
    Execute a tool by name via bridge, attach card when applicable.
    Never raises — errors returned as {"error": "..."}.
    """
    from aquin.compute import bridge, card_mapper
 
    try:
        result = bridge.call(name, args, ctx)
    except Exception as exc:
        return {"error": str(exc)}
 
    if not isinstance(result, dict):
        return {"content": result}
 
    card = result.get("card")
    if card is None:
        card = card_mapper.to_card(name, result)
    if card is not None:
        result = {**result, "card": card}
 
    return result