1. Sign up
Go to app.getmuster.io and create your account. Your 14-day free trial starts immediately — no credit card required.
2. Add your agent
From the dashboard, click Add Agent and fill in the details. Muster assigns a risk score automatically.
3. Send your first signal
After each job completes, POST quality signals to Muster. No SDK, no new dependencies — just an HTTP call.
import httpx, threading
def report_to_muster(job_id, checks):
try:
httpx.post(
f"https://app.getmuster.io/api/v1/jobs/{job_id}/quality",
json={
"agent_id": "invoice-processor-v2",
"job_id": job_id,
"overall_passed": all(c["passed"] for c in checks),
"checks": checks,
},
timeout=2.0,
)
except Exception:
pass # never block your agent
# Fire and forget — don't await this
threading.Thread(
target=report_to_muster,
args=(job_id, [
{"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)},
]),
daemon=True,
).start()
4. View your data
Head to the Health Heatmap in your Muster dashboard. Within minutes of your first job completing you’ll see pass rates, trends, and risk signals.
The quality endpoint is unauthenticated by design — no tokens to manage, no secrets to rotate. Rate limit: 1,000 requests/minute per agent.
Optional: use the SDK
If you prefer a pip package that handles fire-and-forget, retries, and token tracking automatically:
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),
])
Next steps
n8n integration
Add one HTTP Request node to any n8n workflow.
All integrations
LangChain, LangGraph, Flowise, Dify and more.
