# apps/aroflo_connector_app/agent/agent_cli.py from __future__ import annotations import json import os import re import logging from typing import Any, Dict, List, Optional import click from dotenv import load_dotenv from openai import OpenAI from ..client import AroFloClient from ..services.capabilities import build_capabilities_manifest from .tool_schema import build_tools_from_manifest from .executor import execute_tool_call from .pending_store import load_pending, save_pending, clear_pending, match_token log = logging.getLogger("aroflo_agent_cli") load_dotenv(os.path.join(os.getcwd(), ".env")) CONFIRM_RE = re.compile(r"\bCONFIRM\b\s+([A-Za-z0-9_\-\.~]+)", re.IGNORECASE) def make_aroflo_client() -> AroFloClient: return AroFloClient() def build_tools() -> List[Dict[str, Any]]: client = make_aroflo_client() manifest = build_capabilities_manifest(client) return build_tools_from_manifest(manifest) def _extract_confirm_token(text: str) -> Optional[str]: m = CONFIRM_RE.search(text or "") return m.group(1) if m else None def _dump_json(obj: Any) -> str: if hasattr(obj, "model_dump"): obj = obj.model_dump() return json.dumps(obj, indent=2, ensure_ascii=False) def _extract_proposed_tool(text: str) -> Optional[Dict[str, Any]]: """ Busca un bloque PROPOSED_TOOL: { ...json... } """ if not text: return None m = re.search(r"PROPOSED_TOOL:\s*(\{.*\})", text, re.DOTALL) if not m: return None blob = m.group(1).strip() try: proposed = json.loads(blob) if not isinstance(proposed, dict): return None if "name" not in proposed: return None if "args" not in proposed or not isinstance(proposed["args"], dict): proposed["args"] = {} return proposed except Exception: return None def _tool_params_map(tools: List[Dict[str, Any]]) -> Dict[str, Dict[str, Any]]: """ tool_name -> properties dict """ m: Dict[str, Dict[str, Any]] = {} for t in tools or []: name = t.get("name") props = ((t.get("parameters") or {}).get("properties") or {}) if name: m[name] = props return m def _supports_arg(tool_props: Dict[str, Dict[str, Any]], tool_name: str, arg_name: str) -> bool: props = tool_props.get(tool_name) or {} return arg_name in props @click.group() @click.option("--debug", is_flag=True, help="Logs detallados.") def cli(debug: bool) -> None: logging.basicConfig(level=logging.DEBUG if debug else logging.INFO) @cli.command("ask") @click.argument("question", nargs=-1, required=True) @click.option("--model", default="gpt-4.1-mini", show_default=True) @click.option("--max-tool-loops", default=6, type=int, show_default=True) @click.option("--raw", is_flag=True, help="Imprime respuesta completa del response (debug).") def ask_cmd(question: tuple[str, ...], model: str, max_tool_loops: int, raw: bool) -> None: q = " ".join(question).strip() if not q: raise SystemExit("Pregunta vacía.") if not os.getenv("OPENAI_API_KEY"): raise SystemExit("Falta OPENAI_API_KEY en el entorno o .env") oai = OpenAI() tools = build_tools() tool_props = _tool_params_map(tools) # 0) Si viene "CONFIRM " ejecutamos el pending_action sin pedirle nada al modelo. token = _extract_confirm_token(q) if token: pending = load_pending() hit = match_token(token, pending) if not hit: click.echo("No encontré una acción pendiente válida para ese token (puede haber expirado).") raise SystemExit(2) payload = hit["payload"] tool_name = payload["tool_name"] tool_args = payload["tool_args"] # Seguridad: si la tool soporta dry_run, forzamos ejecución real en confirmación # (por si alguien guardó args con dry_run=True accidentalmente) if isinstance(tool_args, dict) and "dry_run" in tool_args: tool_args["dry_run"] = False click.echo(f"Confirmación recibida. Ejecutando: {tool_name} {tool_args}") result = execute_tool_call(tool_name=tool_name, tool_args=tool_args) # Limpiamos pending aunque falle, para evitar reintentos accidentales clear_pending() if result.get("ok"): click.echo("✅ Cambio aplicado correctamente.") else: click.echo("❌ No se pudo aplicar el cambio.") click.echo(_dump_json(result)) return # 1) Flujo normal: el modelo puede leer y proponer escrituras. input_items: List[Dict[str, Any]] = [ { "role": "system", "content": ( "Eres un asistente que consulta AroFlo usando herramientas.\n" "\n" "Reglas:\n" "- Responde en el mismo idioma del usuario.\n" "- Para consultas usa tools de lectura.\n" "- Para cambios (crear/actualizar):\n" " 1) Primero identifica el registro con tools de lectura.\n" " 2) Muestra el estado actual y el valor propuesto.\n" " 3) NO ejecutes tools de escritura.\n" " 4) SIEMPRE incluye un bloque en una sola línea así:\n" " PROPOSED_TOOL: {\"name\":\"\",\"args\":{...}}\n" " 5) Luego pide confirmación.\n" "\n" "Escrituras (crear/actualizar):\n" "- NUNCA ejecutes una tool de escritura directamente.\n" "- Primero debes:\n" " 1) identificar el objeto (por ejemplo, obtener userid)\n" " 2) mostrar el cambio propuesto\n" " 3) pedir confirmación\n" "- Cuando pidas confirmación, incluye un bloque EXACTO:\n" " PROPOSED_TOOL: {\"name\":\"\",\"args\":{...}}\n" "- El sistema fuera del modelo se encargará del commit con CONFIRM.\n" "\n" "Ejemplo:\n" "PROPOSED_TOOL: {\"name\":\"aroflo__users__update_mobile\",\"args\":{\"postxml\":\"...\"}}\n" ), }, {"role": "user", "content": q}, ] last_response_id: Optional[str] = None for _ in range(max_tool_loops): resp = oai.responses.create( model=model, input=input_items, tools=tools, temperature=0, previous_response_id=last_response_id, ) last_response_id = resp.id if raw: click.echo("\n--- RAW RESPONSE (DEBUG) ---") click.echo(_dump_json(resp)) assistant_text = (resp.output_text or "").strip() if assistant_text: click.echo(assistant_text) proposed = _extract_proposed_tool(assistant_text) if proposed: tool_name = proposed["name"] tool_args = proposed.get("args", {}) or {} # 1) PREVIEW (dry_run) si la tool lo soporta preview_result = None if _supports_arg(tool_props, tool_name, "dry_run"): preview_args = dict(tool_args) preview_args["dry_run"] = True # En preview suele ser útil devolver raw (si existe) pero NO lo forzamos preview_result = execute_tool_call(tool_name=tool_name, tool_args=preview_args) click.echo("") click.echo("=== PREVIEW (dry_run) ===") click.echo(_dump_json(preview_result)) click.echo("=========================") click.echo("") # 2) Guardar acción pendiente con args originales (forzamos dry_run=False si existe) pending_args = dict(tool_args) if _supports_arg(tool_props, tool_name, "dry_run"): pending_args["dry_run"] = False pending = save_pending( { "tool_name": tool_name, "tool_args": pending_args, "preview": preview_result, # opcional, por trazabilidad } ) click.echo("Para confirmar y aplicar el cambio ejecuta:") click.echo( f' python -m apps.aroflo_connector_app.agent.agent_cli ask "CONFIRM {pending["token"]}"' ) return # Ejecutar function_calls calls = [] for item in (resp.output or []): data = item.model_dump() if hasattr(item, "model_dump") else item if isinstance(data, dict) and data.get("type") == "function_call": calls.append(data) if not calls: return for call in calls: name = call.get("name") arguments = call.get("arguments") or {} call_id = call.get("call_id") if isinstance(arguments, str): try: arguments = json.loads(arguments) except Exception: arguments = {"_raw": arguments} result = execute_tool_call(tool_name=name, tool_args=arguments) input_items.append( { "type": "function_call_output", "call_id": call_id, "output": json.dumps(result, ensure_ascii=False), } ) click.echo(f"(Se alcanzó el límite de {max_tool_loops} iteraciones.)") raise SystemExit(2) if __name__ == "__main__": cli()