MCP Server LLM cost tracking — token usage headers, budget guardrails, per-tool attribution, spend alerts

Token usage extraction across providers

Build a provider-agnostic adapter that normalizes token counts from OpenAI and Anthropic responses into a single shape. Anthropic Claude models support prompt caching — cache reads cost ~10% of base input price and cache writes cost ~25% more, so they need separate tracking to calculate accurate cost.

import OpenAI from 'openai';
import Anthropic from '@anthropic-ai/sdk';
import Database from 'better-sqlite3';

const openai    = new OpenAI();
const anthropic = new Anthropic();

// Pricing table (USD per 1M tokens) — update when providers change rates
const MODEL_PRICING: Record<string, {
  input:       number;
  output:      number;
  cacheRead?:  number;
  cacheWrite?: number;
}> = {
  'gpt-4o':                       { input: 2.50,  output: 10.00 },
  'gpt-4o-mini':                  { input: 0.15,  output: 0.60  },
  'gpt-4.1':                      { input: 2.00,  output: 8.00  },
  'claude-opus-4-7':              { input: 15.00, output: 75.00, cacheRead: 1.50,  cacheWrite: 18.75 },
  'claude-sonnet-4-6':            { input: 3.00,  output: 15.00, cacheRead: 0.30,  cacheWrite: 3.75  },
  'claude-haiku-4-5-20251001':    { input: 0.80,  output: 4.00,  cacheRead: 0.08,  cacheWrite: 1.00  },
};

interface UsageRecord {
  modelId:          string;
  inputTokens:      number;
  outputTokens:     number;
  cacheReadTokens:  number;
  cacheWriteTokens: number;
  costUsd:          number;
}

function calcCost(modelId: string, usage: Omit<UsageRecord, 'costUsd' | 'modelId'>): number {
  // Match by prefix for versioned model IDs (e.g. 'gpt-4o-mini-2024-07-18' → 'gpt-4o-mini')
  const priceKey = Object.keys(MODEL_PRICING).find(k =>
    modelId === k || modelId.startsWith(k)
  ) ?? modelId;

  const prices = MODEL_PRICING[priceKey];
  if (!prices) return 0;  // unknown model — log but don't block

  const perMillion = 1_000_000;
  return (
    (usage.inputTokens      * (prices.input      ?? 0)) / perMillion +
    (usage.outputTokens     * (prices.output     ?? 0)) / perMillion +
    (usage.cacheReadTokens  * (prices.cacheRead  ?? 0)) / perMillion +
    (usage.cacheWriteTokens * (prices.cacheWrite ?? 0)) / perMillion
  );
}

// OpenAI usage extractor — handles both cached (prompt_tokens_details) and uncached
function extractOpenAIUsage(completion: OpenAI.Chat.ChatCompletion): Omit<UsageRecord, 'costUsd'> {
  const usage  = completion.usage;
  const modelId = completion.model;  // use response.model, NOT the request model param

  const cacheReadTokens = usage?.prompt_tokens_details?.cached_tokens ?? 0;

  return {
    modelId,
    inputTokens:      (usage?.prompt_tokens     ?? 0) - cacheReadTokens,
    outputTokens:     usage?.completion_tokens  ?? 0,
    cacheReadTokens,
    cacheWriteTokens: 0,  // OpenAI doesn't distinguish cache writes separately
  };
}

// Anthropic usage extractor — cache read and write billed separately
function extractAnthropicUsage(message: Anthropic.Message): Omit<UsageRecord, 'costUsd'> {
  const usage   = message.usage;
  const modelId = message.model;  // from response, not request

  return {
    modelId,
    inputTokens:      usage.input_tokens,
    outputTokens:     usage.output_tokens,
    cacheReadTokens:  usage.cache_read_input_tokens  ?? 0,
    cacheWriteTokens: usage.cache_creation_input_tokens ?? 0,
  };
}

Streaming token extraction

Streaming responses accumulate usage on the final chunk when stream_options: { include_usage: true } is set. For Anthropic streaming, usage appears in the message_delta event with type 'message_delta'. Never estimate tokens by counting words — always use the provider's reported usage for billing accuracy.

// OpenAI streaming with accurate token extraction
async function openaiStreamWithUsage(
  params: OpenAI.Chat.ChatCompletionCreateParamsStreaming
): Promise<{ text: string; usage: Omit<UsageRecord, 'costUsd'> }> {
  const stream = await openai.chat.completions.create({
    ...params,
    stream:         true,
    stream_options: { include_usage: true },  // required — omit and usage is null
  });

  let text         = '';
  let finalModelId = params.model;  // fallback — overwritten by response model
  let usage: OpenAI.CompletionUsage | null = null;

  for await (const chunk of stream) {
    text += chunk.choices[0]?.delta?.content ?? '';

    // Model ID from response — may differ from requested model (e.g. proxy routing)
    if (chunk.model) finalModelId = chunk.model;

    // usage only present on the final chunk when include_usage: true
    if (chunk.usage) usage = chunk.usage;
  }

  const cacheReadTokens = usage?.prompt_tokens_details?.cached_tokens ?? 0;

  return {
    text,
    usage: {
      modelId:          finalModelId,
      inputTokens:      (usage?.prompt_tokens     ?? 0) - cacheReadTokens,
      outputTokens:     usage?.completion_tokens  ?? 0,
      cacheReadTokens,
      cacheWriteTokens: 0,
    },
  };
}

