Safety Guard - AG2

08 · Safety Guard

A custom BaseObserver (PathGuardian) watches every tool call and emits a Severity.FATAL ObserverAlert when an Agent tries to write to a forbidden path like /etc/. The alert routes through AlertPolicy → HaltEvent → _HaltCheckMiddleware, which short-circuits the next LLM call with a synthetic HALTED: ... response. The first ask (writing to /tmp/...) succeeds; the second (writing to /etc/passwd) is blocked end-to-end.

What it covers

Building a BaseObserver that watches a specific event type (here ToolCallEvent) via EventWatch.
Returning an ObserverAlert(severity=Severity.FATAL, ...) from an observer to signal a hard-stop condition.
How AlertPolicy (in the assembly chain) translates a FATAL alert into a HaltEvent and appends a halt notice to the system prompt.
How _HaltCheckMiddleware (auto-wired when assembly is non-empty) sees the HaltEvent and short-circuits the next LLM call.
Subscribing to HaltEvent and ObserverAlert from outside the Agent to verify the halt fired.

Primitives covered

BaseObserver + EventWatch(ToolCallEvent)
ObserverAlert with Severity.FATAL
AlertPolicy in the assembly= chain
Auto-wired _HaltCheckMiddleware (no explicit middleware setup needed)
HaltEvent lifecycle event

Source

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 ```
"""08 · Safety guard — FATAL alert halts the Agent
A hand-rolled BaseObserver watches every tool call and flags anything
that looks dangerous (here: a write_file tool asked to touch
/etc/). It emits a Severity.FATAL ObserverAlert. The flow from
there is fully wired by the framework:
1. The alert lands on the agent's stream.
2. AlertPolicy (an assembly policy) picks it up before the next LLM
call, emits a HaltEvent on the stream, and appends a halt notice
to the system prompt.
3. _HaltCheckMiddleware (wired in automatically when assembly is
non-empty) sees the HaltEvent and short-circuits the LLM call with
a synthetic HALTED: ... response.
Run::
.venv-beta/bin/python 08_safety_guard.py
"""
import asyncio
from autogen.beta import Agent
from autogen.beta import Context
from autogen.beta.config import GeminiConfig
from autogen.beta.events import BaseEvent, ToolCallEvent, HaltEvent, ObserverAlert, Severity
from autogen.beta.observer import BaseObserver
from autogen.beta.policies import AlertPolicy
from autogen.beta.stream import MemoryStream
from autogen.beta.watch import EventWatch
def section(title: str) -> None:
print(f"\n── {title} ───")
# ---- Tool under supervision -------------------------------------------------
def write_file(path: str, content: str) -> str:
"""Pretend-write content to path. This playground never touches disk."""
return f"[ok] wrote {len(content)} bytes to {path}"
# ---- Guardian observer ------------------------------------------------------
class PathGuardian(BaseObserver):
"""Emits a FATAL alert if anything tries to write outside /tmp."""
def init(self) -> None:
super().init("path-guardian", watch=EventWatch(ToolCallEvent))
async def process(self, events: list[BaseEvent], ctx: Context) -> ObserverAlert


<br> 1<br> 2<br> 3<br> 4<br> 5<br> 6<br> 7<br> 8<br> 9<br> 10<br> 11<br> 12<br> 13<br> 14<br> 15<br> 16<br> 17<br> 18<br> 19<br> 20<br> 21<br> 22<br> 23<br> 24<br> 25<br> 26<br> 27<br> 28<br> 29<br> 30<br> 31<br> 32<br> 33<br> 34<br> 35<br> 36<br> 37<br> 38<br> 39<br> 40<br> 41<br> 42<br> 43<br> 44<br> 45<br> 46<br> 47<br> 48<br> 49<br> 50<br> 51<br> 52<br> 53<br> 54<br> 55<br> 56<br> 57<br> 58<br> 59<br> 60<br> 61<br> 62<br> 63<br> 64<br> 65<br> 66<br> 67<br> 68<br> 69<br> 70<br> 71<br> 72<br> 73<br> 74<br> 75<br> 76<br> 77<br> 78<br> 79<br> 80<br> 81<br> 82<br> 83<br> 84<br> 85<br> 86<br> 87<br> 88<br> 89<br> 90<br> 91<br> 92<br> 93<br> 94<br> 95<br> 96<br> 97<br> 98<br> 99<br>100<br>101<br>102<br>103<br>104<br>105<br>106<br>107<br>108<br>109<br>110<br>	``` """08 · Safety guard — FATAL alert halts the Agent A hand-rolled `BaseObserver` watches every tool call and flags anything that looks dangerous (here: a `write_file` tool asked to touch `/etc/`). It emits a `Severity.FATAL` `ObserverAlert`. The flow from there is fully wired by the framework: 1. The alert lands on the agent's stream. 2. `AlertPolicy` (an assembly policy) picks it up before the next LLM call, emits a `HaltEvent` on the stream, and appends a halt notice to the system prompt. 3. `_HaltCheckMiddleware` (wired in automatically when `assembly` is non-empty) sees the `HaltEvent` and short-circuits the LLM call with a synthetic `HALTED: ...` response. Run:: .venv-beta/bin/python 08_safety_guard.py """ import asyncio from autogen.beta import Agent from autogen.beta import Context from autogen.beta.config import GeminiConfig from autogen.beta.events import BaseEvent, ToolCallEvent, HaltEvent, ObserverAlert, Severity from autogen.beta.observer import BaseObserver from autogen.beta.policies import AlertPolicy from autogen.beta.stream import MemoryStream from autogen.beta.watch import EventWatch def section(title: str) -> None: print(f"\n── {title} ───") # ---- Tool under supervision ------------------------------------------------- def write_file(path: str, content: str) -> str: """Pretend-write `content` to `path`. This playground never touches disk.""" return f"[ok] wrote {len(content)} bytes to {path}" # ---- Guardian observer ------------------------------------------------------ class PathGuardian(BaseObserver): """Emits a FATAL alert if anything tries to write outside /tmp.""" def init(self) -> None: super().init("path-guardian", watch=EventWatch(ToolCallEvent)) async def process(self, events: list[BaseEvent], ctx: Context) -> ObserverAlert