API & MCP

Code examples

Call the Sonar MCP server from curl, JavaScript and Python, with or without an MCP library.

Every example reads your API key from the SONAR_API_KEY environment variable, so the key stays out of your code. Create a key under Settings → API keys, then:

export SONAR_API_KEY=sonar_YOUR_KEY

The site IDs in the examples (49 and 50) are Sonar's demo sites. Use list_sites to find yours.

curl

List the tools:

curl -s https://sonar-app.co.uk/api/mcp \
  -H "Authorization: Bearer $SONAR_API_KEY" \
  -H "Content-Type: application/json" \
  -H "Accept: application/json, text/event-stream" \
  -d '{"jsonrpc": "2.0", "id": 1, "method": "tools/list"}'

Call one. Add | jq -r '.result.content[0].text | fromjson' to see just the data:

curl -s https://sonar-app.co.uk/api/mcp \
  -H "Authorization: Bearer $SONAR_API_KEY" \
  -H "Content-Type: application/json" \
  -H "Accept: application/json, text/event-stream" \
  -d '{
    "jsonrpc": "2.0",
    "id": 1,
    "method": "tools/call",
    "params": { "name": "get_overview", "arguments": { "site_id": 50, "period": "30d" } }
  }'

JavaScript (no libraries)

Plain fetch, in Node 18 or later. callTool does the request and returns the tool's data, or throws with Sonar's message. This prints last week's visitors for each of your sites:

// Node 18+ (built-in fetch). Run with: SONAR_API_KEY=sonar_... node sonar.mjs
const SONAR_URL = 'https://sonar-app.co.uk/api/mcp'

async function callTool(name, args = {}) {
  const res = await fetch(SONAR_URL, {
    method: 'POST',
    headers: {
      Authorization: `Bearer ${process.env.SONAR_API_KEY}`,
      'Content-Type': 'application/json',
      Accept: 'application/json, text/event-stream',
    },
    body: JSON.stringify({ jsonrpc: '2.0', id: 1, method: 'tools/call', params: { name, arguments: args } }),
  })
  const body = await res.json()
  if (body.error) throw new Error(body.error.message) // a bad key or plan
  const text = body.result.content[0].text
  if (body.result.isError) throw new Error(text) // e.g. a site you can't see
  return JSON.parse(text)
}

const sites = await callTool('list_sites')
for (const site of sites) {
  const { current, previous_period } = await callTool('get_overview', { site_id: site.id, period: '7d' })
  console.log(`${site.name}: ${current.visitors} visitors this week (${previous_period.visitors} the week before)`)
}

Paging through visits

list_sessions returns 20 visits at a time. Pass next_before back as before until it's null:

// Every visit in the last 7 days, 20 at a time, and the share that came from mobiles.
const SONAR_URL = 'https://sonar-app.co.uk/api/mcp'

async function callTool(name, args = {}) {
  const res = await fetch(SONAR_URL, {
    method: 'POST',
    headers: {
      Authorization: `Bearer ${process.env.SONAR_API_KEY}`,
      'Content-Type': 'application/json',
      Accept: 'application/json, text/event-stream',
    },
    body: JSON.stringify({ jsonrpc: '2.0', id: 1, method: 'tools/call', params: { name, arguments: args } }),
  })
  const body = await res.json()
  if (body.error) throw new Error(body.error.message) // a bad key or plan
  const text = body.result.content[0].text
  if (body.result.isError) throw new Error(text) // e.g. a site you can't see
  return JSON.parse(text)
}

const visits = []
let before
do {
  const page = await callTool('list_sessions', { site_id: 49, period: '7d', ...(before && { before }) })
  visits.push(...page.sessions)
  before = page.next_before
} while (before)

const mobile = visits.filter(v => v.device_type === 'mobile').length
console.log(`${visits.length} visits, ${Math.round((mobile / visits.length) * 100)}% on mobile`)

Python (no libraries beyond requests)

# pip install requests. Run with: SONAR_API_KEY=sonar_... python sonar_requests.py
import json
import os

import requests

SONAR_URL = "https://sonar-app.co.uk/api/mcp"


def call_tool(name, arguments=None):
    response = requests.post(
        SONAR_URL,
        headers={
            "Authorization": f"Bearer {os.environ['SONAR_API_KEY']}",
            "Accept": "application/json, text/event-stream",
        },
        json={"jsonrpc": "2.0", "id": 1, "method": "tools/call",
              "params": {"name": name, "arguments": arguments or {}}},
        timeout=60,
    )
    body = response.json()
    if "error" in body:  # a bad key or plan
        raise RuntimeError(body["error"]["message"])
    text = body["result"]["content"][0]["text"]
    if body["result"].get("isError"):  # e.g. a site you can't see
        raise RuntimeError(text)
    return json.loads(text)


