cerebricks-events 0.1.0

Agent-agnostic event emission SDK for Quadrant Event Manager

Cerebricks Events SDK

Lightweight SDK to emit standardized events to the Event Manager API and compute health across common service targets.

Install (local)

pip install -e .

Or build wheels and install:

python -m build
pip install dist/*.whl

Environment variables:

• EVENT_MANAGER_URL (default set in code)
• CBX_HTTP_TIMEOUT_S (default 10)
• CBX_RETRIES (default 2)

Usage

from cerebricks_events import (
  post_start, post_status_change, post_health, post_error, post_heartbeat,
  compute_health, DEFAULT_TARGETS
)

# Extract headers from request
agent_id = headers.get("agent_id")
session_id = headers.get("session_id")
initiated_by = headers.get("initiatedby")
authtoken = headers.get("authorization")

# Start
post_start(agent_id, session_id, initiated_by, authtoken)

# Heartbeat
post_heartbeat(agent_id, session_id, initiated_by, authtoken,
               status="alive", active_task="My stage2", previous_task="My stage1", nexttaskdelay=10)

# Health
score, condition, unhealthy = compute_health(DEFAULT_TARGETS)
post_health(agent_id, session_id, initiated_by, authtoken, score, condition, unhealthy)

# Status change
post_status_change(agent_id, session_id, initiated_by, authtoken, new_status="Completed", status_message="Done")

# Error
post_error(agent_id, session_id, initiated_by, authtoken, "E001", "Test error")

Health computation

compute_health(targets) probes each URL via HEAD/OPTIONS/GET (any <500 is success; JSON with {status: ok|healthy|pass} counts as ok). Returns (health_score:int, condition:str, unhealthy:list).

License

MIT