watch_events

Give a running Agent fresh event data at the next iteration boundary.

Use it in an Agent

main.py
python
from connectonion import Agent
from connectonion.useful_plugins import watch_events

# claim_events() atomically returns [{"id": "stable-id", "content": "What changed"}]
agent = Agent("worker", plugins=[watch_events(claim_events)])

Copy the implementation with co copy watch_events to edit how your Agent receives and formats in-flight observations.

Lifecycle

The plugin checks at before_iteration. If an event arrives during the final model call, after_iteration requests one more iteration. It runs only while Agent.input() is active.

A session-owned runner must observe sources and wake an idle Agent. That runtime is tracked in issue #1788. This plugin only handles events during an active turn.