Skip to main content

Documentation Index

Fetch the complete documentation index at: https://docs.getmuster.io/llms.txt

Use this file to discover all available pages before exploring further.

​
When do you need the SDK?

OTel and platform connectors already produce quality signals for every agent — Muster’s check inference engine infers checks automatically from agent name, output structure, and historical patterns. For most agents, that’s enough. The SDK is for agents where inferred checks aren’t precise enough:
SituationUse
Need subtotal + tax == total with actual valuesSDK
Need grounding checks (extracted value appears in source doc)SDK
Need entity matching against known recordsSDK
General monitoring, cost tracking, trend detectionOTel / connectors
Rule of thumb: Start with OTel or a connector. Add the SDK to your Finance, Compliance, Legal, and Risk agents. 
pip install muster-sdk

from muster_sdk import MusterAgent

agent = MusterAgent(
    agent_id   = "invoice-processor-v2",
    muster_url = "https://backend.YOUR-INSTANCE.getmuster.io",
    token      = "your-agent-token",
)

@agent.govern(
    required_fields=["vendor_name", "invoice_number", "total"],
    arithmetic=[{
        "check_id":         "total_check",
        "result_field":     "total",
        "component_fields": ["subtotal", "tax_amount"],
        "operation":        "sum",
    }],
)
async def process_invoice(input_data: dict) -> dict:
    return extracted_output  # your existing code, unchanged
Checks run after every job. Full SDK guide →

​
Option B — Plain HTTP (no library)

For agents where you want to send specific checks but don’t want a library dependency. No authentication required on the quality endpoint. 
import httpx, threading

def muster_emit(job_id, checks, token_input=None, token_output=None, model=None, latency_ms=None):
    """Fire-and-forget — never blocks your agent."""
    def _send():
        try:
            httpx.post(
                f"https://backend.YOUR-INSTANCE.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,
                    "token_input": token_input,
                    "token_output": token_output,
                    "model": model,
                    "latency_ms": latency_ms,
                },
                timeout=2.0,
            )
        except Exception:
            pass  # never fail your agent
    threading.Thread(target=_send, daemon=True).start()

muster_emit(
    job_id=job_id,
    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,
)

​
Option C — n8n / no-code webhook

Add an HTTP Request node at the end of your workflow. Muster also receives signals from the n8n connector automatically — use this only if you need specific business-rule checks. 
POST https://backend.YOUR-INSTANCE.getmuster.io/api/v1/jobs/{{$runIndex}}/quality
Body: {"agent_id": "my-workflow", "job_id": "{{$runIndex}}", "overall_passed": true,
       "checks": [{"check_id": "output_not_empty", "severity": "HIGH", "passed": {{bool(output)}}}]}
→ Full n8n guide

​
Relationship to OTel and connectors

The SDK does not replace OTel or platform connectors — it complements them:
  • OTel / connectors → discovery, cost tracking, inferred quality signals, execution visibility
  • SDK → exact business-rule checks, grounding, entity matching
An agent can have both OTel (for traces + token usage) and SDK (for precise checks) active at the same time. The signals are merged in the Health Heatmap.

​
Security

  • No inbound access required — agents make outbound HTTPS calls only
  • No secrets needed — the quality endpoint is intentionally unauthenticated
  • Non-blocking — 2-second timeout, fail-open, runs in background thread
  • Air-gapped option — deploy Muster inside your own network for zero external calls