// Anthropic streaming — usage in message_delta event
async function anthropicStreamWithUsage(
  params: Anthropic.MessageCreateParamsStreaming
): Promise<{ text: string; usage: Omit<UsageRecord, 'costUsd'> }> {
  const stream = anthropic.messages.stream(params);

  let text         = '';
  let finalModelId = params.model;
  let inputTokens  = 0;
  let outputTokens = 0;
  let cacheReadTokens  = 0;
  let cacheWriteTokens = 0;

  for await (const event of stream) {
    if (event.type === 'content_block_delta' && event.delta.type === 'text_delta') {
      text += event.delta.text;
    }

    // message_start contains input token counts (including cache tokens)
    if (event.type === 'message_start') {
      const u = event.message.usage;
      inputTokens      = u.input_tokens;
      cacheReadTokens  = u.cache_read_input_tokens  ?? 0;
      cacheWriteTokens = u.cache_creation_input_tokens ?? 0;
      finalModelId     = event.message.model;
    }

    // message_delta contains output token count
    if (event.type === 'message_delta' && event.usage) {
      outputTokens = event.usage.output_tokens;
    }
  }

  return {
    text,
    usage: {
      modelId:          finalModelId,
      inputTokens:      inputTokens - cacheReadTokens,
      outputTokens,
      cacheReadTokens,
      cacheWriteTokens,
    },
  };
}

Per-tool cost ledger and budget guardrails

Store every LLM call in a SQLite cost ledger. Before each call, query the hourly spend and reject the call if it would push over the soft limit. Log after each call with the actual model ID from the response — not the requested model — so cost attribution is accurate when providers route to different models.

import Database from 'better-sqlite3';
import { z } from 'zod';

const db = new Database('./data.db');

// One-time schema setup
db.exec(`
  CREATE TABLE IF NOT EXISTS llm_cost_log (
    id         INTEGER PRIMARY KEY AUTOINCREMENT,
    tool_name  TEXT    NOT NULL,
    model_id   TEXT    NOT NULL,
    input_tok  INTEGER NOT NULL DEFAULT 0,
    output_tok INTEGER NOT NULL DEFAULT 0,
    cache_read INTEGER NOT NULL DEFAULT 0,
    cache_write INTEGER NOT NULL DEFAULT 0,
    cost_usd   REAL    NOT NULL DEFAULT 0,
    session_id TEXT,
    created_at TEXT    NOT NULL DEFAULT (datetime('now'))
  );
  CREATE INDEX IF NOT EXISTS idx_llm_cost_log_created ON llm_cost_log(created_at);
  CREATE INDEX IF NOT EXISTS idx_llm_cost_log_tool    ON llm_cost_log(tool_name, created_at);
`);

const insertCostLog = db.prepare(`
  INSERT INTO llm_cost_log (tool_name, model_id, input_tok, output_tok, cache_read, cache_write, cost_usd, session_id)
  VALUES (?, ?, ?, ?, ?, ?, ?, ?)
`);

const getHourlySpend = db.prepare(`
  SELECT COALESCE(SUM(cost_usd), 0) AS total
  FROM llm_cost_log
  WHERE created_at > datetime('now', '-1 hour')
`);

const getToolHourlySpend = db.prepare(`
  SELECT COALESCE(SUM(cost_usd), 0) AS total
  FROM llm_cost_log
  WHERE tool_name = ? AND created_at > datetime('now', '-1 hour')
`);

// Soft limits (USD) — adjust per your budget
const HOURLY_LIMIT_USD      = parseFloat(process.env.LLM_HOURLY_LIMIT_USD ?? '5.00');
const PER_TOOL_HOURLY_LIMIT = parseFloat(process.env.LLM_PER_TOOL_LIMIT_USD ?? '2.00');

function checkBudget(toolName: string): void {
  const hourlyTotal    = (getHourlySpend.get() as { total: number }).total;
  const toolHourly     = (getToolHourlySpend.get(toolName) as { total: number }).total;

  if (hourlyTotal >= HOURLY_LIMIT_USD) {
    throw new Error(
      `Hourly LLM budget exceeded ($${hourlyTotal.toFixed(4)} / $${HOURLY_LIMIT_USD}). ` +
      `Reset in ${60 - new Date().getMinutes()} minutes.`
    );
  }
  if (toolHourly >= PER_TOOL_HOURLY_LIMIT) {
    throw new Error(
      `Per-tool hourly budget exceeded for '${toolName}' ($${toolHourly.toFixed(4)} / $${PER_TOOL_HOURLY_LIMIT}).`
    );
  }
}

