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.

To observe sources and wake an idle Agent, use the Host watch configuration. Host stores events in SQLite and binds this plugin to the active turn.

watch_events

Use it in an Agent

Lifecycle

Star us on GitHub