> ## 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.

# Python snippet

> Add quality signals to any Python agent in 5 lines. No pip install required.

## The snippet

Copy this into your agent. It runs in a background thread and never blocks your agent.

```python theme={null}
import httpx, threading, time

def muster_emit(job_id, checks, token_input=None, token_output=None, model=None, latency_ms=None):
    """Fire-and-forget. Covers: correctness checks + cost + latency in one call."""
    def _send():
        try:
            httpx.post(
                f"https://backend.getmuster.io/api/v1/jobs/{job_id}/quality",
                json={
                    "agent_id":      "your-agent-name",   # ← replace with your agent name
                    "job_id":        job_id,
                    "overall_passed": all(c["passed"] for c in checks),
                    "checks":        checks,
                    "token_input":   token_input,          # from LLM response.usage
                    "token_output":  token_output,
                    "model":         model,
                    "latency_ms":    latency_ms,
                },
                timeout=2.0,
            )
        except Exception:
            pass  # never block your agent
    threading.Thread(target=_send, daemon=True).start()
```

## Usage

```python theme={null}
start = time.time()
result = your_agent.run(input)   # ← your existing code, unchanged

muster_emit(
    job_id=job_id,
    checks=[
        {"check_id": "output_not_empty",    "severity": "HIGH",   "passed": bool(result)},
        {"check_id": "subtotal_arithmetic", "severity": "HIGH",
         "passed": abs(computed - declared) < 0.01,
         "expected": str(declared), "actual": str(computed)},
    ],
    token_input=result.usage.prompt_tokens,    # OpenAI / Anthropic response
    token_output=result.usage.completion_tokens,
    model="gpt-4o",
    latency_ms=int((time.time() - start) * 1000),
)
```

## What each field powers

| Field                                    | Powers in dashboard                          |
| ---------------------------------------- | -------------------------------------------- |
| `checks`                                 | Health Heatmap pass rates, anomaly detection |
| `token_input` + `token_output` + `model` | Cost Dashboard — \$ calculated automatically |
| `latency_ms`                             | SLA monitoring, latency trends               |
| `overall_passed: false`                  | Anomaly detection — failure rate spike       |

## Recommended check IDs

| check\_id                 | What to check                          |
| ------------------------- | -------------------------------------- |
| `output_not_empty`        | Agent produced a non-empty response    |
| `subtotal_arithmetic`     | Numeric totals add up correctly        |
| `required_fields_present` | All required output fields are present |
| `no_refusal_in_output`    | Agent didn't say it can't help         |
| `decision_is_valid_enum`  | Decision is one of the expected values |
| `source_cited`            | Claims include source references       |
| `grand_total_arithmetic`  | Subtotal + tax = grand total           |
| `latency_within_sla`      | Response time under threshold          |

## Getting token counts

```python theme={null}
# OpenAI
token_input  = response.usage.prompt_tokens
token_output = response.usage.completion_tokens

# Anthropic
token_input  = response.usage.input_tokens
token_output = response.usage.output_tokens

# LangChain
from langchain.callbacks import get_openai_callback
with get_openai_callback() as cb:
    result = chain.invoke({"input": "..."})
token_input  = cb.prompt_tokens
token_output = cb.completion_tokens
```