Ceci est l implementation pedagogique complete de l exemple de l article Ce que l agent peut faire (et ne peut pas).
Si vous n avez pas encore lu l article, commencez par la. Ici, le focus est uniquement sur le code : comment le systeme autorise ou bloque les actions de l agent avant execution.
Ce Que Cet Exemple Demontre
- Comment le systeme separe les actions par niveaux :
read,write,execute,delete - Comment le policy gateway verifie les droits avant chaque appel
- Comment le principe de least privilege bloque les actions risquées
- Comment l agent peut continuer la tache meme apres blocage d une partie des etapes
Structure du Projet
foundations/
└── allowed-actions/
└── python/
├── main.py # etapes du modele + journal des decisions du gateway
├── gateway.py # policy boundary et verification des permissions
├── tools.py # outils avec differents niveaux de risque
└── requirements.txt
Comment Executer
1. Clone le depot et va dans le dossier :
git clone https://github.com/AgentPatterns-tech/agentpatterns.git
cd foundations/allowed-actions/python
2. Installe les dependances (cet exemple n a pas de packages externes) :
pip install -r requirements.txt
3. Lance la demo :
python main.py
Ce Que Nous Construisons Dans Le Code
Nous construisons un policy gateway simple entre le modele et les outils.
- le modele propose une action (
action+parameters) - le gateway determine le niveau de l action (
read/write/execute/delete) - policy compare le niveau de l action avec les permissions de l agent
- si le niveau est interdit, le systeme renvoie une erreur controlee et n execute pas l outil
Idee cle : le modele propose, policy decide.
Code
tools.py - outils avec differents niveaux de risque
from typing import Any
USERS: dict[int, dict[str, Any]] = {
42: {"id": 42, "name": "Anna", "status": "active"},
}
def read_user(user_id: int) -> dict[str, Any]:
user = USERS.get(user_id)
if not user:
return {"ok": False, "error": f"user {user_id} not found"}
return {"ok": True, "user": dict(user)}
def update_user_status(user_id: int, status: str) -> dict[str, Any]:
user = USERS.get(user_id)
if not user:
return {"ok": False, "error": f"user {user_id} not found"}
user["status"] = status
return {"ok": True, "user": dict(user)}
def send_webhook(event: str) -> dict[str, Any]:
return {"ok": True, "sent": event}
def delete_user(user_id: int) -> dict[str, Any]:
if user_id not in USERS:
return {"ok": False, "error": f"user {user_id} not found"}
del USERS[user_id]
return {"ok": True, "deleted": user_id}
gateway.py - policy boundary (autoriser ou bloquer)
from typing import Any
from tools import delete_user, read_user, send_webhook, update_user_status
TOOL_REGISTRY = {
"read_user": read_user,
"update_user_status": update_user_status,
"send_webhook": send_webhook,
"delete_user": delete_user,
}
TOOL_LEVEL = {
"read_user": "read",
"update_user_status": "write",
"send_webhook": "execute",
"delete_user": "delete",
}
# Least-privilege policy for this agent.
AGENT_ALLOWED_LEVELS = {"read", "write"}
def execute_action(call: dict[str, Any], history: list[dict[str, Any]]) -> dict[str, Any]:
action = str(call.get("action") or "")
params = call.get("parameters") or {}
level = TOOL_LEVEL.get(action, "unknown")
history.append({"action": action, "level": level, "status": "requested"})
tool = TOOL_REGISTRY.get(action)
if tool is None:
history.append({"action": action, "level": level, "status": "blocked"})
return {
"ok": False,
"action": action,
"error": f"action '{action}' is not found",
"history": list(history),
}
if level not in AGENT_ALLOWED_LEVELS:
history.append({"action": action, "level": level, "status": "blocked"})
return {
"ok": False,
"action": action,
"error": f"action '{action}' is blocked by policy (level={level})",
"history": list(history),
}
result = tool(**params)
history.append({"action": action, "level": level, "status": "allowed"})
return {
"ok": True,
"action": action,
"result": result,
"history": list(history),
}
main.py - scenario avec actions autorisees et bloquees
import json
from gateway import execute_action
MODEL_CALLS = [
{"action": "read_user", "parameters": {"user_id": 42}},
{"action": "send_webhook", "parameters": {"event": "user-reviewed"}},
{"action": "update_user_status", "parameters": {"user_id": 42, "status": "paused"}},
{"action": "delete_user", "parameters": {"user_id": 42}},
]
def compact_result(execution: dict) -> str:
base = (
'{'
f'"ok": {str(bool(execution.get("ok"))).lower()}, '
f'"action": {json.dumps(execution.get("action"), ensure_ascii=False)}, '
'"history": [{...}]'
)
if execution.get("ok"):
return (
base
+ ', '
+ '"result": '
+ json.dumps(execution.get("result"), ensure_ascii=False)
+ '}'
)
return (
base
+ ', '
+ '"error": '
+ json.dumps(execution.get("error"), ensure_ascii=False)
+ '}'
)
def run() -> None:
history: list[dict] = []
for step, call in enumerate(MODEL_CALLS, start=1):
print(f"\n=== STEP {step} ===")
print("Model call:", json.dumps(call, ensure_ascii=False))
execution = execute_action(call, history)
print("Gateway result:", compact_result(execution))
print("\nDone: policy boundary demonstrated.")
if __name__ == "__main__":
run()
requirements.txt
# No external dependencies for this learning example.
Exemple de Sortie
python main.py
=== STEP 1 ===
Model call: {"action": "read_user", "parameters": {"user_id": 42}}
Gateway result: {"ok": true, "action": "read_user", "history": [{...}], "result": {"ok": true, "user": {"id": 42, "name": "Anna", "status": "active"}}}
=== STEP 2 ===
Model call: {"action": "send_webhook", "parameters": {"event": "user-reviewed"}}
Gateway result: {"ok": false, "action": "send_webhook", "history": [{...}], "error": "action 'send_webhook' is blocked by policy (level=execute)"}
=== STEP 3 ===
Model call: {"action": "update_user_status", "parameters": {"user_id": 42, "status": "paused"}}
Gateway result: {"ok": true, "action": "update_user_status", "history": [{...}], "result": {"ok": true, "user": {"id": 42, "name": "Anna", "status": "paused"}}}
=== STEP 4 ===
Model call: {"action": "delete_user", "parameters": {"user_id": 42}}
Gateway result: {"ok": false, "action": "delete_user", "history": [{...}], "error": "action 'delete_user' is blocked by policy (level=delete)"}
Done: policy boundary demonstrated.
Note : dans la sortie,
historyest volontairement affiche en forme courte -"history": [{...}].
Point cle de cette demo :read/writepassent, tandis queexecute/deletesont bloques par le policy gateway.
Ce Que L on Voit En Pratique
| Sans policy boundary | Avec policy boundary | |
|---|---|---|
| Le modele peut lancer execute/delete | ✅ | ❌ |
| Il y a une regle explicite de least privilege | ❌ | ✅ |
| Il y a un fallback controle quand une action est interdite | ❌ | ✅ |
Ce Qu Il Faut Changer Dans Cet Exemple
- Autorise seulement
readet verifie que memeupdate_user_statuscommence a etre bloque - Ajoute une allowlist separee par actions concretes (pas seulement par niveau)
- Ajoute un mode
dry_runouwritepasse la validation mais ne change pas l etat - Ajoute une etape de human approval avant tout
delete
Code Complet sur GitHub
Le depot contient la version complete de cette demo : niveaux d acces, policy boundary et journalisation short-history.
Voir le code complet sur GitHub ↗