This is the full learning implementation of the example from the article What the Agent Is Allowed to Do (and What It Is Not).
If you have not read the article yet, start there. Here the focus is only on code: how the system allows or blocks agent actions before execution.
What This Example Demonstrates
- How the system splits actions by levels:
read,write,execute,delete - How the policy gateway checks permissions before each call
- How the least-privilege principle blocks risky actions
- How the agent can continue the task even after part of the steps are blocked
Project Structure
foundations/
βββ allowed-actions/
βββ python/
βββ main.py # model steps + gateway decision log
βββ gateway.py # policy boundary and permission checks
βββ tools.py # tools with different risk levels
βββ requirements.txt
How to Run
1. Clone the repository and go to the folder:
git clone https://github.com/AgentPatterns-tech/agentpatterns.git
cd foundations/allowed-actions/python
2. Install dependencies (this example has no external packages):
pip install -r requirements.txt
3. Run the demo:
python main.py
What We Build in Code
We build a simple policy gateway between the model and tools.
- the model proposes an action (
action+parameters) - the gateway determines action level (
read/write/execute/delete) - policy compares action level with agent permissions
- if the level is denied, the system returns a controlled error and does not execute the tool
Key idea: the model proposes, policy decides.
Code
tools.py - tools with different risk levels
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 (allow or block)
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 with allowed and denied actions
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.
Example Output
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: in output,
historyis intentionally shown in short form -"history": [{...}].
The key point in this demo:read/writepass, whileexecute/deleteare blocked by the policy gateway.
What You See in Practice
| Without policy boundary | With policy boundary | |
|---|---|---|
| Model can run execute/delete | β | β |
| There is an explicit least-privilege rule | β | β |
| There is a controlled fallback for denied action | β | β |
What to Change in This Example
- Allow only
readand verify that evenupdate_user_statusstarts being blocked - Add a separate allowlist for specific actions (not only by level)
- Add a
dry_runmode wherewritepasses validation but does not change state - Add a human-approval step before any
delete
Full Code on GitHub
The repository contains the full version of this demo: access levels, policy boundary, and short-history logging.
View full code on GitHub β