for page in call_tool("top_pages", {"site_id": 50, "period": "30d", "limit": 5}):
    print(f"{page['visitors']:>5}  {page['path']}")

With the official MCP SDKs

The MCP libraries handle the protocol for you. They're the better choice when you're building an agent or an app that works with more than one MCP server.

TypeScript (@modelcontextprotocol/sdk version 1):

// npm install @modelcontextprotocol/sdk. Run with: SONAR_API_KEY=sonar_... node sonar-sdk.mjs
import { Client } from '@modelcontextprotocol/sdk/client/index.js'
import { StreamableHTTPClientTransport } from '@modelcontextprotocol/sdk/client/streamableHttp.js'

const transport = new StreamableHTTPClientTransport(new URL('https://sonar-app.co.uk/api/mcp'), {
  requestInit: { headers: { Authorization: `Bearer ${process.env.SONAR_API_KEY}` } },
})
const sonar = new Client({ name: 'my-report', version: '1.0.0' })
await sonar.connect(transport)

const result = await sonar.callTool({ name: 'interactions', arguments: { site_id: 49, period: '30d' } })
const report = JSON.parse(result.content[0].text)
console.log(`${report.totals.interactions} enquiries from ${report.totals.sessions} visits`)
for (const row of report.by_source.filter(r => r.interactions > 0)) {
  console.log(`${row.label}: ${row.interactions} (${(row.rate * 100).toFixed(1)}% of visits)`)
}

await sonar.close()

Python (mcp version 2 or later; version 1 connects differently):

# pip install "mcp>=2". Run with: SONAR_API_KEY=sonar_... python sonar_mcp.py
import asyncio
import json
import os

import httpx2
from mcp import Client
from mcp.client.streamable_http import streamable_http_client

SONAR_URL = "https://sonar-app.co.uk/api/mcp"


async def main():
    headers = {"Authorization": f"Bearer {os.environ['SONAR_API_KEY']}"}
    async with httpx2.AsyncClient(headers=headers) as http:
        async with Client(streamable_http_client(SONAR_URL, http_client=http)) as sonar:
            tools = await sonar.list_tools()
            print("Tools:", ", ".join(tool.name for tool in tools.tools))

            result = await sonar.call_tool("ask_sonar", {
                "site_id": 50,
                "question": "Which day of the week gets the most visitors?",
                "period": "30d",
            })
            answer = json.loads(result.content[0].text)
            for row in answer["rows"]:
                print(row)


asyncio.run(main())

Recipe: a weekly summary in Slack

Visitors, change from the week before, conversions and the top page for every site you can see, printed and posted to Slack. Run it every Monday with cron or a scheduled job. Create a Slack incoming webhook and set its address as SLACK_WEBHOOK_URL to post it.

# Weekly summary for every site, printed and (optionally) posted to Slack.
# pip install requests. Run with: SONAR_API_KEY=sonar_... python weekly_summary.py
# Set SLACK_WEBHOOK_URL to a Slack incoming webhook to post it there too.
import json
import os

import requests

SONAR_URL = "https://sonar-app.co.uk/api/mcp"


def call_tool(name, arguments=None):
    response = requests.post(
        SONAR_URL,
        headers={
            "Authorization": f"Bearer {os.environ['SONAR_API_KEY']}",
            "Accept": "application/json, text/event-stream",
        },
        json={"jsonrpc": "2.0", "id": 1, "method": "tools/call",
              "params": {"name": name, "arguments": arguments or {}}},
        timeout=60,
    )
    body = response.json()
    if "error" in body:
        raise RuntimeError(body["error"]["message"])
    text = body["result"]["content"][0]["text"]
    if body["result"].get("isError"):
        raise RuntimeError(text)
    return json.loads(text)


def change(now, before):
    if not before:
        return "new"
    return f"{(now - before) / before:+.0%}"


lines = ["*Website summary: last 7 days*"]
for site in call_tool("list_sites"):
    overview = call_tool("get_overview", {"site_id": site["id"], "period": "7d"})
    now, before = overview["current"], overview["previous_period"]
    top_page = overview["top_pages"][0]["path"] if overview["top_pages"] else "none"
    lines.append(
        f"• *{site['name']}*: {now['visitors']} visitors ({change(now['visitors'], before['visitors'])}), "
        f"{now['conversions']} conversions, top page {top_page}"
    )

summary = "\n".join(lines)
print(summary)

if os.environ.get("SLACK_WEBHOOK_URL"):
    requests.post(os.environ["SLACK_WEBHOOK_URL"], json={"text": summary}, timeout=30)

It prints something like:

*Website summary: last 7 days*
• *Awesome Agency*: 105 visitors (-50%), 8 conversions, top page /
• *Harbour Dental*: 26 visitors (-56%), 0 conversions, top page /
• *Northwind Coffee*: 29 visitors (-61%), 0 conversions, top page /