feat(agents): has_scope_access lit desormais la table agents (DB = source de verite, plus de redeploiement pour ajouter un agent)

This commit is contained in:
2026-07-30 08:40:51 +00:00
parent 9a881ebd1b
commit 5fd3c4ff02
+34 -8
View File
@@ -1,14 +1,20 @@
import json
import sqlite3
from typing import List
# Scopes matrix according to prompt.md
# agent -> list of allowed scopes
AGENT_SCOPES = {
DB_PATH = "data/context_hub.db"
# Filet de securite si la table agents est absente/vide (bootstrap avant premier
# seed, ou table non encore migree) -- la table agents (SQLite) est la source
# de verite reelle ; ce dict n'est qu'un repli pour ne jamais faire echouer une
# verification d'acces si la DB n'est pas encore prete.
_FALLBACK_AGENT_SCOPES = {
"CLAUDE": ["infra", "llm", "nyora", "perso", "tt", "coding"],
"HERMES_TT": ["infra", "llm", "tt"],
"HERMES_NYORA": ["infra", "llm", "nyora"],
"HERMES_PERSO": ["infra", "llm", "nyora", "perso"],
"GEMINI": ["infra", "llm", "nyora", "coding"],
"NABIL": ["infra", "llm", "nyora", "perso", "tt", "coding"] # Assuming Nabil has access to all
"NABIL": ["infra", "llm", "nyora", "perso", "tt", "coding"],
}
# "coding" est transverse (memory_entries), pas dans ALL_SCOPES qui reste les
@@ -17,9 +23,29 @@ AGENT_SCOPES = {
ALL_SCOPES = ["infra", "llm", "nyora", "perso", "tt"]
MEMORY_SCOPES = ["coding"]
def has_scope_access(agent_name: str, scope: str) -> bool:
allowed_scopes = AGENT_SCOPES.get(agent_name.upper(), [])
return scope in allowed_scopes
def get_allowed_scopes(agent_name: str) -> List[str]:
return AGENT_SCOPES.get(agent_name.upper(), [])
"""
Lecture synchrone directe de la table `agents` (SQLite tolere plusieurs
lecteurs concurrents sans conflit). Volontairement synchrone : has_scope_access
est appele dans une douzaine d'endroits (api.py, mcp.py) qui ne sont pas tous
async-friendly a moindre risque -- garder cette fonction sync evite de toucher
chaque appelant. Le cout d'une connexion sqlite3 par appel est negligeable
pour un si petit fichier et un si faible volume de requetes.
"""
agent_name = agent_name.upper()
try:
con = sqlite3.connect(DB_PATH)
cur = con.cursor()
cur.execute("SELECT scopes FROM agents WHERE agent_name=?", (agent_name,))
row = cur.fetchone()
con.close()
if row:
return json.loads(row[0])
except Exception:
pass
return _FALLBACK_AGENT_SCOPES.get(agent_name, [])
def has_scope_access(agent_name: str, scope: str) -> bool:
return scope in get_allowed_scopes(agent_name)