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 (default http://localhost:8080).
  • Each reads the bearer token from SIPNAB_API_KEY.

Filter parameters: /v1/dialogs accepts:

  • state (e.g. Failed, Completed, InCall) and from (a regular expression on the From user),
  • filter, a filter expression such as method == 'INVITE', the same language sipnab --filter takes, or a named alias such as problems,
  • after and before, RFC 3339 instants that bound when the dialog opened.

/v1/streams accepts orphaned and mos_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_id gets 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_count and timing.
  • /v1/aggregate?by=response_code counts the dialogs by final response code.
  • Individual SIP messages, and responses other than the final one, are not exposed by /v1/dialogs or /v1/dialogs/{id}.

To work with per-message data programmatically, use either:

curl + jq one-liners

Set the server address and token once per shell.

  • Every command below reads $API and $KEY.
  • So a command pasted into a shell without them requests /v1/dialogs on 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 --json records mentioned in FullDialog’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")