pip install mcp-audit

from mcp_audit import AuditEngine, traced_call

engine = AuditEngine()
session = engine.start_session(agent_id="my-agent")

# Wrap any function call with automatic tracing
with traced_call(engine, session_id=session.id, tool_name="web_search") as tc:
    tc.set_cost(0.003)
    tc.set_tokens(input_tokens=500, output_tokens=200)
    result = search("best MCP servers")
    tc.set_result(result)

# Query analytics
stats = engine.get_stats(session_id=session.id)
print(f"Total cost: ${stats['total_cost_usd']}")
print(f"P95 latency: {stats['p95_latency_ms']}ms")
print(f"Error rate: {stats['error_rate']}%")

{
  "mcpServers": {
    "mcp-audit": {
      "command": "mcp-audit",
      "args": ["stdio"]
    }
  }
}

from mcp_audit import create_fastmcp_server

# Get a FastMCP instance with all 17 tools registered
server = create_fastmcp_server()
server.run(transport="stdio")

from mcp_audit import AuditEngine
from mcp_audit.sqlite_store import SQLiteStore

store = SQLiteStore("audit.db")  # persists to disk
engine = AuditEngine(store=store)

# All calls, sessions, events, and rules now survive process restarts
session = engine.start_session(agent_id="prod-agent")

from mcp_audit import AuditEngine
from mcp_audit.decorator import audit_call, bind_session

engine = AuditEngine()
session = engine.start_session(agent_id="my-agent")

# Option 1: explicit engine + session
@audit_call(engine, session_id=session.id, cost_fn=lambda *_: 0.001)
def search(query: str) -> list:
    return [{"title": "result"}]

# Option 2: bind once, decorate everywhere
ctx = bind_session(engine, session.id)

@audit_call()  # uses bound engine + session
def fetch(url: str) -> dict:
    return requests.get(url).json()

@audit_call(tool_name="llm_complete", cost_fn=compute_cost)
def complete(prompt: str) -> str:
    return llm.generate(prompt)

ctx.reset()  # unbind when done

from mcp_audit.export import export_calls_jsonl, export_calls_csv

# JSONL for log shippers (Datadog, Splunk, ELK)
export_calls_jsonl(engine, "audit.jsonl", session_id=sid)

# CSV for spreadsheet analysis
export_calls_csv(engine, "costs.csv", agent_id="prod-agent")

# Or export to a string
from mcp_audit.export import export_to_string
text = export_to_string(engine, fmt="jsonl", limit=100)

from mcp_audit.otlp import export_otlp_http, export_otlp_jsonl
from mcp_audit.metrics import export_otlp_metrics_http, build_metrics

# Export traces to a local OTel collector
export_otlp_http(engine, endpoint="http://localhost:4318/v1/traces")

# Export metrics (counters, histograms, gauges) to OTel collector
export_otlp_metrics_http(engine, endpoint="http://localhost:4318/v1/metrics")

# Or build OTLP metric dicts for inspection
metrics = build_metrics(engine)

from mcp_audit.prometheus import build_prometheus_exposition, export_prometheus_file

# Build exposition text (serve on /metrics endpoint)
text = build_prometheus_exposition(engine)

# Write to file for node_exporter textfile collector
export_prometheus_file(engine, "/var/lib/node_exporter/mcp_audit.prom")

from flask import Flask
from mcp_audit.prometheus import build_prometheus_exposition, PROMETHEUS_CONTENT_TYPE

app = Flask(__name__)

@app.route("/metrics")
def metrics():
    text = build_prometheus_exposition(engine)
    return text, 200, {"Content-Type": PROMETHEUS_CONTENT_TYPE}

from mcp_audit.prometheus import export_prometheus_http

export_prometheus_http(engine, endpoint="http://localhost:9091", job_name="mcp-audit")

engine.create_rule(
    name="high_error_rate",
    metric="error_rate",       # error_rate | p95_latency | cost_per_call | total_cost | call_volume
    operator=">",              # > | >= | < | <= | ==
    threshold=50.0,            # threshold value
    window=100,                # evaluate last N calls
)

engine.create_rule(
    name="budget_exceeded",
    metric="total_cost",
    operator=">=",
    threshold=10.0,
)

# Check if any rules are breached
alerts = engine.evaluate_rules()

┌─────────────────────────────────────────────────────┐
│                  AI Agent (Claude, etc.)              │
│                                                       │
│  ┌──────────┐  ┌──────────┐  ┌───────────────────┐  │
│  │ MCP Tool │  │ MCP Tool │  │   mcp-audit       │  │
│  │ Server A │  │ Server B │  │   (this server)   │  │
│  └────┬─────┘  └────┬─────┘  └────────┬──────────┘  │
│       │              │                  │             │
│       └──────┬───────┘                  │             │
│              │  Agent calls record_call │             │
│              └──────────────────────────┘             │
└─────────────────────────────────────────────────────┘

git clone https://github.com/nyx-builds/mcp-audit.git
cd mcp-audit
uv venv .venv
VIRTUAL_ENV=$(pwd)/.venv uv pip install -e ".[dev]"
.venv/bin/python -m pytest -q