agent_bootstrap.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.
 
"""
Dump system prompt + OpenAI tool schemas for the Node agent loop.
 
  python -m aquin.engine.agent_bootstrap
  python -m aquin.engine.agent_bootstrap --model <id>
 
Stdout: one JSON object { system, tools, state }.
"""
from __future__ import annotations
 
import argparse
import json
import sys
from typing import Any
 
 
def main() -> int:
    parser = argparse.ArgumentParser(prog="aquin-agent-bootstrap")
    parser.add_argument("--model", default=None, help="optional active model id for tool filter")
    ns = parser.parse_args()
 
    # Windows consoles often use cp1252 — force UTF-8 for JSON stdout.
    try:
        sys.stdout.reconfigure(encoding="utf-8")
    except Exception:
        pass
 
    state: dict[str, Any] = {}
    if ns.model:
        state["activeModelId"] = ns.model
    else:
        try:
            from aquin.compute.model_loader import get_active_model_id
 
            mid = get_active_model_id()
            if mid:
                state["activeModelId"] = mid
        except Exception:
            pass
 
    try:
        from aquin.engine.agent import _build_tools_schema, _load_system_prompt
 
        system = _load_system_prompt(state)
        tools = _build_tools_schema(state)
    except Exception as exc:
        _write_json({"ok": False, "error": str(exc)})
        return 1
 
    # Fallback: always expose model-free memory tools so Node can demo agents
    # even when no model is loaded (empty schema from session_mode).
    if not tools:
        tools = [
            {
                "type": "function",
                "function": {
                    "name": "write_session_memory",
                    "description": "Persist a key/value pair to session memory.",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "key": {"type": "string"},
                            "value": {},
                        },
                        "required": ["key", "value"],
                    },
                },
            },
            {
                "type": "function",
                "function": {
                    "name": "read_session_memory",
                    "description": "Read a previously stored key from session memory.",
                    "parameters": {
                        "type": "object",
                        "properties": {"key": {"type": "string"}},
                        "required": ["key"],
                    },
                },
            },
        ]
 
    _write_json({"ok": True, "system": system, "tools": tools, "state": state})
    return 0
 
 
def _write_json(payload: dict[str, Any]) -> None:
    data = json.dumps(payload, ensure_ascii=False) + "\n"
    try:
        sys.stdout.buffer.write(data.encode("utf-8"))
        sys.stdout.buffer.flush()
    except Exception:
        # Last resort: ASCII-escaped JSON
        print(json.dumps(payload, ensure_ascii=True), flush=True)
 
 
if __name__ == "__main__":
    sys.exit(main())