function logCost(toolName: string, usageRecord: UsageRecord, sessionId?: string): void {
  insertCostLog.run(
    toolName,
    usageRecord.modelId,
    usageRecord.inputTokens,
    usageRecord.outputTokens,
    usageRecord.cacheReadTokens,
    usageRecord.cacheWriteTokens,
    usageRecord.costUsd,
    sessionId ?? null
  );
}

// MCP tool with full cost tracking + guardrail
server.tool(
  'generate_report',
  {
    topic:      z.string().min(1).max(500),
    session_id: z.string().optional(),
  },
  async ({ topic, session_id }) => {
    // Check budget before making the call — prevent runaway spend
    checkBudget('generate_report');

    const completion = await openai.chat.completions.create({
      model:      'gpt-4o-mini',
      messages:   [{ role: 'user', content: `Write a short report on: ${topic}` }],
      max_tokens: 512,
    });

    const rawUsage = extractOpenAIUsage(completion);
    const costUsd  = calcCost(rawUsage.modelId, rawUsage);

    logCost('generate_report', { ...rawUsage, costUsd }, session_id);

    const text = completion.choices[0]?.message?.content ?? '';
    return {
      content: [{
        type: 'text',
        text: JSON.stringify({
          report:       text,
          cost_usd:     costUsd,
          model_used:   rawUsage.modelId,
          input_tokens: rawUsage.inputTokens + rawUsage.cacheReadTokens,
          output_tokens: rawUsage.outputTokens,
        }),
      }],
    };
  }
);

Spend reporting and anomaly detection

Expose a cost summary resource so agents can query spend before initiating expensive workflows. Add an anomaly check that triggers when per-tool cost spikes suddenly — a 10× increase over the rolling average is a reliable signal of prompt injection, runaway loops, or an accidental switch to a premium model.

// MCP resource — spending report by tool and model for the last 24h
server.resource('llm_spend_report', 'llm://spend/24h', async () => {
  const byTool = db.prepare(`
    SELECT
      tool_name,
      model_id,
      COUNT(*)           AS calls,
      SUM(input_tok)     AS total_input_tokens,
      SUM(output_tok)    AS total_output_tokens,
      SUM(cost_usd)      AS total_cost_usd,
      AVG(cost_usd)      AS avg_cost_per_call,
      MAX(cost_usd)      AS max_cost_per_call
    FROM llm_cost_log
    WHERE created_at > datetime('now', '-24 hours')
    GROUP BY tool_name, model_id
    ORDER BY total_cost_usd DESC
  `).all();

  const totals = db.prepare(`
    SELECT
      SUM(cost_usd) AS total_usd,
      COUNT(*)      AS total_calls,
      SUM(input_tok + cache_read) AS total_input_tokens,
      SUM(output_tok)             AS total_output_tokens
    FROM llm_cost_log
    WHERE created_at > datetime('now', '-24 hours')
  `).get() as { total_usd: number; total_calls: number; total_input_tokens: number; total_output_tokens: number };

  return {
    contents: [{
      uri:  'llm://spend/24h',
      text: JSON.stringify({
        period:       'last_24h',
        total_usd:    totals?.total_usd    ?? 0,
        total_calls:  totals?.total_calls  ?? 0,
        by_tool:      byTool,
        generated_at: new Date().toISOString(),
      }),
    }],
  };
});

// Background anomaly check — call from a periodic health job
async function checkSpendAnomaly(toolName: string): Promise<void> {
  const recent = db.prepare(`
    SELECT AVG(cost_usd) AS avg_cost
    FROM llm_cost_log
    WHERE tool_name = ? AND created_at > datetime('now', '-1 hour')
  `).get(toolName) as { avg_cost: number | null };

  const baseline = db.prepare(`
    SELECT AVG(cost_usd) AS avg_cost
    FROM llm_cost_log
    WHERE tool_name = ?
      AND created_at > datetime('now', '-7 days')
      AND created_at < datetime('now', '-1 hour')
  `).get(toolName) as { avg_cost: number | null };

  const recentAvg   = recent.avg_cost   ?? 0;
  const baselineAvg = baseline.avg_cost ?? 0;

  if (baselineAvg > 0 && recentAvg > baselineAvg * 10) {
    console.error(
      `[cost-anomaly] Tool '${toolName}' average cost spiked ` +
      `${(recentAvg / baselineAvg).toFixed(1)}× above 7-day baseline ` +
      `($${recentAvg.toFixed(6)} vs $${baselineAvg.toFixed(6)} avg per call). ` +
      `Check for model changes, prompt injection, or runaway loops.`
    );
    // In production: send alert to Slack, PagerDuty, or your on-call channel
  }
}

Run checkSpendAnomaly from your MCP server's startup or from a setInterval every 15 minutes. A 10× spike in average call cost is a strong signal — normal prompt variation rarely changes cost by more than 2–3×. The anomaly check catches accidental model upgrades, prompt injections that dramatically increase output length, and infinite retry loops before they exhaust your budget.