Observers, watches, and alerts

When to use

Observability — log model responses, tool calls, token usage.
Runtime safety — block dangerous tool arguments, halt the agent.
Reactive metrics — fire on every Nth response, or every M seconds.
Loop / repetition detection — catch infinite tool-call loops.
Stateful monitoring — anything that needs to remember prior events to decide what to do next.

Two observer shapes

Shape	When	Use
Stateless function	One-off event hook (logging, metrics)	`@observer(EventType)`
Stateful class	Counters / windows / thresholds / composed triggers	Subclass `BaseObserver`

Both are stream subscribers under the hood — registered on the agent rather than directly on the stream.

60-second recipe —

@observer

python

from ag2 import Agent, observer
from ag2.config import OpenAIConfig
from ag2.events import ModelResponse

@observer(ModelResponse)
async def log_response(event: ModelResponse) -> None:
    print(f"Model said: {event.content}")

agent = Agent(
    "assistant",
    config=OpenAIConfig(model="gpt-4o-mini"),
    observers=[log_response],
)

Or attach after construction with

@agent.observer(...)

. Per-call observers also supported (

agent.ask("...", observers=[...])

Observer callbacks support full dependency injection (

Context

Inject

Variable

Depends

). Filter by event type, multiple types (

ModelRequest | ModelResponse

), or field value (

ToolCallEvent.name == "search"

). Use

interrupt=True

to modify or suppress events before regular subscribers see them.

Built-in stateful observers

python

from ag2 import Agent
from ag2.observers import LoopDetector, TokenMonitor

agent = Agent(
    "assistant",
    config=config,
    observers=[
        TokenMonitor(warn_threshold=50_000, alert_threshold=100_000),
        LoopDetector(window_size=10, repeat_threshold=3),
    ],
)

TokenMonitor
— tracks cumulative tokens across
```
ModelResponse
```
and
```
TaskCompleted
```
. Emits
```
WARNING
```
/
```
CRITICAL
```
```
ObserverAlert
```
s as thresholds are crossed. Read state via
```
monitor.total_tokens
```
.
LoopDetector
— sliding window of recent tool calls. Emits a
```
WARNING
```
alert when
```
repeat_threshold
```
consecutive identical calls are seen.

Custom

BaseObserver

BaseObserver

pairs a

Watch

(when to fire) with a

process()

method (what to do):

python

from ag2 import Context
from ag2.observers import BaseObserver
from ag2.watch import CadenceWatch
from ag2.events import BaseEvent, ModelResponse
from ag2.events.alert import ObserverAlert, Severity

class AvgCompletionObserver(BaseObserver):
    """Every N responses, emit an INFO alert with avg completion-token count."""

    def __init__(self, window: int = 5) -> None:
        super().__init__("avg-completion", watch=CadenceWatch(n=window, condition=ModelResponse))
        self._window = window

    async def process(self, events: list[BaseEvent], ctx: Context) -> ObserverAlert | None:
        tokens = [e.usage.completion_tokens for e in events if isinstance(e, ModelResponse) and e.usage]
        if not tokens:
            return None
        return ObserverAlert(
            source=self.name,
            severity=Severity.INFO,
            message=f"Avg completion tokens over last {self._window}: {sum(tokens) / len(tokens):.0f}",
        )

process()

returns an

ObserverAlert

, the base class emits it onto the stream. You can also send events manually via

await ctx.send(...)

Watch primitives — picking when to fire

You need	Use
Every matching event	`EventWatch(EventType)` or just `stream.subscribe(fn, condition=...)`
Every N matching events	`CadenceWatch(n=N, condition=EventType)`
Every T seconds (buffered events)	`CadenceWatch(max_wait=T, condition=EventType)`
Either threshold	`CadenceWatch(n=N, max_wait=T, condition=EventType)`
Once after delay	`DelayWatch(seconds)`
Periodic timer	`IntervalWatch(seconds)`
Cron schedule	`CronWatch("0 9 * * MON")`
All sub-watches must fire	`AllOf(w1, w2)`
Any sub-watch fires	`AnyOf(w1, w2)`
In order	`Sequence(w1, w2)`

All importable from

ag2.watch

. Callback signature is uniform:

async def cb(events: list[BaseEvent], ctx: Context) -> None

. Time-driven watches pass

events=[]

ObserverAlert

— the alert type

python

from ag2.events.alert import ObserverAlert, Severity

ObserverAlert(
    source="my-observer",
    severity=Severity.WARNING,    # INFO, WARNING, CRITICAL, FATAL
    message="What happened",
)

Important:

ObserverAlert

is on the stream and persisted in history, but the default provider mappers do not render it back to the LLM. To make the agent see alerts, add

AlertPolicy()

assembly=[...]

python

from ag2.policies import AlertPolicy
agent = Agent("assistant", config=config, assembly=[AlertPolicy()])

FATAL alerts →

HaltEvent

→ short-circuit

AlertPolicy

does two things on

Severity.FATAL

Emits a
```
HaltEvent
```
on the stream.
Appends a halt notice to the system prompt.

When

assembly=[...]

is non-empty, the harness automatically wires

_HaltCheckMiddleware

which sees the

HaltEvent

and short-circuits the next LLM call with a synthetic

HALTED: ...

response.

python

from ag2 import Context
from ag2.observers import BaseObserver
from ag2.events import BaseEvent, ToolCallEvent
from ag2.events.alert import HaltEvent, ObserverAlert, Severity
from ag2.policies import AlertPolicy
from ag2.watch import EventWatch

class PathGuardian(BaseObserver):
    def __init__(self) -> None:
        super().__init__("path-guardian", watch=EventWatch(ToolCallEvent))

    async def process(self, events: list[BaseEvent], ctx: Context) -> ObserverAlert | None:
        for event in events:
            if not isinstance(event, ToolCallEvent) or event.name != "write_file":
                continue
            if "/etc/" in event.arguments or "/usr/" in event.arguments:
                return ObserverAlert(
                    source=self.name,
                    severity=Severity.FATAL,
                    message=f"blocked dangerous write: {event.arguments}",
                )
        return None

agent = Agent(
    "safe-shell",
    prompt="...",
    config=config,
    tools=[write_file],
    observers=[PathGuardian()],
    assembly=[AlertPolicy()],   # routes FATAL → HaltEvent
)

The first dangerous tool call triggers FATAL → halt; the agent's next ask is short-circuited. Full runnable demo:

assets/safety_guard.py

Subscribing to alerts and halts from outside

python

from ag2 import MemoryStream
from ag2.events.alert import HaltEvent, ObserverAlert

stream = MemoryStream()
stream.where(ObserverAlert).subscribe(lambda e: print(f"[{e.severity}] {e.source}: {e.message}"))
stream.where(HaltEvent).subscribe(lambda e: print(f"HALT: {e.reason}"))
await agent.ask("...", stream=stream)

Observers vs Middleware vs Stream subscribers

Feature	Observer	Middleware	Stream subscriber
Registered on	Agent	Agent	Stream
Lifecycle	Scoped to execution	Scoped to execution	Manual
Boilerplate	Function (or `BaseObserver` )	`BaseMiddleware` class	Function
Can modify events	`interrupt=True`	Yes (wraps execution)	`interrupt=True`
DI support	Yes	Yes	Yes
Use case	Monitoring, metrics, alerts	Cross-cutting (retry, auth, rate limit)	Low-level event wiring

Going deeper

assets/token_watchdog.py

— three observers (

TokenMonitor

LoopDetector

, custom

AlertConsole

) on one agent. Mirrors

code_examples/04

assets/safety_guard.py

—

PathGuardian

→ FATAL →

AlertPolicy

→

HaltEvent

→ short-circuit. Mirrors

code_examples/08

Source docs:

website/docs/user-guide/advanced/observers.mdx

—

@observer

BaseObserver

, registration, built-ins,

ObserverAlert

```
website/docs/user-guide/advanced/watches.mdx
```
— every Watch primitive, composition rules.

website/docs/user-guide/advanced/stream.mdx

— Stream API,

where

subscribe

, interrupters,

RedisStream

website/docs/user-guide/advanced/assembly.mdx

—

AlertPolicy

ordering and dedup.

Common pitfalls

Alerts not reaching the model —
```
ObserverAlert
```
events are on the stream but invisible to the LLM by default. Add
```
AlertPolicy()
```
to
```
assembly=[...]
```
.
FATAL not halting —
```
AlertPolicy
```
is what creates
```
HaltEvent
```
. Without
```
assembly=[..., AlertPolicy(), ...]
```
(or any non-empty assembly chain enabling
```
_HaltCheckMiddleware
```
), nothing halts.
Sharing one
AlertPolicy()
across agents — dedup state lives on the instance. Give each agent its own.
Watch callback assumes
events
is non-empty — for time-driven watches (
```
DelayWatch
```
,
```
IntervalWatch
```
,
```
CronWatch
```
),
```
events
```
is always
```
[]
```
.
Forgetting
process()
is async —
```
BaseObserver.process
```
must be
```
async def
```
.
Subscribing with
subscribe(fn)
when you wanted
subscribe()
decorator — both work; the bare-call form is
```
stream.subscribe(fn)
```
, the decorator form is
```
@stream.subscribe()
```
(with parens).
CadenceWatch
with no
n
and no
max_wait
— raises
```
ValueError
```
; at least one is required.

ag2-observers-and-alerts

NPX Install

Tags

SKILL.md Content

Observers, watches, and alerts

When to use

Two observer shapes

60-second recipe —
`@observer`

Built-in stateful observers

Custom
`BaseObserver`

Watch primitives — picking when to fire

`ObserverAlert`
— the alert type

FATAL alerts →
`HaltEvent`
→ short-circuit

Subscribing to alerts and halts from outside

Observers vs Middleware vs Stream subscribers

Going deeper

Common pitfalls