tool_exec.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.
 
"""
Headless tool runner for npm / @aquin/engine-client spawn fallback.
 
Usage:
  echo '{"name":"read_session_memory","args":{"key":"x"},"ctx":{}}' \\
    | python -m aquin.engine.tool_exec
 
Prints **exactly one** JSON object on stdout (logs must go to stderr).
"""
from __future__ import annotations
 
import json
import sys
from typing import Any
 
from aquin.engine.json_safe import dumps_json_safe
 
 
def _read_payload() -> dict[str, Any]:
    raw = sys.stdin.read()
    if not raw.strip():
        raise ValueError("expected JSON payload on stdin")
    data = json.loads(raw)
    if not isinstance(data, dict):
        raise ValueError("payload must be a JSON object")
    return data
 
 
def _emit(payload: dict[str, Any]) -> None:
    # Single line, strict JSON (no NaN/Inf) so Node JSON.parse never fails.
    sys.stdout.write(dumps_json_safe(payload) + "\n")
    sys.stdout.flush()
 
 
def main() -> int:
    try:
        payload = _read_payload()
        name = payload.get("name")
        if not name:
            raise ValueError("name is required")
        args = payload.get("args") or {}
        ctx = payload.get("ctx") or {}
        if not isinstance(args, dict):
            raise ValueError("args must be an object")
        if not isinstance(ctx, dict):
            raise ValueError("ctx must be an object")
 
        from aquin.engine.sync_dispatch import run_dispatch
        from aquin.engine.tools.registry import _load_stubs
 
        _load_stubs()
        result = run_dispatch(str(name), args, ctx)
        if isinstance(result, dict) and result.get("error"):
            _emit({"ok": False, "error": result["error"], "result": result})
            return 1
        _emit({"ok": True, "result": result})
        return 0
    except Exception as exc:
        _emit({"ok": False, "error": str(exc)})
        return 1
 
 
if __name__ == "__main__":
    sys.exit(main())