# Watches
A **Watch** is a reactive trigger primitive. You arm it on a [Stream](https://docs.ag2.ai/0.13.2/docs/beta/advanced/stream/), and it fires a callback when its condition is met — where "condition" can be an event match, a count, a time window, a schedule, or a composition of other watches.
Watches are the mechanism behind trigger-driven [Observers](https://docs.ag2.ai/0.13.2/docs/beta/advanced/observers/), but they can be used standalone for custom reactive logic on any Stream.
## When to use a Watch [\#](https://docs.ag2.ai/0.13.2/docs/beta/advanced/watches/#when-to-use-a-watch "Permanent link")
Use a Watch when a simple `stream.subscribe(...)` is not enough — i.e. when you need buffering, timing, ordering, or composition:
| You need | Use |
|------------------------------------------------|---------------------------------------------------------------|
| Run a callback on every matching event | `stream.subscribe(fn, condition=...)` (no Watch needed) |
| Fire every N events | `CadenceWatch(n=N, condition=...)` |
| Collect events over a time window | `CadenceWatch(max_wait=seconds, condition=...)` |
| Fire on "N events OR T seconds, whichever first" | `CadenceWatch(n=N, max_wait=seconds, condition=...)` |
| Fire once after a delay | `DelayWatch(seconds)` |
| Fire on a schedule | `IntervalWatch(seconds)` or `CronWatch(expr)` |
| Wait for two separate events in any order | `AllOf(w1, w2)` |
| Wait for an ordered sequence | `Sequence(w1, w2, ...)` |
| Fire on the earliest of several triggers | `AnyOf(w1, w2, ...)` |
## Anatomy of a Watch [\#](https://docs.ag2.ai/0.13.2/docs/beta/advanced/watches/#anatomy-of-a-watch "Permanent link")
Every Watch implements the same `Watch` protocol (importable from `autogen.beta`):
| | |
|-----|-----|
| ```
1
2
3
4
5
6
7
8
9
10
``` | ```
from typing import Protocol
class Watch(Protocol):
@property
def id(self) -> str: ...
@property
def is_armed(self) -> bool: ...
def arm(self, stream, callback) -> None: ...
def disarm(self) -> None: ...
``` |
The callback signature is uniform across all Watch kinds:
| | |
|-----|-----|
| ```
1
2
3
4
``` | ```
from autogen.beta import Context
from autogen.beta.events import BaseEvent
async def callback(events: list[BaseEvent], ctx: Context) -> None: ...
``` |
For event-driven watches (`EventWatch`, `CadenceWatch`, `Sequence`), `events` contains the matched events. For time-driven watches (`DelayWatch`, `IntervalWatch`, `CronWatch`), `events` is empty — the trigger is the timer itself.
## Event-driven Watches [\#](https://docs.ag2.ai/0.13.2/docs/beta/advanced/watches/#event-driven-watches "Permanent link")
### EventWatch [\#](https://docs.ag2.ai/0.13.2/docs/beta/advanced/watches/#eventwatch "Permanent link")
Fires immediately on each matching event.
| | |
|-----|-----|
| ```
1
2
3
4
5
6
7
8
9
10
11
``` | ```
from autogen.beta import MemoryStream
from autogen.beta.watch import EventWatch
from autogen.beta.events import ModelResponse
stream = MemoryStream()
watch = EventWatch(ModelResponse)
async def on_response(events, ctx):
print(f"Model responded: {events[0].content}")
watch.arm(stream, on_response)
``` |
`EventWatch` also supports field conditions and negation:
| | |
|-----|-----|
| ```
1
2
3
4
5
6
7
``` | ```
from autogen.beta.watch import EventWatch
from autogen.beta.events import ToolCallEvent
watch = EventWatch(ToolCallEvent.name == "search")
# Fire on every event that is NOT a ToolCallEvent
watch = EventWatch(~ToolCallEvent)
``` |
### CadenceWatch [\#](https://docs.ag2.ai/0.13.2/docs/beta/advanced/watches/#cadencewatch "Permanent link")
Buffers matching events and fires once the buffer reaches **size**`n`, once **time**`max_wait` seconds have elapsed since the first buffered event, or whichever comes first when both are set. At least one of `n` and `max_wait` is required.
| | |
|-----|-----|
| ```
1
2
3
4
5
6
7
8
9
10
11
12
13
14
``` | ```
from autogen.beta.watch import CadenceWatch
from autogen.beta.events import ModelResponse
# Count-only: fire every 5 model responses
batch = CadenceWatch(n=5, condition=ModelResponse)
# Time-only: flush the buffer once a minute, whenever there's something in it
window = CadenceWatch(max_wait=60.0, condition=ModelResponse)
# Size OR time: fire at 5 events OR 60 seconds after the first event, whichever first
hybrid = CadenceWatch(n=5, max_wait=60.0, condition=ModelResponse)
async def summarize(events, ctx):
print(f"Batched {len(events)} responses")
``` |
The timer starts on the **first** buffered event in a cadence — a quiet stream produces no firings. Any unfilled buffer at disarm-time is discarded.
## Time-driven Watches [\#](https://docs.ag2.ai/0.13.2/docs/beta/advanced/watches/#time-driven-watches "Permanent link")
### DelayWatch [\#](https://docs.ag2.ai/0.13.2/docs/beta/advanced/watches/#delaywatch "Permanent link")
Fires exactly once after a delay, then auto-disarms.
| | |
|-----|-----|
| ```
1
2
3
4
5
6
7
8
9
``` | ```
from autogen.beta.watch import DelayWatch
watch = DelayWatch(30.0)
async def timeout_guard(events, ctx):
# events is always [] for time-driven watches
print("30 seconds elapsed")
watch.arm(stream, timeout_guard)
``` |
### IntervalWatch [\#](https://docs.ag2.ai/0.13.2/docs/beta/advanced/watches/#intervalwatch "Permanent link")
Fires periodically at a fixed interval until disarmed.
| | |
|-----|-----|
| ```
1
2
3
4
5
6
``` | ```
from autogen.beta.watch import IntervalWatch
watch = IntervalWatch(60.0)
async def heartbeat(events, ctx):
print("tick")
``` |
### CronWatch [\#](https://docs.ag2.ai/0.13.2/docs/beta/advanced/watches/#cronwatch "Permanent link")
Fires on a standard 5-field cron expression.
| | |
|-----|-----|
| ```
1
2
3
``` | ```
from autogen.beta.watch import CronWatch
watch = CronWatch("0 9 * * MON") # every Monday at 9am
``` |
Supports `*`, `*/n`, `a-b`, comma lists, and `SUN`–`SAT` day-of-week names.
## Composite Watches [\#](https://docs.ag2.ai/0.13.2/docs/beta/advanced/watches/#composite-watches "Permanent link")
### AllOf [\#](https://docs.ag2.ai/0.13.2/docs/beta/advanced/watches/#allof "Permanent link")
Fires once when **every** sub-watch has fired at least once.
| | |
|-----|-----|
| ```
1
2
3
4
5
6
7
8
9
10
11
``` | ```
from autogen.beta.watch import AllOf, EventWatch
from autogen.beta.events import ModelResponse, ToolCallEvent
watch = AllOf(
EventWatch(ModelResponse),
EventWatch(ToolCallEvent),
)
async def both_seen(events, ctx):
# Gets the combined events from all sub-watches
print(f"Saw both types, got {len(events)} events total")
``` |
After firing, the gate resets — both sub-watches must fire again for the next firing.
### AnyOf [\#](https://docs.ag2.ai/0.13.2/docs/beta/advanced/watches/#anyof "Permanent link")
Fires on **any** sub-watch, every time.
| | |
|-----|-----|
| ```
1
2
3
4
5
6
7
``` | ```
from autogen.beta.watch import AnyOf, EventWatch
from autogen.beta.events import ObserverAlert
watch = AnyOf(
EventWatch(ObserverAlert.severity == "critical"),
EventWatch(ObserverAlert.severity == "fatal"),
)
``` |
### Sequence [\#](https://docs.ag2.ai/0.13.2/docs/beta/advanced/watches/#sequence "Permanent link")
Fires when sub-watches trigger **in order**. Each sub-watch is armed only after the previous has fired.
| | |
|-----|-----|
| ```
1
2
3
4
5
6
7
8
9
10
11
``` | ```
from autogen.beta.watch import Sequence, EventWatch
from autogen.beta.events import ModelRequest, ModelResponse
watch = Sequence(
EventWatch(ModelRequest),
EventWatch(ModelResponse),
)
async def round_trip(events, ctx):
# events contains one ModelRequest followed by one ModelResponse
print("round-trip complete")
``` |
After the last sub-watch fires, the sequence resets.
## Emitting events from a callback [\#](https://docs.ag2.ai/0.13.2/docs/beta/advanced/watches/#emitting-events-from-a-callback "Permanent link")
A Watch callback can send events back onto the stream — this is how [trigger-driven Observers](https://docs.ag2.ai/0.13.2/docs/beta/advanced/observers/#trigger-driven-observers-baseobserver) emit alerts:
| | |
|-----|-----|
| ```
1
2
3
4
5
6
7
8
9
10
11
12
``` | ```
from autogen.beta import MemoryStream
from autogen.beta.watch import DelayWatch
from autogen.beta.events import ObserverAlert, Severity
watch = DelayWatch(30.0)
async def timeout_alert(events, ctx):
await ctx.send(ObserverAlert(
source="timeout-watch",
severity=Severity.WARNING,
message="Agent has been running for >30s.",
))
``` |
Tip
Emitting events from within a callback means other subscribers and watches can react. This is the foundation for composing reactive workflows.
## Arming and lifecycle [\#](https://docs.ag2.ai/0.13.2/docs/beta/advanced/watches/#arming-and-lifecycle "Permanent link")
A Watch is a stateful object. Calling `arm()` on an already-armed Watch will first disarm it, so re-arming is safe. `disarm()` cleans up any subscriptions and cancels any timers.
Most of the time you won't call `arm()` / `disarm()` directly — you'll hand the Watch to a [BaseObserver](https://docs.ag2.ai/0.13.2/docs/beta/advanced/observers/#trigger-driven-observers-baseobserver), which manages the lifecycle against the agent's stream.
## Next steps [\#](https://docs.ag2.ai/0.13.2/docs/beta/advanced/watches/#next-steps "Permanent link")
- Use Watches inside a [BaseObserver](https://docs.ag2.ai/0.13.2/docs/beta/advanced/observers/#trigger-driven-observers-baseobserver) for agent-scoped monitoring.
- See [Stream](https://docs.ag2.ai/0.13.2/docs/beta/advanced/stream/) for event publishing and subscription mechanics.