Esta es la implementacion educativa completa del ejemplo del articulo Que puede hacer el agente (y que no).
Si aun no leiste el articulo, empieza por ahi. Aqui el foco es solo en codigo: como el sistema permite o bloquea acciones del agente antes de ejecutar.
Que Demuestra Este Ejemplo
- Como el sistema divide acciones por niveles:
read,write,execute,delete - Como el policy gateway verifica permisos antes de cada llamada
- Como el principio de least privilege bloquea acciones riesgosas
- Como el agente puede continuar la tarea incluso cuando parte de los pasos quedan bloqueados
Estructura del Proyecto
foundations/
└── allowed-actions/
└── python/
├── main.py # pasos del modelo + log de decisiones del gateway
├── gateway.py # policy boundary y verificacion de permisos
├── tools.py # herramientas con diferentes niveles de riesgo
└── requirements.txt
Como Ejecutar
1. Clona el repositorio y entra en la carpeta:
git clone https://github.com/AgentPatterns-tech/agentpatterns.git
cd foundations/allowed-actions/python
2. Instala dependencias (este ejemplo no tiene paquetes externos):
pip install -r requirements.txt
3. Ejecuta la demo:
python main.py
Que Construimos en el Codigo
Construimos un policy gateway simple entre el modelo y las herramientas.
- el modelo propone una accion (
action+parameters) - el gateway define el nivel de accion (
read/write/execute/delete) - policy compara el nivel de accion con los permisos del agente
- si el nivel esta bloqueado, el sistema devuelve un error controlado y no ejecuta la herramienta
Idea clave: el modelo propone y policy decide.
Codigo
tools.py - herramientas con diferentes niveles de riesgo
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 (permitir o bloquear)
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 - escenario con acciones permitidas y bloqueadas
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.
Ejemplo de Salida
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.
Nota: en la salida,
historyse muestra intencionalmente en forma corta -"history": [{...}].
Punto clave de esta demo:read/writepasan, mientrasexecute/deleteson bloqueadas por el policy gateway.
Lo Que Se Ve en la Practica
| Sin policy boundary | Con policy boundary | |
|---|---|---|
| El modelo puede ejecutar execute/delete | ✅ | ❌ |
| Hay una regla explicita de least privilege | ❌ | ✅ |
| Hay fallback controlado para accion bloqueada | ❌ | ✅ |
Que Cambiar en Este Ejemplo
- Permite solo
ready verifica que inclusoupdate_user_statusempiece a bloquearse - Agrega un allowlist separado por acciones concretas (no solo por nivel)
- Agrega modo
dry_run, dondewritepasa validacion pero no cambia estado - Agrega un paso de human approval antes de cualquier
delete
Codigo Completo en GitHub
En el repositorio esta la version completa de esta demo: niveles de acceso, policy boundary y logging de short-history.
Ver codigo completo en GitHub ↗