#!/usr/bin/env python3 """ CyberBrain26 SAFE-OI API v0.5 - Sync interpreter s dvoufázovým approval tokem SAFE: model navrhne plán → API vytvoří approval → uživatel schválí → API provede FULL: autonomní provoz s auto_run=True """ import os import sys import uuid import json import logging import re from typing import Dict, Optional, Any, List from pathlib import Path from datetime import datetime from fastapi import FastAPI, HTTPException from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel import uvicorn # --- Logging --- logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", handlers=[ logging.FileHandler("/tmp/cb26-safeoi-api.log"), logging.StreamHandler() ] ) logger = logging.getLogger("safe_oi_api") # --- Konfigurace --- LITELLM_PROXY = os.getenv("LITELLM_PROXY", "https://openrouter.ai/api/v1") LITELLM_MASTER_KEY = os.getenv("OPENROUTER_API_KEY") or os.getenv("LITELLM_MASTER_KEY") DEFAULT_MODEL = "nvidia/nemotron-3-ultra-550b-a55b:free" HOST = "127.0.0.1" PORT = 7682 OI_MAX_TOKENS = 1024 MODEL_ALIASES = { "nemo-free": "nvidia/nemotron-3-ultra-550b-a55b:free", "gpt-oss-free": "openai/gpt-oss-20b:free", "glm-free": "z-ai/glm-5.2:free", "north-code-free": "cohere/north-mini-code:free", "claude-haiku": "anthropic/claude-haiku-4.5", "gemini-flash": "google/gemini-2.5-flash", "llama-3.1": "meta-llama/llama-3.1-70b-instruct", "mistral-large": "mistralai/mistral-large-2512", } app = FastAPI(title="CyberBrain26 SAFE-OI API", version="0.5") app.add_middleware( CORSMiddleware, allow_origins=["http://localhost:8080", "http://127.0.0.1:8080"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) sessions: Dict[str, dict] = {} pending_approvals: Dict[str, dict] = {} executed_approvals: Dict[str, dict] = {} class ChatRequest(BaseModel): thread_id: Optional[str] = None message: str model: str = DEFAULT_MODEL mode: str = "SAFE" depth: str = "normal" confirm_full: bool = False class ApproveRequest(BaseModel): action: str = "approve" def resolve_model(model: str) -> str: return MODEL_ALIASES.get(model, model) def get_or_create_session(thread_id: Optional[str], model: str, mode: str) -> str: if not thread_id or thread_id not in sessions: thread_id = thread_id or str(uuid.uuid4()) sessions[thread_id] = { "history": [], "mode": mode.upper(), "model": model, "interpreter": None, "pending_approval": None, } logger.info(f"Nová session: {thread_id} | mode={mode}") return thread_id def create_interpreter(model: str, auto_run: bool): """Vytvoří a nakonfiguruje Open Interpreter instanci""" try: from interpreter import interpreter as oi oi.llm.model = f"openai/{model}" oi.llm.api_base = LITELLM_PROXY oi.llm.api_key = LITELLM_MASTER_KEY oi.llm.headers = {"Authorization": f"Bearer {LITELLM_MASTER_KEY}"} oi.llm.max_tokens = OI_MAX_TOKENS oi.auto_run = auto_run oi.offline = False oi.safe_mode = "off" oi.force_task_completion = False logger.info(f"Interpreter vytvořen | model={model} | auto_run={auto_run}") return oi except Exception as e: logger.error(f"Nepodařilo se vytvořit interpreter: {e}") raise HTTPException(status_code=500, detail=f"Chyba při inicializaci: {e}") def extract_code_from_response(text: str) -> tuple: """Extrahuje code blocky z odpovědi modelu""" code_blocks = re.findall(r'```(\w+)?\n(.*?)```', text, re.DOTALL) if not code_blocks: return None, None, None last_block = code_blocks[-1] language = last_block[0] or "python" code = last_block[1].strip() return language, code, code_blocks def classify_action(code: str, language: str) -> tuple: """Klasifikuje typ akce""" lang = (language or "").strip().lower() # Read-only příkazy read_only_shell = {"pwd", "ls", "dir", "find", "cat", "head", "tail", "grep", "echo", "stat", "wc"} if lang in ("shell", "bash", "sh"): first_word = code.strip().split()[0] if code.strip() else "" if first_word in read_only_shell: return "read", ["pouze čtení"] return "write", ["shell příkaz"] elif lang in ("python", "py"): if any(x in code for x in ["open(", "write(", "Path(", "remove(", "rmdir", "unlink", "rmtree", "shutil"]): return "write", ["zápis nebo mazání souborů"] if any(x in code for x in ["subprocess", "os.system", "os.popen", "exec(", "eval("]): return "shell", ["spouštění shell příkazů"] return "read", ["Python kód (čtení/výpočet)"] return "unknown", ["neznámá akce"] @app.get("/health") def health(): return { "status": "ok", "service": "safe_oi_api", "version": "0.5", "sessions": len(sessions), "proxy": LITELLM_PROXY } @app.post("/chat") def chat(req: ChatRequest): mode = req.mode.upper() if mode == "FULL" and not req.confirm_full: raise HTTPException( status_code=403, detail="FULL-OI je možné aktivovat pouze s confirm_full=true" ) resolved_model = resolve_model(req.model) thread_id = get_or_create_session(req.thread_id, resolved_model, mode) session = sessions[thread_id] session["mode"] = mode session["model"] = resolved_model # Vytvořit nebo získat interpreter auto_run = (mode == "FULL") if session["interpreter"] is None: session["interpreter"] = create_interpreter(resolved_model, auto_run=auto_run) interp = session["interpreter"] # Připravit systémovou instrukci if mode == "SAFE": system_instruction = ( "Jsi CyberBrain26 AI asistent v SAFE režimu. " "DŮLEŽITÉ: Běžíš v headless API bez terminálu. " "NIKDY nepoužívej display_markdown_message, input(), nebo jiné terminálové funkce. " "Před provedením jakékoli akce která mění soubory, maže, stahuje nebo spouští příkazy: " "1) Nejprve popiš co plánuješ udělat. " "2) Ukaž přesný kód v code blocku. " "3) Počkej na potvrzení uživatele. " "Nikdy nespouštěj kód bez explicitního potvrzení v SAFE režimu." ) elif mode == "FULL": system_instruction = ( "Jsi CyberBrain26 AI asistent v FULL režimu. " "DŮLEŽITÉ: Běžíš v headless API bez terminálu. " "NIKDY nepoužívej display_markdown_message, input(), nebo jiné terminálové funkce. " "Můžeš autonomně provádět akce." ) else: system_instruction = "Jsi CyberBrain26 AI asistent." # Připravit zprávu if req.depth == "fast": prefix = "" elif req.depth == "deep": prefix = "[Přemýšlej důkladně] " else: prefix = "" user_message = f"{system_instruction}\n\n{prefix}{req.message}" logger.info(f"[{thread_id}] {mode} | {resolved_model} | {user_message[:80]}...") try: # Reset messages pro novou interakci interp.messages = interp.messages[-10:] if len(interp.messages) > 10 else [] # Odeslat zprávu response = interp.chat(user_message) # Extrahovat odpověď if isinstance(response, list) and len(response) > 0: last = response[-1] if isinstance(last, dict): reply = last.get("content") or last.get("message") or str(last) else: reply = str(last) else: reply = str(response) # V SAFE režimu hledat code blocky approval_id = None has_approval = False if mode == "SAFE": language, code, all_blocks = extract_code_from_response(reply) if code: category, reasons = classify_action(code, language) # Pokud není pure read-only, vytvořit approval if category != "read": approval_id = str(uuid.uuid4()) pending_approvals[approval_id] = { "thread_id": thread_id, "code": code, "language": language, "category": category, "reasons": reasons, "all_blocks": all_blocks, "original_reply": reply, "status": "pending", "created_at": datetime.now().isoformat(), } session["pending_approval"] = approval_id has_approval = True # Upravit odpověď pro uživatele reply += f"\n\n---\n**Čeká na potvrzení**\n\nTyp: {category}\nDůvod: {', '.join(reasons)}\n\nKód:" reply += f"\n```{language}\n{code}\n```" reply += f"\n\nPoužij POST /approvals/{approval_id} pro schválení nebo zamítnutí." # Uložit do historie session["history"].append({"role": "user", "content": req.message}) session["history"].append({"role": "assistant", "content": reply}) return { "thread_id": thread_id, "reply": reply, "mode": mode, "model": req.model, "auto_run": auto_run, "has_approval": has_approval, "approval_id": approval_id } except EOFError as e: # auto_run=False čekal na input() - extrahovat kód z messages logger.warning(f"EOF při čekání na potvrzení - extrahuji kód z messages") # Najít poslední code message code_content = None code_lang = None for msg in reversed(interp.messages): if msg.get("type") == "code" and msg.get("format") and msg.get("content"): code_lang = msg.get("format") code_content = msg.get("content") break if code_content: category, reasons = classify_action(code_content, code_lang) if category != "read": approval_id = str(uuid.uuid4()) pending_approvals[approval_id] = { "thread_id": thread_id, "code": code_content, "language": code_lang, "category": category, "reasons": reasons, "status": "pending", "created_at": datetime.now().isoformat(), } session["pending_approval"] = approval_id reply = f"Návrh kódu ({code_lang}):\n\n```{code_lang}\n{code_content}\n```\n\n**Čeká na potvrzení**\nTyp: {category}\nDůvod: {', '.join(reasons)}\n\nPoužij POST /approvals/{approval_id}" session["history"].append({"role": "user", "content": req.message}) session["history"].append({"role": "assistant", "content": reply}) return { "thread_id": thread_id, "reply": reply, "mode": mode, "model": req.model, "auto_run": auto_run, "has_approval": True, "approval_id": approval_id } raise HTTPException(status_code=500, detail="Nepodařilo se extrahovat kód") except Exception as e: logger.exception(f"Chyba v /chat: {e}") raise HTTPException(status_code=500, detail=str(e)) @app.get("/approvals") def list_approvals(): return { "count": len(pending_approvals), "approvals": [ { "approval_id": aid, "category": a.get("category"), "status": a.get("status"), "created_at": a.get("created_at") } for aid, a in pending_approvals.items() ] } @app.get("/approvals/{approval_id}") def get_approval(approval_id: str): if approval_id not in pending_approvals: raise HTTPException(status_code=404, detail="Approval not found") return pending_approvals[approval_id] @app.post("/approvals/{approval_id}") def approve_action(approval_id: str, req: ApproveRequest): if approval_id not in pending_approvals: raise HTTPException(status_code=404, detail="Approval not found") approval = pending_approvals[approval_id] thread_id = approval["thread_id"] if thread_id not in sessions: raise HTTPException(status_code=404, detail="Session not found") session = sessions[thread_id] interp = session.get("interpreter") if interp is None: raise HTTPException(status_code=500, detail="Interpreter not found") if req.action == "approve": approval["status"] = "approved" session["pending_approval"] = None # Spustit schválený kód code = approval["code"] language = approval["language"] logger.info(f"Schváleno: {approval_id} | {language} kód") try: # Vytvořit zprávu s kódem k provedení exec_message = f"Proveď následující {language} kód:\n\n```{language}\n{code}\n```\n\nPo provedení popiš výsledek jednou větou." # Nastavit auto_run=True pro provedení old_auto_run = interp.auto_run interp.auto_run = True response = interp.chat(exec_message) # Vrátit auto_run interp.auto_run = old_auto_run # Extrahovat odpověď if isinstance(response, list) and len(response) > 0: last = response[-1] if isinstance(last, dict): exec_reply = last.get("content") or str(last) else: exec_reply = str(response) else: exec_reply = str(response) # Uložit do executed_approvals executed_approvals[approval_id] = { "code": code, "language": language, "result": exec_reply, "executed_at": datetime.now().isoformat(), } return { "status": "approved", "approval_id": approval_id, "result": exec_reply } except Exception as e: logger.error(f"Chyba při provádění: {e}") interp.auto_run = old_auto_run raise HTTPException(status_code=500, detail=f"Chyba při provádění: {e}") elif req.action == "deny": approval["status"] = "denied" session["pending_approval"] = None logger.info(f"Zamítnuto: {approval_id}") return {"status": "denied", "approval_id": approval_id} else: raise HTTPException(status_code=400, detail="Neznámá akce") @app.get("/executed/{approval_id}") def get_executed(approval_id: str): if approval_id not in executed_approvals: raise HTTPException(status_code=404, detail="Execution not found") return executed_approvals[approval_id] @app.get("/sessions") def list_sessions(): return { "count": len(sessions), "sessions": [ { "thread_id": tid, "mode": s.get("mode"), "model": s.get("model"), "messages": len(s.get("history", [])), "has_pending": s.get("pending_approval") is not None } for tid, s in sessions.items() ] } @app.delete("/sessions/{thread_id}") def delete_session(thread_id: str): if thread_id in sessions: del sessions[thread_id] logger.info(f"Session smazána: {thread_id}") return {"status": "deleted"} raise HTTPException(status_code=404, detail="Session not found") if __name__ == "__main__": logger.info(f"Starting SAFE-OI API v0.5 on http://{HOST}:{PORT}") uvicorn.run(app, host=HOST, port=PORT, log_level="info")