Dies ist die vollstandige Lernimplementierung des Beispiels aus dem Artikel Was der Agent tun darf (und was nicht).
Wenn du den Artikel noch nicht gelesen hast, beginne dort. Hier liegt der Fokus nur auf Code: wie das System Agent-Aktionen vor der Ausfuhrung erlaubt oder blockiert.
Was Dieses Beispiel Zeigt
- Wie das System Aktionen nach Ebenen trennt:
read,write,execute,delete - Wie das Policy-Gateway Berechtigungen vor jedem Aufruf pruft
- Wie das Least-Privilege-Prinzip riskante Aktionen blockiert
- Wie der Agent die Aufgabe fortsetzen kann, selbst wenn Teile der Schritte blockiert werden
Projektstruktur
foundations/
└── allowed-actions/
└── python/
├── main.py # Modellschritte + Log der Gateway-Entscheidungen
├── gateway.py # policy boundary und Berechtigungsprufung
├── tools.py # Tools mit unterschiedlichen Risikostufen
└── requirements.txt
Ausfuhren
1. Repository klonen und in den Ordner wechseln:
git clone https://github.com/AgentPatterns-tech/agentpatterns.git
cd foundations/allowed-actions/python
2. Abhangigkeiten installieren (dieses Beispiel hat keine externen Pakete):
pip install -r requirements.txt
3. Demo starten:
python main.py
Was Wir Im Code Bauen
Wir bauen ein einfaches Policy-Gateway zwischen Modell und Tools.
- das Modell schlagt eine Aktion vor (
action+parameters) - das Gateway bestimmt die Aktionsebene (
read/write/execute/delete) - policy vergleicht Aktionsebene mit Agent-Berechtigungen
- wenn die Ebene gesperrt ist, gibt das System einen kontrollierten Fehler zuruck und fuhrt das Tool nicht aus
Kernidee: das Modell schlagt vor, policy entscheidet.
Code
tools.py - Tools mit unterschiedlichen Risikostufen
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 (erlauben oder blockieren)
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 - Szenario mit erlaubten und gesperrten Aktionen
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.
Beispielausgabe
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.
Hinweis: in der Ausgabe wird
historyabsichtlich in Kurzform gezeigt -"history": [{...}].
Wichtig in dieser Demo:read/writegehen durch, wahrendexecute/deletevom Policy-Gateway blockiert werden.
Was Man In Der Praxis Sieht
| Ohne policy boundary | Mit policy boundary | |
|---|---|---|
| Modell kann execute/delete ausfuhren | ✅ | ❌ |
| Es gibt eine explizite least-privilege Regel | ❌ | ✅ |
| Es gibt einen kontrollierten Fallback bei gesperrter Aktion | ❌ | ✅ |
Was Du In Diesem Beispiel Andern Kannst
- Erlaube nur
readund prufe, dass selbstupdate_user_statusblockiert wird - Fuge eine separate Allowlist fur konkrete Aktionen hinzu (nicht nur nach Ebene)
- Fuge einen
dry_run-Modus hinzu, in demwritedie Validierung passiert, aber keinen Zustand andert - Fuge einen Human-Approval-Schritt vor jedem
deletehinzu
Vollstandiger Code auf GitHub
Im Repository liegt die vollstandige Version dieser Demo: Zugriffsebenen, policy boundary und short-history Logging.
Vollstandigen Code auf GitHub ansehen ↗