#!/usr/bin/env python3 """MCP vault second brain — backend SQLite + RAG, API DB-nativa. Tool esposti (4): upsert_item, patch_item, delete_item, execute_query. Fonte di verità: vault.db (SQLite). Items type='os' (kernel-elon, elon-memory) — entrambi scrivibili via upsert_item da Elon. Auth: OAuth 2.0 authorization code grant, in-memory. """ from __future__ import annotations import json import os import re import secrets import sqlite3 import time from pathlib import Path from typing import Any from dotenv import load_dotenv from mcp.server.auth.provider import ( AccessToken, AuthorizationCode, AuthorizationParams, OAuthAuthorizationServerProvider, RefreshToken, construct_redirect_uri, ) from mcp.server.auth.settings import AuthSettings, ClientRegistrationOptions, RevocationOptions from mcp.server.fastmcp import FastMCP from mcp.server.transport_security import TransportSecuritySettings from mcp.shared.auth import OAuthClientInformationFull, OAuthToken from pydantic import AnyHttpUrl, AnyUrl TRUSTED_REDIRECT_HOSTS = {"claude.ai", "claude.com", "localhost", "127.0.0.1"} class TrustedClient(OAuthClientInformationFull): """Client il cui redirect_uri non è fissato a una stringa esatta: viene accettato qualsiasi URI con host in TRUSTED_REDIRECT_HOSTS. Il client_secret resta obbligatorio allo scambio del token, quindi l'autenticità non dipende da questo.""" def validate_redirect_uri(self, redirect_uri: AnyUrl | None) -> AnyUrl: if redirect_uri is not None and redirect_uri.host in TRUSTED_REDIRECT_HOSTS: return redirect_uri return super().validate_redirect_uri(redirect_uri) load_dotenv(Path(__file__).resolve().parent / ".env") ISSUER_URL = os.environ.get("ISSUER_URL", "https://vault.privcloud.dev") CLIENT_ID = os.environ["VAULT_OAUTH_CLIENT_ID"] CLIENT_SECRET = os.environ["VAULT_OAUTH_CLIENT_SECRET"] ACCESS_TOKEN_TTL = 3600 * 24 * 30 # 30 giorni — passo 0, non serve rotazione frequente TOKENS_PATH = Path("/mnt/ssd/data/adrian-ops/oauth_tokens.json") class VaultOAuthProvider(OAuthAuthorizationServerProvider): """Provider OAuth con persistenza token su disco. Client pre-registrato per Elon/Claude.ai + registrazione dinamica per Claude Code (RFC 7591). """ def __init__(self) -> None: self._auth_codes: dict[str, AuthorizationCode] = {} self._access_tokens: dict[str, AccessToken] = {} self._refresh_tokens: dict[str, RefreshToken] = {} self._load_tokens() # Registro client: pre-registrato + dinamici self._clients: dict[str, OAuthClientInformationFull] = { CLIENT_ID: TrustedClient( client_id=CLIENT_ID, client_secret=CLIENT_SECRET, redirect_uris=[AnyUrl("https://claude.ai/api/mcp/auth_callback")], grant_types=["authorization_code", "refresh_token"], response_types=["code"], token_endpoint_auth_method="client_secret_post", ) } def _load_tokens(self) -> None: if not TOKENS_PATH.exists(): return try: data = json.loads(TOKENS_PATH.read_text()) now = time.time() for tok, d in data.get("access_tokens", {}).items(): if d.get("expires_at", 0) > now: self._access_tokens[tok] = AccessToken(**d) for tok, d in data.get("refresh_tokens", {}).items(): self._refresh_tokens[tok] = RefreshToken(**d) except Exception: pass # token corrotti o incompatibili — ripartenza pulita def _save_tokens(self) -> None: try: data = { "access_tokens": {t: v.model_dump() for t, v in self._access_tokens.items()}, "refresh_tokens": {t: v.model_dump() for t, v in self._refresh_tokens.items()}, } TOKENS_PATH.write_text(json.dumps(data)) except Exception: pass async def get_client(self, client_id: str) -> OAuthClientInformationFull | None: return self._clients.get(client_id) async def register_client(self, client_info: OAuthClientInformationFull) -> None: # Accetta qualsiasi registrazione dinamica (RFC 7591) — nessun login reale, # siamo gli unici utenti previsti. self._clients[client_info.client_id] = client_info async def authorize(self, client: OAuthClientInformationFull, params: AuthorizationParams) -> str: code = secrets.token_urlsafe(32) self._auth_codes[code] = AuthorizationCode( code=code, scopes=params.scopes or [], expires_at=time.time() + 600, client_id=client.client_id, code_challenge=params.code_challenge, redirect_uri=params.redirect_uri, redirect_uri_provided_explicitly=params.redirect_uri_provided_explicitly, resource=params.resource, ) return construct_redirect_uri(str(params.redirect_uri), code=code, state=params.state) async def load_authorization_code( self, client: OAuthClientInformationFull, authorization_code: str ) -> AuthorizationCode | None: return self._auth_codes.get(authorization_code) async def exchange_authorization_code( self, client: OAuthClientInformationFull, authorization_code: AuthorizationCode ) -> OAuthToken: del self._auth_codes[authorization_code.code] access_token = secrets.token_urlsafe(32) refresh_token = secrets.token_urlsafe(32) self._access_tokens[access_token] = AccessToken( token=access_token, client_id=client.client_id, scopes=authorization_code.scopes, expires_at=int(time.time() + ACCESS_TOKEN_TTL), resource=authorization_code.resource, ) self._refresh_tokens[refresh_token] = RefreshToken( token=refresh_token, client_id=client.client_id, scopes=authorization_code.scopes, ) self._save_tokens() return OAuthToken( access_token=access_token, token_type="bearer", expires_in=ACCESS_TOKEN_TTL, refresh_token=refresh_token, scope=" ".join(authorization_code.scopes), ) async def load_refresh_token(self, client: OAuthClientInformationFull, refresh_token: str) -> RefreshToken | None: return self._refresh_tokens.get(refresh_token) async def exchange_refresh_token( self, client: OAuthClientInformationFull, refresh_token: RefreshToken, scopes: list[str], ) -> OAuthToken: del self._refresh_tokens[refresh_token.token] access_token = secrets.token_urlsafe(32) new_refresh = secrets.token_urlsafe(32) self._access_tokens[access_token] = AccessToken( token=access_token, client_id=client.client_id, scopes=scopes or refresh_token.scopes, expires_at=int(time.time() + ACCESS_TOKEN_TTL), ) self._refresh_tokens[new_refresh] = RefreshToken( token=new_refresh, client_id=client.client_id, scopes=scopes or refresh_token.scopes, ) self._save_tokens() return OAuthToken( access_token=access_token, token_type="bearer", expires_in=ACCESS_TOKEN_TTL, refresh_token=new_refresh, scope=" ".join(scopes or refresh_token.scopes), ) async def load_access_token(self, token: str) -> AccessToken | None: at = self._access_tokens.get(token) if at and at.expires_at and at.expires_at < time.time(): del self._access_tokens[token] return None return at async def revoke_token(self, token) -> None: self._access_tokens.pop(getattr(token, "token", token), None) self._refresh_tokens.pop(getattr(token, "token", token), None) self._save_tokens() mcp = FastMCP( "vault-secondbrain", transport_security=TransportSecuritySettings( allowed_hosts=["vault.privcloud.dev", "10.0.0.2:8765", "localhost:8765"], allowed_origins=["https://vault.privcloud.dev"], ), auth_server_provider=VaultOAuthProvider(), auth=AuthSettings( issuer_url=AnyHttpUrl(ISSUER_URL), resource_server_url=AnyHttpUrl(f"{ISSUER_URL}/mcp"), client_registration_options=ClientRegistrationOptions(enabled=True), revocation_options=RevocationOptions(enabled=True), ), ) # ── RAG ─────────────────────────────────────────────────────────────────────── LANCEDB_PATH = Path("/mnt/ssd/data/adrian-ops/lancedb") RAG_MODEL = "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2" _rag_model = None _lance_db = None _lance_tbl = None def _get_rag(): global _rag_model, _lance_db, _lance_tbl if _rag_model is None: from sentence_transformers import SentenceTransformer import lancedb _rag_model = SentenceTransformer(RAG_MODEL) _lance_db = lancedb.connect(str(LANCEDB_PATH)) _lance_tbl = _lance_db.open_table("documenti") return _rag_model, _lance_tbl def _normalize_chunk(text: str) -> str: """Normalizza artefatti da estrazione PDF: tab→spazio, mid-word hyphen-newline, spazi multipli.""" text = text.replace("\t", " ") text = re.sub(r"-\n(\w)", r"\1", text) # "soft-\nware" → "software" text = re.sub(r" {2,}", " ", text) # spazi multipli → uno return text.strip() # ── SQLite ──────────────────────────────────────────────────────────────────── DB_PATH = Path("/mnt/ssd/data/vault-secondbrain/vault.db") VAULT_TIPI = {"cantiere", "persona", "filo", "dinamica", "episodio", "decisione", "entità", "agent_memory", "limbo"} # Tipi esenti dal gate semantico (staging o plumbing interno) VAULT_TIPI_UNGATED = {"limbo", "agent_memory"} OS_WRITABLE = {"elon-memory", "kernel-elon"} # items type='os' scrivibili via upsert_item OS_WRITABLE_PREFIX = ("msg-e2a-", "msg-a2e-", "ack-e2a-", "ack-a2e-", "ping-elon") # pattern msg/ack (legacy, lasciati per sicurezza) def _db() -> sqlite3.Connection: conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row conn.execute("PRAGMA journal_mode=WAL") return conn # ── Tool ────────────────────────────────────────────────────────────────────── @mcp.tool() def upsert_item(id: str, tipo: str, body: str = "", meta: str = "{}") -> str: """ Crea o aggiorna un nodo nel DB. id: slug univoco (es. 'mario-rossi', 'progetto-alpha') tipo: cantiere | persona | filo | dinamica | episodio | decisione | entità | agent_memory | limbo body: corpo testuale (note, descrizione, contenuto libero) meta: JSON con campi strutturati es. '{"nome": "Mario Rossi", "stato": "aperto"}' Il tipo viene salvato in meta.tipo; il type DB è sempre 'nodo'. """ try: meta_dict = json.loads(meta) except json.JSONDecodeError as e: raise ValueError(f"meta non è JSON valido: {e}") conn = _db() existing = conn.execute("SELECT type FROM items WHERE id=?", (id,)).fetchone() is_msg_ack = id.startswith(OS_WRITABLE_PREFIX) # Items type='os' non scrivibili (tranne OS_WRITABLE e pattern msg/ack) if existing and existing["type"] == "os" and id not in OS_WRITABLE and not is_msg_ack: conn.close() raise PermissionError(f"Item '{id}' è di sistema (type='os') e non modificabile via MCP.") # msg/ack inter-agente → sempre type='os' (plumbing di sistema, fuori dal grafo) if is_msg_ack: db_type = "os" elif existing and existing["type"] == "os": db_type = "os" else: db_type = "nodo" # ── Gate semantico (solo nodi, tipi ungated esenti) ───────────────────────── if db_type == "nodo": # Regola 1: tipo deve essere riconosciuto if tipo not in VAULT_TIPI: conn.close() raise ValueError( f"Tipo '{tipo}' non riconosciuto. Tipi validi: {', '.join(sorted(VAULT_TIPI))}" ) if tipo not in VAULT_TIPI_UNGATED: # Regola 2: body non vuoto (min 20 chars) if len(body.strip()) < 20: conn.close() raise ValueError( f"body troppo corto ({len(body.strip())} chars, minimo 20). " "Descrivi il nodo in modo significativo." ) # Regola 3: relazioni con label obbligatoria relazioni = meta_dict.get("relazioni") or [] if relazioni: for i, rel in enumerate(relazioni): if not isinstance(rel, dict): conn.close() raise ValueError(f"relazioni[{i}] deve essere un oggetto {{id, label}}.") if not rel.get("id"): conn.close() raise ValueError(f"relazioni[{i}] manca di 'id'.") if not rel.get("label", "").strip(): conn.close() raise ValueError( f"relazioni[{i}] (→ {rel.get('id')}) ha label vuota. " "Specifica una label che descriva la relazione." ) # ──────────────────────────────────────────────────────────────────────────── meta_dict["tipo"] = tipo conn.execute( """INSERT INTO items(id, type, body, file, meta) VALUES (?,?,?,NULL,?) ON CONFLICT(id) DO UPDATE SET type=excluded.type, body=excluded.body, meta=excluded.meta""", (id, db_type, body, json.dumps(meta_dict, ensure_ascii=False)), ) conn.commit() conn.close() return f"OK: {db_type}/{tipo}/{id}" @mcp.tool() def patch_item(id: str, old_string: str, new_string: str, replace_all: bool = False) -> str: """ Modifica mirata del body di un nodo esistente, senza riscriverlo per intero. id: id del nodo da modificare (deve esistere già — usa upsert_item per crearne uno nuovo) old_string: testo esatto da sostituire nel body. Stringa vuota "" = append (new_string viene aggiunto in coda al body attuale, senza toccare il resto). new_string: testo con cui sostituire old_string (o da accodare, se old_string è "") replace_all: se True sostituisce tutte le occorrenze di old_string; se False (default) old_string deve essere univoco nel body, altrimenti errore (evita ambiguità) Usa questo strumento — non upsert_item — quando aggiungi, correggi o annoti un dettaglio su un nodo già esistente: evita di dover rileggere e ricostruire l'intero body, riducendo il rischio di perdere sfumature nella riformulazione. upsert_item resta per creare nodi nuovi o per una riscrittura completa e dichiarata (mai come sotto-prodotto di un update). """ conn = _db() row = conn.execute("SELECT type, body, meta FROM items WHERE id=?", (id,)).fetchone() if row is None: conn.close() raise KeyError(f"Item non trovato: {id}. Usa upsert_item per crearne uno nuovo.") existing_type = row["type"] is_msg_ack = id.startswith(OS_WRITABLE_PREFIX) if existing_type == "os" and id not in OS_WRITABLE and not is_msg_ack: conn.close() raise PermissionError(f"Item '{id}' è di sistema (type='os') e non modificabile via MCP.") body = row["body"] or "" if old_string == "": new_body = body + new_string else: count = body.count(old_string) if count == 0: conn.close() raise ValueError( f"old_string non trovato nel body di '{id}'. Verifica il testo esatto " "(spazi, maiuscole, a-capo)." ) if count > 1 and not replace_all: conn.close() raise ValueError( f"old_string trovato {count} volte nel body di '{id}' — non univoco. " "Aggiungi più contesto per renderlo unico, oppure passa replace_all=True." ) if replace_all: new_body = body.replace(old_string, new_string) else: new_body = body.replace(old_string, new_string, 1) meta_dict = json.loads(row["meta"] or "{}") tipo = meta_dict.get("tipo") if existing_type == "nodo" and tipo not in VAULT_TIPI_UNGATED: if len(new_body.strip()) < 20: conn.close() raise ValueError( f"body risultante troppo corto ({len(new_body.strip())} chars, minimo 20)." ) conn.execute("UPDATE items SET body=? WHERE id=?", (new_body, id)) conn.commit() conn.close() preview = new_body[-200:] if len(new_body) > 200 else new_body return f"OK: {id} aggiornato ({len(body)} -> {len(new_body)} chars). Coda body:\n...{preview}" @mcp.tool() def delete_item(id: str) -> str: """Elimina un item e tutti i suoi edge dal DB.""" conn = _db() deleted = conn.execute("DELETE FROM items WHERE id=?", (id,)).rowcount conn.execute("DELETE FROM edges WHERE src=? OR dst=?", (id, id)) conn.commit() conn.close() if deleted == 0: raise KeyError(f"Item non trovato: {id}") return f"Eliminato: {id}" @mcp.tool() def execute_query(sql: str) -> list[dict]: """ Esegue una query SQL read-only sul vault.db. Solo SELECT. Schema (unica tabella reale): items(id TEXT, type TEXT, body TEXT, file TEXT, meta TEXT, created_at TEXT, updated_at TEXT) Valori di type: 'nodo' — nodi vault; tipo semantico in json_extract(meta,'$.tipo') tipi: cantiere | persona | filo | dinamica | episodio | decisione | entità relazioni in json_extract(meta,'$.relazioni') → array [{id, label}] 'os' — sistema: elon-memory (scrivibile), kernel-elon (scrivibile) Usa json_extract(meta,'$.campo') per leggere campi strutturati. Usa items_fts per full-text: SELECT * FROM items_fts WHERE items_fts MATCH 'query' Esempi: -- Cantieri aperti SELECT id, json_extract(meta,'$.nome') as nome, json_extract(meta,'$.stato') as stato FROM items WHERE type='nodo' AND json_extract(meta,'$.tipo')='cantiere' -- Nodi collegati a X (outgoing) SELECT j.value FROM items, json_each(json_extract(meta,'$.relazioni')) j WHERE id='vendita-palmanova' -- Chi punta a X (incoming) SELECT id FROM items i, json_each(json_extract(i.meta,'$.relazioni')) j WHERE json_extract(j.value,'$.id')='vendita-palmanova' AND type='nodo' """ sql_upper = sql.strip().upper() for forbidden in ("INSERT", "UPDATE", "DELETE", "DROP", "CREATE", "ALTER", "ATTACH"): if re.search(rf"\b{forbidden}\b", sql_upper): raise ValueError(f"Query non consentita: solo SELECT è permesso ({forbidden} rilevato)") conn = _db() try: rows = conn.execute(sql).fetchall() return [dict(r) for r in rows] finally: conn.close() # rag_query rimossa dall'esposizione MCP (Layer 2 sospeso 2026-07-12) if __name__ == "__main__": import uvicorn uvicorn.run(mcp.streamable_http_app(), host="10.0.0.2", port=8765)