# CONNECTOR — Guide d'intégration universel pour Agents IA Ce document est la référence technique à fournir à tout agent IA (Claude Code, Antigravity, Cursor, scripts autonomes, etc.) devant interagir avec le serveur central **Connector** du homelab. --- ## 1. Informations Générales & Endpoints de Base - **Base URL** : `https://connector.arbisa.fr` - **Schéma OpenAPI / Swagger** : `https://connector.arbisa.fr/docs` (ou `/openapi.json`) - **Headers d'authentification** (OBLIGATOIRE sur toutes les routes API et logs) : - `X-Connector-Key: ` - Ou en paramètre d'URL pour le streaming SSE : `?api_key=` - Ou `Authorization: Bearer ` - *Note pour l'IA* : Toutes les requêtes (ingestion de logs, consultation de logs, streaming, tunnel, gestion d'apps) doivent fournir cette clé pour être acceptées par le serveur. --- ## 2. Module Mises à Jour (Self-Update pour les Apps) Chaque application développée pour le homelab doit pouvoir vérifier si une version plus récente est disponible sur Connector et s'auto-mettre à jour. ### 2.1 Endpoint de vérification ```http GET /api/v1/updates/{app_id}/check?current_version={version}&channel=stable ``` #### Paramètres : - `app_id` : Identifiant en minuscules de l'application (ex: `kanri`, `pc-diag`, `neuroghost`). - `current_version` : Version actuelle de l'application cliente (ex: `1.0.0` ou `v1.0.0`). - `channel` : (Optionnel, défaut: `stable`) : `stable`, `beta`, ou `dev`. #### Exemple de réponse JSON : ```json { "app_id": "kanri", "has_update": true, "current_version": "1.0.0", "latest_version": "1.1.0", "channel": "stable", "changelog": "- Correction d'un bug de synchronisation\n- Ajout des logs temps réel", "download_url": "https://connector.arbisa.fr/api/v1/updates/kanri/download/1.1.0", "sha256": "4b227777d4dd1fc61c6f884f48641d02b4d121d3fd328cb08b5531fcacdabf8a", "file_size": 2541280, "is_mandatory": false, "min_app_version": null, "published_at": "2026-08-30T16:00:00Z" } ``` ### 2.2 Algorithme d'Auto-Mise à Jour (Python Ready-to-Use) À copier/coller ou adapter dans n'importe quel projet Python du homelab : ```python import hashlib import json import os import sys import tempfile import urllib.request CONNECTOR_URL = "https://connector.arbisa.fr" def check_and_apply_update(app_id: str, current_version: str, target_file_path: str = None) -> bool: """Vérifie si une mise à jour est disponible et l'applique avec contrôle d'intégrité SHA256.""" url = f"{CONNECTOR_URL}/api/v1/updates/{app_id}/check?current_version={current_version}" try: req = urllib.request.Request(url, headers={"User-Agent": f"{app_id}/{current_version}"}) with urllib.request.urlopen(req, timeout=10) as response: data = json.loads(response.read().decode()) except Exception as e: print(f"[Connector] Impossible de vérifier les mises à jour : {e}") return False if not data.get("has_update") or not data.get("download_url"): print(f"[Connector] {app_id} est à jour (version {current_version}).") return False latest_ver = data["latest_version"] download_url = data["download_url"] expected_sha256 = data.get("sha256") print(f"[Connector] Mise à jour disponible : {current_version} -> {latest_ver}") if data.get("changelog"): print(f"[Connector] Notes de version :\n{data['changelog']}") # Téléchargement dans un fichier temporaire with tempfile.NamedTemporaryFile(delete=False) as tmp_file: tmp_path = tmp_file.name with urllib.request.urlopen(download_url, timeout=60) as resp: hasher = hashlib.sha256() while chunk := resp.read(64 * 1024): hasher.update(chunk) tmp_file.write(chunk) downloaded_sha256 = hasher.hexdigest() # Vérification d'intégrité if expected_sha256 and downloaded_sha256.lower() != expected_sha256.lower(): print(f"[Connector] ERREUR : Checksum invalide ! (Attendu: {expected_sha256}, Reçu: {downloaded_sha256})") os.remove(tmp_path) return False # Remplacement du fichier cible (ex: script en cours ou binaire) if target_file_path: backup_path = f"{target_file_path}.bak" os.replace(target_file_path, backup_path) try: os.replace(tmp_path, target_file_path) print(f"[Connector] Mise à jour {latest_ver} installée avec succès. Redémarrage recommandé.") return True except Exception as err: print(f"[Connector] Échec du remplacement : {err}, restauration du backup...") os.replace(backup_path, target_file_path) return False return True import threading import time def start_autonomous_updater(app_id: str, current_version: str, target_file_path: str, interval_hours: int = 6): """Lance un thread d'arrière-plan totalement autonome qui surveille et applique les mises à jour sans intervention humaine.""" def _loop(): while True: time.sleep(interval_hours * 3600) try: check_and_apply_update(app_id, current_version, target_file_path) except Exception as e: print(f"[AutoUpdater] Erreur lors de la vérification automatique : {e}") t = threading.Thread(target=_loop, daemon=True, name="ConnectorAutonomousUpdater") t.start() ``` --- ## 3. Module Logs & Débogage en Temps Réel Le module Logs permet à une application en production ou en phase de test d'envoyer ses logs, erreurs et tracebacks à Connector, et permet aux IA de surveillance de lire ou streamer ces logs en direct. ### 3.1 Ingestion de logs depuis une application cliente ```http POST /api/v1/logs/ingest Content-Type: application/json ``` #### Payload : ```json { "app_id": "kanri", "level": "ERROR", "message": "Connexion à la base de données impossible", "traceback": "Traceback (most recent call last):\n File 'worker.py', line 42, in connect\n raise ConnectionError('Timeout')", "context": { "host": "node-1", "attempt": 3, "env": "production" } } ``` *Note : Pour envoyer plusieurs logs d'un coup, utilisez `POST /api/v1/logs/ingest-batch` avec `{ "logs": [...] }`.* ### 3.2 Snippet Python : Handler de logs automatique Pour relier automatiquement le module standard `logging` de Python à Connector : ```python import logging import json import urllib.request class ConnectorLogHandler(logging.Handler): def __init__(self, app_id: str, connector_url: str = "https://connector.arbisa.fr"): super().__init__() self.app_id = app_id self.endpoint = f"{connector_url.rstrip('/')}/api/v1/logs/ingest" def emit(self, record): try: tb = self.formatException(record.exc_info) if record.exc_info else None payload = { "app_id": self.app_id, "level": record.levelname, "message": record.getMessage(), "traceback": tb, "context": {"module": record.module, "line": record.lineno} } req = urllib.request.Request( self.endpoint, data=json.dumps(payload).encode("utf-8"), headers={"Content-Type": "application/json"}, method="POST" ) urllib.request.urlopen(req, timeout=2) except Exception: pass # Ne jamais faire planter l'app si Connector est momentanément inaccessible # Utilisation : # logger = logging.getLogger("mon_app") # logger.addHandler(ConnectorLogHandler("mon_app")) # logger.error("Quelque chose a échoué !", exc_info=True) ``` --- ## 4. Consultation et Streaming des Logs par une IA en Débug Lorsqu'un agent IA doit débugger une application du homelab, il peut utiliser l'une des méthodes suivantes : ### 4.1 Consulter l'historique récent des logs (Search & Filter) ```http GET /api/v1/logs?app_id={app_id}&level=ERROR&limit=50 ``` - Récupère les 50 dernières erreurs avec messages et tracebacks complets au format JSON. - Paramètres utiles : `search=mot_cle`, `since=2026-08-30T12:00:00Z`, `level=WARNING`. ### 4.2 Streamer les logs en direct (SSE - Server Sent Events) Idéal pour surveiller en temps réel le comportement d'un script ou d'un conteneur pendant qu'on le teste : ```bash # Dans le terminal ou via curl : curl -N "https://connector.arbisa.fr/api/v1/logs/stream?app_id=kanri&min_level=DEBUG" ``` Chaque événement reçu est préfixé par `event: log` et contient la ligne de log sérialisée en JSON. --- ## 5. Module Événements & Notifications Inter-Applications Pour faire communiquer des applications du homelab ou notifier la fin d'un job : ### 5.1 Publier un événement ```http POST /api/v1/events/publish Content-Type: application/json { "topic": "backup.completed", "app_id": "system", "severity": "info", "data": { "target": "nas_synology", "size_gb": 42.5, "duration_sec": 120 } } ``` ### 5.2 S'abonner aux événements ```bash curl -N "https://connector.arbisa.fr/api/v1/events/stream?topic=backup." ``` --- ## 6. Publication d'une Release (Pour CI/CD ou Agent Déployeur) Pour publier une nouvelle version d'une application (nécessite la clé `X-Connector-Key`) : ```bash curl -X POST "https://connector.arbisa.fr/api/v1/updates/{app_id}/releases" \ -H "X-Connector-Key: VOTRE_CLE_ADMIN" \ -F "version=1.2.0" \ -F "channel=stable" \ -F "changelog=- Ajout de la commande de diagnostic" \ -F "file=@./dist/mon-app-v1.2.0.zip" ``` *Le serveur calcule automatiquement la somme de contrôle SHA256 et la taille du fichier.* --- ## 7. Auto-Mise à Jour de Connector à Distance (Pour IA & CI/CD) Pour mettre à jour le serveur Connector lui-même sans connexion SSH : ```http POST /api/v1/system/self-update Headers: X-Connector-Key: Content-Type: multipart/form-data Body: file: restart: true ``` Ou simplement en exécutant le script `deploy.py` fourni : ```bash python deploy.py --key ``` Le serveur effectue une sauvegarde dans `/app/data/backups/`, applique les nouveaux fichiers et redémarre automatiquement le conteneur. En cas de problème, un rollback est possible via `POST /api/v1/system/rollback`.