Sends a structured event with type, timestamp, and optional data to a registered callback, enabling external monitoring or logging of agent activity.
def _emit_event(self, event_type: str, data: dict = None):
"""Emit event"""
if self.on_event:
try:
self.on_event({
"type": event_type,
"timestamp": time.time(),
"data": data or {}
})
except Exception as e:
logger.error(f"Event callback error: {e}")
See the full source, get the GitHub permalink, and search 40K more like it.
Get a free API key