53 lines
2.1 KiB
Python
53 lines
2.1 KiB
Python
import json
|
|
import sqlite3
|
|
from typing import List
|
|
|
|
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"],
|
|
"DSH": ["infra", "llm", "nyora", "coding"],
|
|
"NABIL": ["infra", "llm", "nyora", "perso", "tt", "coding"],
|
|
}
|
|
|
|
# "coding" est transverse (memory_entries), pas dans ALL_SCOPES qui reste les
|
|
# scopes "rules" historiques (dict plat) ; get_agent_config/get_my_context ne
|
|
# doivent pas essayer de charger "coding" comme un scope rules classique.
|
|
ALL_SCOPES = ["infra", "llm", "nyora", "perso", "tt"]
|
|
MEMORY_SCOPES = ["coding"]
|
|
|
|
|
|
def get_allowed_scopes(agent_name: str) -> List[str]:
|
|
"""
|
|
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)
|