Python Integration

import httpx
import threading

MUSTER_URL = "https://app.getmuster.io"

def muster_emit(job_id: str, agent_id: str, checks: list, **kwargs):
    """Fire-and-forget quality signal. Never blocks, never raises."""
    def _send():
        try:
            httpx.post(
                f"{MUSTER_URL}/api/v1/jobs/{job_id}/quality",
                json={"agent_id": agent_id, "job_id": job_id,
                      "overall_passed": all(c.get("passed", True) for c in checks),
                      "checks": checks, **kwargs},
                timeout=2.0,
            )
        except Exception:
            pass
    threading.Thread(target=_send, daemon=True).start()

muster_emit(
    job_id=job_id,
    agent_id="invoice-processor-v2",
    checks=[
        {"check_id": "output_not_empty",    "severity": "HIGH", "passed": bool(output)},
        {"check_id": "subtotal_arithmetic", "severity": "HIGH",
         "passed": abs(computed - declared) < 0.01,
         "expected": str(declared), "actual": str(computed)},
    ],
    token_input=1247,
    token_output=312,
    model="gpt-4o",
    latency_ms=1840,
)

pip install muster-sdk

from muster_sdk import beacon, quality

beacon.register(agent_id="invoice-processor-v2")

quality.emit(
    job_id=job_id,
    checks=[
        quality.Check("output_not_empty",    "HIGH", bool(output)),
        quality.Check("subtotal_arithmetic", "HIGH",
                      abs(computed - declared) < 0.01,
                      expected=str(declared), actual=str(computed)),
    ],
    token_input=1247,
    token_output=312,
    model="gpt-4o",
)