#!/usr/bin/env python3 """ CyberBrain26 - SAFE-OI / FULL-OI API v0.3 Port: 7682 SAFE = čtení automaticky, zápis/mazání/download/installace/network vyžadují potvrzení. FULL = autonomní režim, sudo je povoleno, ale heslo se nikdy neukládá. """ import os import sys import uuid import json import logging import asyncio import threading from typing import Dict, Optional, Any, List from pathlib import Path from datetime import datetime import httpx from fastapi import FastAPI, HTTPException from fastapi.responses import StreamingResponse 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 = "anthropic/claude-3-haiku" HOST = "127.0.0.1" PORT = 7682 OI_MAX_TOKENS = 4096 MODEL_ALIASES = { "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.3") 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] = {} 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, "last_auto_run": None, } logger.info(f"Nová session: {thread_id} | mode={mode}") return thread_id def create_interpreter(model: str, auto_run: bool): """Vytvoří a nakonfiguruje instanci Open Interpreter""" 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 = True 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 Open Interpreter: {e}") def get_interpreter(session: dict, model: str, mode: str): """Vrátí existující nebo nový interpreter pro session""" auto_run = (mode.upper() == "FULL") if session["interpreter"] is None or session.get("last_auto_run") != auto_run: session["interpreter"] = create_interpreter(model, auto_run=auto_run) session["last_auto_run"] = auto_run return session["interpreter"] @app.get("/health") def health(): return { "status": "ok", "service": "safe_oi_api", "version": "0.3", "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" ) thread_id = get_or_create_session(req.thread_id, req.model, mode) session = sessions[thread_id] session["mode"] = mode session["model"] = req.model try: interp = get_interpreter(session, req.model, mode) if req.depth == "fast": prefix = "[Odpověz stručně] " elif req.depth == "deep": prefix = "[Přemýšlej důkladně, postupuj krok za krokem] " else: prefix = "" user_message = prefix + req.message logger.info(f"[{thread_id}] {mode} | {req.model} | {user_message[:80]}...") response = interp.chat(user_message) 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) 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": (mode == "FULL") } 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, "summary": a.get("summary")} 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] if req.action == "approve": approval["status"] = "approved" logger.info(f"Schváleno: {approval_id}") return {"status": "approved", "approval_id": approval_id} elif req.action == "deny": approval["status"] = "denied" 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("/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", [])) } for tid, s in sessions.items() ] } @app.delete("/sessions/{thread_id}") def delete_session(thread_id: str): if thread_id in sessions: try: interp = sessions[thread_id].get("interpreter") if interp: pass except Exception: pass 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 on http://{HOST}:{PORT}") uvicorn.run(app, host=HOST, port=PORT, log_level="info")