Docs Call the REST API from code
Call the REST API from code
Ready-to-adapt clients for the REST API in curl, Python, Node/TypeScript, Rust and Go.
On this page
Client examples
End-to-end examples in five languages for the REST API. Each full client covers:
- bearer-token auth,
- listing dialogs filtered by state, a page at a time,
- fetching a single dialog,
- and error handling.
The Python one also scrapes /metrics. Adapt to your environment.
The Go, TypeScript and Rust clients are complete programs in the repository, under clients/. The code on this page is the code of those programs.
- Each reads the base URL from
SIPNAB_URL(defaulthttp://localhost:8080). - Each reads the bearer token from
SIPNAB_API_KEY.
Filter parameters:
/v1/dialogsaccepts:
state(e.g.Failed,Completed,InCall) andfrom(a regular expression on the From user),filter, a filter expression such asmethod == 'INVITE', the same languagesipnab --filtertakes, or a named alias such asproblems,afterandbefore, RFC 3339 instants that bound when the dialog opened.
/v1/streamsacceptsorphanedandmos_below. The REST API reference lists every parameter of every route.
Status codes: the REST API returns 503 Service Unavailable, not 429, in two cases:
- the per-peer rate limiter turns a request away,
- or the in-flight request cap is full.
A missing or wrong token gets 401. An unknown
call_idgets 404.
Per-message data is not on REST.
- The REST API returns each dialog as a summary. A list row carries
call_id,state,method,from_user,to_user,final_status_code,duration_sec,msg_countandtiming./v1/aggregate?by=response_codecounts the dialogs by final response code.- Individual SIP messages, and responses other than the final one, are not exposed by
/v1/dialogsor/v1/dialogs/{id}.To work with per-message data programmatically, use either:
- the CLI
sipnab -N -I capture.pcap --jsonmode.
- It emits one JSON object per SIP message with
is_request,status_code,reason, etc.- Field reference: Output Formats. See also cookbook Recipe “Find every failed call, grouped by response code”.
- or the MCP
get_dialogtool. It returns paginatedmessages[](see MCP).
curl + jq one-liners
Set the server address and token once per shell.
- Every command below reads
$APIand$KEY. - So a command pasted into a shell without them requests
/v1/dialogson no host with no bearer token. - Every command here is a read. It changes nothing on the server.
# Run all of these, in order.
API="http://localhost:8080"
KEY="my-secret-token"
Then run whichever of these you need:
-
Check the server is up (no token needed):
curl -fsS "$API/health" -
List failed dialogs (the
state=query parameter):curl -fsS "$API/v1/dialogs?state=Failed&limit=20" -H "Authorization: Bearer $KEY" | jq -
List dialogs from one user (
from=takes a regex):curl -fsS "$API/v1/dialogs?from=alice&limit=20" -H "Authorization: Bearer $KEY" | jq -
Get one dialog, aggregated. REST carries no per-message data:
curl -fsS "$API/v1/dialogs/abc123@host" -H "Authorization: Bearer $KEY" | jq -
Get a call report. This endpoint returns JSON only:
curl -fsS "$API/v1/dialogs/abc123@host/report" -H "Authorization: Bearer $KEY" | jq -
List streams that are not orphaned (
orphaned=false):curl -fsS "$API/v1/streams?orphaned=false" -H "Authorization: Bearer $KEY" | jq -
List streams below a MOS threshold:
curl -fsS "$API/v1/streams?mos_below=3.5" -H "Authorization: Bearer $KEY" | jq -
Read the aggregate counters:
curl -fsS "$API/v1/stats" -H "Authorization: Bearer $KEY" | jq -
Count failed calls. REST exposes no per-message data, so this counts dialogs:
curl -fsS "$API/v1/dialogs?state=Failed&limit=1000" -H "Authorization: Bearer $KEY" | jq '.total' -
Read the Prometheus metrics:
curl -fsS "$API/metrics" -H "Authorization: Bearer $KEY" | grep '^sipnab_' -
Count the dialogs by final response code:
curl -fsS "$API/v1/aggregate?by=response_code" -H "Authorization: Bearer $KEY" | jq -r '.buckets[] | "\(.count) \(.value)"'
To count every response message rather than one final code per call, use the
CLI NDJSON mode. sipnab -N --json emits one record per message (see
Output Formats):
sipnab -N -I capture.pcap --filter "state == 'Failed'" --json \
| jq -r 'select(.is_request == false) | .status_code' \
| sort | uniq -c | sort -rn
To handle errors in a script, branch on the status code. The server returns 503, not 429, when a rate limit or the connection cap turns the request away:
# Run all of these, in order.
http_code=$(curl -s -o /dev/null -w '%{http_code}' \
"$API/v1/dialogs/no-such-call" -H "Authorization: Bearer $KEY")
case "$http_code" in
200) echo "found" ;;
401) echo "auth failed — check --api-key" ;;
404) echo "dialog not found" ;;
503) echo "rate-limited or connection cap reached" ;;
*) echo "unexpected $http_code" ;;
esac
The per-message NDJSON records referenced above (
is_request,status_code,reason,cseq, and so on) appear in Output Formats.
Python (sync, requests)
The Python examples on this page use requests and httpx:
pip install requests httpx
clients/python/requirements-examples.txt
pins the versions the repository tests the examples with.
"""sipnab REST client — sync version using requests."""
from __future__ import annotations
import os
import sys
from typing import Any
import requests
API = os.environ.get("SIPNAB_URL", "http://localhost:8080")
KEY = os.environ["SIPNAB_API_KEY"] # raises KeyError if unset
class SipnabError(Exception):
pass
class SipnabClient:
def __init__(self, base_url: str = API, token: str = KEY,
timeout: float = 10.0) -> None:
self.base = base_url.rstrip("/")
self.session = requests.Session()
self.session.headers["Authorization"] = f"Bearer {token}"
self.timeout = timeout
def _get(self, path: str, **params: Any) -> Any:
r = self.session.get(f"{self.base}{path}", params=params,
timeout=self.timeout)
if r.status_code == 401:
raise SipnabError("authentication failed (401)")
if r.status_code == 503:
raise SipnabError("rate-limited or connection cap reached (503)")
r.raise_for_status()
return r.json()
def health(self) -> bool:
r = self.session.get(f"{self.base}/health", timeout=self.timeout)
return r.ok
def list_dialogs(self, *, state: str | None = None,
from_regex: str | None = None,
limit: int = 50, offset: int = 0) -> list[dict]:
"""List dialog summaries.
The REST API supports filtering by `state` (exact match against
DialogState e.g. 'Failed', 'Completed', 'InCall') and `from` (regex).
For full DSL filtering use the MCP server's list_dialogs tool.
"""
params: dict[str, Any] = {"limit": limit, "offset": offset}
if state:
params["state"] = state
if from_regex:
params["from"] = from_regex
return self._get("/v1/dialogs", **params)["dialogs"]
def get_dialog(self, call_id: str) -> dict:
from urllib.parse import quote
return self._get(f"/v1/dialogs/{quote(call_id, safe='')}")
def call_report(self, call_id: str) -> dict:
from urllib.parse import quote
return self._get(f"/v1/dialogs/{quote(call_id, safe='')}/report")
def stats(self) -> dict:
return self._get("/v1/stats")
def metrics(self) -> str:
r = self.session.get(f"{self.base}/metrics", timeout=self.timeout)
if r.status_code == 401:
raise SipnabError("authentication failed")
if r.status_code == 503:
raise SipnabError("rate-limited")
r.raise_for_status()
return r.text
# ── Usage ─────────────────────────────────────────────────────────
if __name__ == "__main__":
c = SipnabClient()
if not c.health():
sys.exit("sipnab not reachable")
print("Stats:", c.stats())
# Pull every failed call, page through
failed: list[dict] = []
offset = 0
while True:
page = c.list_dialogs(state="Failed", limit=100, offset=offset)
if not page:
break
failed.extend(page)
offset += len(page)
print(f"{len(failed)} failed dialogs")
# Show the first few — note the REST shape doesn't expose
# per-message status_code. See module-level note above for how to
# build a response-code histogram via CLI or MCP.
for d in failed[:5]:
full = c.get_dialog(d["call_id"])
diag = full.get("diagnosis", {})
print(f" {d['call_id']:30s} state={d['state']:10s} "
f"diagnosis={ {k: v for k, v in diag.items() if v} }")
Run it:
SIPNAB_API_KEY=my-secret-token python3 sipnab_client.py
Python (async, httpx)
For tailing dialogs in near-real-time without blocking:
"""sipnab REST client — async, periodic polling."""
import asyncio
import os
from datetime import datetime, timezone
import httpx
API = os.environ.get("SIPNAB_URL", "http://localhost:8080")
KEY = os.environ["SIPNAB_API_KEY"]
async def tail_dialogs(poll_interval: float = 2.0) -> None:
"""Poll /v1/dialogs every `poll_interval` and print newly-completed calls."""
seen: set[str] = set()
headers = {"Authorization": f"Bearer {KEY}"}
async with httpx.AsyncClient(base_url=API, headers=headers,
timeout=10.0) as client:
while True:
try:
r = await client.get("/v1/dialogs",
params={"limit": 100})
r.raise_for_status()
for d in r.json()["dialogs"]:
if d["call_id"] in seen:
continue
seen.add(d["call_id"])
if d["state"] in ("Completed", "Failed", "Canceled"):
print(f"{datetime.now(timezone.utc).isoformat()} "
f"{d['state']:10s} {d['call_id']} "
f"{d.get('from_user')} → {d.get('to_user')}")
except httpx.HTTPError as e:
print(f"warning: {e}")
await asyncio.sleep(poll_interval)
if __name__ == "__main__":
asyncio.run(tail_dialogs())
Node.js / TypeScript
// sipnab-client.ts: runs on Node 22.18+ (built-in fetch, type stripping)
const API = process.env.SIPNAB_URL ?? "http://localhost:8080";
const KEY = process.env.SIPNAB_API_KEY;
if (!KEY) throw new Error("SIPNAB_API_KEY not set");
interface DialogSummary {
call_id: string;
state: string;
method: string;
from_user: string | null;
to_user: string | null;
duration_sec: number;
msg_count: number;
}
interface DialogsPage {
dialogs: DialogSummary[];
total: number;
limit: number;
offset: number;
}
async function api<T>(
path: string,
params: Record<string, string | number> = {},
): Promise<T> {
const url = new URL(`${API}${path}`);
for (const [k, v] of Object.entries(params)) {
url.searchParams.set(k, String(v));
}
const r = await fetch(url, {
headers: { Authorization: `Bearer ${KEY}` },
});
if (r.status === 401) throw new Error(`GET ${path}: 401 auth failed`);
if (r.status === 503) throw new Error(`GET ${path}: 503 rate-limited or conn cap reached`);
if (!r.ok) throw new Error(`GET ${path}: HTTP ${r.status}`);
return (await r.json()) as T;
}
async function listDialogs(
state: string | null = null,
limit = 50,
): Promise<DialogSummary[]> {
const all: DialogSummary[] = [];
let offset = 0;
for (;;) {
const params: Record<string, string | number> = { limit, offset };
if (state) params.state = state;
const page = await api<DialogsPage>("/v1/dialogs", params);
if (page.dialogs.length === 0) break;
all.push(...page.dialogs);
if (all.length >= page.total) break;
offset += page.dialogs.length;
}
return all;
}
// REST API doesn't expose per-message data — see the note at the top of
// "Client Examples" for how to build per-call response-code histograms
// via the CLI or MCP. Here we just summarize what REST exposes. A timing
// sipnab did not measure is absent from one dialog, not null:
interface FullDialog {
call_id: string;
state: string;
msg_count: number;
timing: { pdd_ms?: number; setup_ms?: number; retransmits: number };
diagnosis: { one_way_audio: boolean; nat_mismatch: boolean; no_media: boolean };
}
// ── Demo ──────────────────────────────────────────────────────────
const failed = await listDialogs("Failed");
console.log(`${failed.length} failed dialogs`);
for (const d of failed.slice(0, 5)) {
const full = await api<FullDialog>(`/v1/dialogs/${encodeURIComponent(d.call_id)}`);
const pdd = full.timing.pdd_ms === undefined ? "—" : `${full.timing.pdd_ms}ms`;
console.log(` ${d.call_id} state=${d.state} ` +
`pdd=${pdd} ` +
`nat_mismatch=${full.diagnosis.nat_mismatch}`);
}
The program is clients/typescript/sipnab-client.ts.
Node.js 22.18 or later runs it directly. The package’s package.json must say "type": "module":
SIPNAB_API_KEY=my-secret-token node sipnab-client.ts
Rust (reqwest)
// Cargo.toml [dependencies]:
// anyhow = "1"
// reqwest = { version = "0.13", default-features = false, features = ["blocking", "json"] }
// serde = { version = "1", features = ["derive"] }
//! Lists every failed dialog, a page at a time, and prints the post-dial
//! delay and NAT diagnosis of the first five.
//!
//! SIPNAB_URL sets the API base URL (default http://localhost:8080) and
//! SIPNAB_API_KEY the bearer token, which is required.
use anyhow::{Result, anyhow, bail};
use reqwest::Url;
use reqwest::blocking::Client;
use serde::Deserialize;
use serde::de::DeserializeOwned;
use std::env;
use std::time::Duration;
#[derive(Debug, Deserialize)]
struct DialogSummary {
call_id: String,
state: String,
}
#[derive(Debug, Deserialize)]
struct DialogsPage {
dialogs: Vec<DialogSummary>,
total: usize,
}
#[derive(Debug, Deserialize)]
struct Timing {
pdd_ms: Option<i64>,
}
#[derive(Debug, Deserialize)]
struct Diagnosis {
nat_mismatch: bool,
}
/// One dialog, aggregated. REST does not expose individual messages; for
/// those, use the CLI `sipnab -N --json` mode or the MCP `get_dialog` tool.
#[derive(Debug, Deserialize)]
struct FullDialog {
timing: Timing,
diagnosis: Diagnosis,
}
struct Sipnab {
base: Url,
client: Client,
}
impl Sipnab {
fn new() -> Result<Self> {
let base = env::var("SIPNAB_URL").unwrap_or_else(|_| "http://localhost:8080".into());
let key = env::var("SIPNAB_API_KEY").map_err(|_| anyhow!("SIPNAB_API_KEY not set"))?;
let mut headers = reqwest::header::HeaderMap::new();
headers.insert(
reqwest::header::AUTHORIZATION,
format!("Bearer {key}").parse()?,
);
let client = Client::builder()
.default_headers(headers)
.timeout(Duration::from_secs(10))
.build()?;
Ok(Self {
base: Url::parse(&base)?,
client,
})
}
/// GET `segments` under the base URL, each one percent-encoded, and
/// decode the JSON body. Every status but 2xx is an error.
fn get<T: DeserializeOwned>(&self, segments: &[&str], query: &[(&str, &str)]) -> Result<T> {
let mut url = self.base.clone();
url.path_segments_mut()
.map_err(|()| anyhow!("{} cannot be a base URL", self.base))?
.pop_if_empty()
.extend(segments);
if !query.is_empty() {
url.query_pairs_mut().extend_pairs(query);
}
let path = url.path().to_string();
let resp = self.client.get(url).send()?;
match resp.status().as_u16() {
401 => bail!("GET {path}: 401 auth failed"),
503 => bail!("GET {path}: 503 rate-limited or conn cap reached"),
code if code >= 400 => bail!("GET {path}: HTTP {code}"),
_ => {}
}
Ok(resp.json()?)
}
fn list_dialogs(&self, state: Option<&str>) -> Result<Vec<DialogSummary>> {
let mut all = Vec::new();
loop {
let offset = all.len().to_string();
let mut query = vec![("limit", "100"), ("offset", offset.as_str())];
if let Some(s) = state {
query.push(("state", s));
}
let page: DialogsPage = self.get(&["v1", "dialogs"], &query)?;
if page.dialogs.is_empty() {
break;
}
all.extend(page.dialogs);
if all.len() >= page.total {
break;
}
}
Ok(all)
}
fn get_dialog(&self, call_id: &str) -> Result<FullDialog> {
self.get(&["v1", "dialogs", call_id], &[])
}
}
fn main() -> Result<()> {
let s = Sipnab::new()?;
let failed = s.list_dialogs(Some("Failed"))?;
println!("{} failed dialogs", failed.len());
for d in failed.iter().take(5) {
let full = s.get_dialog(&d.call_id)?;
let pdd = full
.timing
.pdd_ms
.map_or_else(|| "—".to_string(), |ms| format!("{ms}ms"));
println!(
" {} state={} pdd={} nat_mismatch={}",
d.call_id, d.state, pdd, full.diagnosis.nat_mismatch
);
}
Ok(())
}
The program is clients/rust/src/main.rs.
To run it in a project of your own, put the three dependency lines under [dependencies] in its Cargo.toml:
SIPNAB_API_KEY=my-secret-token cargo run
The per-message
sipnab -N --jsonrecords mentioned inFullDialog’s doc comment appear in Output Formats.
Go (net/http + encoding/json)
// Command sipnab-client lists every failed dialog, a page at a time, and
// prints the post-dial delay and NAT diagnosis of the first five.
//
// SIPNAB_URL sets the API base URL (default http://localhost:8080) and
// SIPNAB_API_KEY the bearer token, which is required.
package main
import (
"encoding/json"
"fmt"
"net/http"
"net/url"
"os"
"time"
)
type DialogSummary struct {
CallID string `json:"call_id"`
State string `json:"state"`
FromUser *string `json:"from_user"`
ToUser *string `json:"to_user"`
DurationSec float64 `json:"duration_sec"`
MsgCount int `json:"msg_count"`
}
type DialogTiming struct {
PddMs *int64 `json:"pdd_ms"`
SetupMs *int64 `json:"setup_ms"`
Retransmits int `json:"retransmits"`
}
type DialogDiagnosis struct {
OneWayAudio bool `json:"one_way_audio"`
NatMismatch bool `json:"nat_mismatch"`
NoMedia bool `json:"no_media"`
}
type FullDialog struct {
CallID string `json:"call_id"`
State string `json:"state"`
Timing DialogTiming `json:"timing"`
Diagnosis DialogDiagnosis `json:"diagnosis"`
}
type DialogsPage struct {
Dialogs []DialogSummary `json:"dialogs"`
Total int `json:"total"`
Limit int `json:"limit"`
Offset int `json:"offset"`
}
type Sipnab struct {
Base string
Token string
Client *http.Client
}
func newSipnab() (*Sipnab, error) {
base := os.Getenv("SIPNAB_URL")
if base == "" {
base = "http://localhost:8080"
}
token := os.Getenv("SIPNAB_API_KEY")
if token == "" {
return nil, fmt.Errorf("SIPNAB_API_KEY not set")
}
return &Sipnab{
Base: base,
Token: token,
Client: &http.Client{Timeout: 10 * time.Second},
}, nil
}
func (s *Sipnab) get(path string, params url.Values, out any) error {
u, err := url.Parse(s.Base + path)
if err != nil {
return err
}
u.RawQuery = params.Encode()
req, err := http.NewRequest(http.MethodGet, u.String(), nil)
if err != nil {
return err
}
req.Header.Set("Authorization", "Bearer "+s.Token)
resp, err := s.Client.Do(req)
if err != nil {
return err
}
defer resp.Body.Close()
switch resp.StatusCode {
case http.StatusUnauthorized:
return fmt.Errorf("GET %s: 401 auth failed", path)
case http.StatusServiceUnavailable:
return fmt.Errorf("GET %s: 503 rate-limited or conn cap reached", path)
}
if resp.StatusCode >= 400 {
return fmt.Errorf("GET %s: HTTP %d", path, resp.StatusCode)
}
return json.NewDecoder(resp.Body).Decode(out)
}
func (s *Sipnab) ListDialogs(state string) ([]DialogSummary, error) {
var all []DialogSummary
offset := 0
for {
params := url.Values{"limit": {"100"}, "offset": {fmt.Sprint(offset)}}
if state != "" {
params.Set("state", state)
}
var page DialogsPage
if err := s.get("/v1/dialogs", params, &page); err != nil {
return nil, err
}
if len(page.Dialogs) == 0 {
break
}
all = append(all, page.Dialogs...)
offset += len(page.Dialogs)
if len(all) >= page.Total {
break
}
}
return all, nil
}
// GetDialog fetches the full (aggregated) dialog. REST has no
// per-message detail — for that, use the CLI --json mode or the
// MCP get_dialog tool.
func (s *Sipnab) GetDialog(callID string) (*FullDialog, error) {
var full FullDialog
if err := s.get("/v1/dialogs/"+url.PathEscape(callID), nil, &full); err != nil {
return nil, err
}
return &full, nil
}
func main() {
if err := run(); err != nil {
fmt.Fprintln(os.Stderr, "sipnab-client:", err)
os.Exit(1)
}
}
func run() error {
s, err := newSipnab()
if err != nil {
return err
}
failed, err := s.ListDialogs("Failed")
if err != nil {
return err
}
fmt.Printf("%d failed dialogs\n", len(failed))
for i, d := range failed {
if i >= 5 {
break
}
full, err := s.GetDialog(d.CallID)
if err != nil {
return err
}
pdd := "—"
if full.Timing.PddMs != nil {
pdd = fmt.Sprintf("%dms", *full.Timing.PddMs)
}
fmt.Printf(" %s state=%s pdd=%s nat_mismatch=%t\n",
d.CallID, d.State, pdd, full.Diagnosis.NatMismatch)
}
return nil
}
The program is clients/go/sipnab-client/main.go. Save it as sipnab-client.go and run it:
SIPNAB_API_KEY=my-secret-token go run sipnab-client.go
Common patterns
Monitor failed calls in real time (Python)
import os
import time
import requests
API = os.environ.get("SIPNAB_URL", "http://127.0.0.1:8080")
KEY = os.environ["SIPNAB_API_KEY"]
HEADERS = {"Authorization": f"Bearer {KEY}"}
seen = set()
while True:
resp = requests.get(f"{API}/v1/dialogs", headers=HEADERS,
params={"state": "Failed"})
resp.raise_for_status()
for d in resp.json()["dialogs"]:
cid = d["call_id"]
if cid not in seen:
seen.add(cid)
print(f"FAILED: {cid} from={d.get('from_user')} to={d.get('to_user')}")
time.sleep(5)
Export all dialogs to CSV (bash)
curl -s -H "Authorization: Bearer $SIPNAB_API_KEY" \
"http://127.0.0.1:8080/v1/dialogs?limit=1000" | \
jq -r '.dialogs[] | [.call_id, .method, .state, .from_user, .to_user, .duration_sec] | @csv'
Alert on poor MOS (bash)
curl -s -H "Authorization: Bearer $SIPNAB_API_KEY" \
"http://127.0.0.1:8080/v1/streams?mos_below=3.0" | \
jq -r '.streams[] | "LOW MOS: SSRC=\(.ssrc) MOS=\(.mos) call=\(.associated_dialog)"'
Grafana dashboard via Prometheus
# prometheus.yml
scrape_configs:
- job_name: sipnab
bearer_token: your-api-key
static_configs:
- targets: ['127.0.0.1:8080']
scrape_interval: 15s
Tip: The metrics endpoint is lightweight. It is suitable for 5-15 second scrape intervals. The repo ships a sample Grafana dashboard JSON. The repository includes it at
contrib/grafana/sipnab-dashboard.json.
Paginate through all dialogs (Python)
import os
import requests
API = os.environ.get("SIPNAB_URL", "http://127.0.0.1:8080")
HEADERS = {"Authorization": f"Bearer {os.environ['SIPNAB_API_KEY']}"}
offset = 0
limit = 100
all_dialogs = []
while True:
resp = requests.get(f"{API}/v1/dialogs",
headers=HEADERS,
params={"limit": limit, "offset": offset})
resp.raise_for_status()
data = resp.json()
all_dialogs.extend(data["dialogs"])
if offset + limit >= data["total"]:
break
offset += limit
print(f"Fetched {len(all_dialogs)} dialogs")