MCP Gateway Log Export API
Use the Log Export API to read MCP Gateway tool call logs from your own data platform, SIEM, warehouse, or scheduled export job.
The API returns newline-delimited JSON. Each response line is one complete JSON object, so clients can stream, parse, and checkpoint logs incrementally.
GET https://<your-gateway-host>/mcpgateway/logs/export
Response content type:
Content-Type: application/x-ndjson
Authentication
Pass a log export key from MCP Gateway:
X-Quilr-Log-Export-Key: sk-export-...
Do not use your MCP API token or admin API key as the request credential for this endpoint. The log export key is separate from tool-call authentication and management API authentication.
MCP Gateway exposes two export scopes:
Both scopes use the same endpoint, header, query parameters, pagination model, and response format. In tenant-scoped exports, each mcpgateway.tool_call event still includes the concrete backend in backend.id and backend.name.
Admin backend management responses include log_export_key for each backend and tenant_log_export_key for the tenant. Treat both as bearer secrets. They cannot call MCP tools, but they can read exportable tool call logs for their scope.
Query Parameters
All query parameters are optional.
Logs are available for a maximum of 15 days. Choose start_time within that retention window when backfilling. Requests with an effective start_time, end_time, or cursor timestamp before the retention window fail with 400.
If neither start_time nor cursor is provided, the API exports a default 24-hour window ending at the effective export end time.
Export Lag
The API does not export logs newer than 15 minutes. Gateway logs and prediction payloads are written asynchronously, so this lag keeps exported rows stable.
If end_time is newer than now - 15 minutes, the server clamps it to the maximum exportable time. The request still succeeds. The export_started and checkpoint events include the effective export bounds.
Request Examples
Start a backend-scoped export window, here the hour that ended two hours ago (logs are kept for 15 days, so fixed dates go stale):
# macOS (BSD date) first, GNU date as fallback
START=$(date -u -v-3H +%Y-%m-%dT%H:%M:%SZ 2>/dev/null || date -u -d '3 hours ago' +%Y-%m-%dT%H:%M:%SZ)
END=$(date -u -v-2H +%Y-%m-%dT%H:%M:%SZ 2>/dev/null || date -u -d '2 hours ago' +%Y-%m-%dT%H:%M:%SZ)
curl -N \
-H "X-Quilr-Log-Export-Key: sk-export-..." \
"https://<your-gateway-host>/mcpgateway/logs/export?start_time=$START&end_time=$END&limit=1000"
Start a tenant-scoped export window:
curl -N \
-H "X-Quilr-Log-Export-Key: sk-exportl-..." \
"https://<your-gateway-host>/mcpgateway/logs/export?start_time=$START&end_time=$END&limit=1000"
Resume from the previous checkpoint:
curl -N \
-H "X-Quilr-Log-Export-Key: sk-export-..." \
"https://<your-gateway-host>/mcpgateway/logs/export?cursor=<next_cursor>"
When resuming with cursor, you do not need to pass start_time or end_time.
Pagination
Rows are ordered by:
started_at ASC, id ASC
The cursor is opaque. Store it exactly as returned in checkpoint.next_cursor and send it back as the cursor query parameter on the next request.
If checkpoint.has_more is true, call the endpoint again immediately with cursor=<next_cursor>.
If checkpoint.has_more is false, there are no more rows in the current effective window. Store next_cursor and poll later with that cursor to continue incremental export.
When an initial request returns zero rows, the API still returns a checkpoint cursor pinned to the effective end time. This lets exporters store one cursor value even for empty windows.
A reliable collector
A minimal Python collector (standard library only) that you can run on a schedule or as a service. It:
- stores rows and the cursor in one SQLite transaction, so the cursor only advances after the page is safely written;
- writes idempotently (
INSERT OR IGNOREon a unique id), so a replayed page never creates duplicates; - treats a mid-stream
errorevent or a stream without acheckpointas a failed page, and retries it with exponential backoff and jitter; - stops on
400,401and403, which retrying cannot fix.
"""Minimal MCP Gateway log collector: cursor checkpoint, retries, idempotent writes."""
import json, os, random, sqlite3, time, urllib.error, urllib.parse, urllib.request
from datetime import datetime, timedelta, timezone
URL = os.environ["EXPORT_URL"] # https://<your-gateway-host>/mcpgateway/logs/export
KEY = os.environ["QUILR_LOG_EXPORT_KEY"]
EVENT = "mcpgateway.tool_call"
row_id = lambda e: str(e["request"]["log_id"]) # unique per tool call log row
db = sqlite3.connect("quilr_logs.db")
db.execute("CREATE TABLE IF NOT EXISTS events (id TEXT PRIMARY KEY, ts TEXT, body TEXT)")
db.execute("CREATE TABLE IF NOT EXISTS state (k TEXT PRIMARY KEY, v TEXT)")
def saved_cursor():
row = db.execute("SELECT v FROM state WHERE k = 'cursor'").fetchone()
return row[0] if row else None
def fetch_page(params):
"""Return (events, checkpoint) for one complete page, or raise."""
req = urllib.request.Request(URL + "?" + urllib.parse.urlencode(params),
headers={"X-Quilr-Log-Export-Key": KEY})
events, checkpoint = [], None
with urllib.request.urlopen(req, timeout=300) as r: # raises HTTPError on 4xx/5xx
for line in r:
if not line.strip():
continue
msg = json.loads(line)
if msg["type"] == "error": # can arrive after HTTP 200
raise RuntimeError(f"export error: {msg['error']}")
if msg["type"] == EVENT:
events.append(msg)
elif msg["type"] == "checkpoint":
checkpoint = msg
if checkpoint is None:
raise RuntimeError("stream ended without a checkpoint")
return events, checkpoint
def with_retries(fn, *args, attempts=6):
for n in range(attempts):
try:
return fn(*args)
except urllib.error.HTTPError as e:
if e.code in (400, 401, 403) or n == attempts - 1:
raise # fix the key or the window; retrying will not help
except (OSError, RuntimeError, ValueError): # network, mid-stream error, truncation
if n == attempts - 1:
raise
time.sleep(min(60, 2 ** n) + random.random()) # backoff with jitter
def run_once():
cursor = saved_cursor()
if cursor:
params = {"cursor": cursor, "limit": 1000}
else: # first run: start one hour back, well inside the 15-day retention
start = datetime.now(timezone.utc) - timedelta(hours=1)
params = {"start_time": start.strftime("%Y-%m-%dT%H:%M:%SZ"), "limit": 1000}
while True:
events, checkpoint = with_retries(fetch_page, params)
with db: # one transaction: rows and cursor commit together
db.executemany(
"INSERT OR IGNORE INTO events VALUES (?, ?, ?)",
[(row_id(e), e["request"]["started_at"], json.dumps(e)) for e in events])
db.execute("INSERT OR REPLACE INTO state VALUES ('cursor', ?)",
(checkpoint["next_cursor"],))
print(f"stored {checkpoint['rows']} rows up to {checkpoint['effective_end_time']}")
if not checkpoint["has_more"]:
return
params = {"cursor": checkpoint["next_cursor"], "limit": 1000}
if __name__ == "__main__":
while True:
run_once()
time.sleep(300) # new rows appear about 15 minutes after the request
export EXPORT_URL='https://<your-gateway-host>/mcpgateway/logs/export'
export QUILR_LOG_EXPORT_KEY='sk-export-...'
python3 collector.py
To forward to a SIEM instead of SQLite, replace the with db: block with your sink's write, and persist next_cursor only after the sink acknowledges the batch. If the collector is down for longer than the 15-day retention, the saved cursor fails with 400; delete it to restart from a recent start_time.
Verify delivery
- Call one tool through an MCP covered by the export key and note the time and tool name.
- After at least 15 minutes, run the collector once.
- Check that the row arrived:
sqlite3 quilr_logs.db "SELECT id, ts, json_extract(body, '$.tool.name') FROM events ORDER BY ts DESC LIMIT 5".
Coverage
The export covers MCP Gateway tool call traffic for the selected export scope, including:
Response Events
Every successful response starts with export_started, contains zero or more mcpgateway.tool_call events, and ends with checkpoint.
export_started
The first line describes the export scope and effective export window.
{
"type": "export_started",
"schema_version": "v1",
"backend": {
"id": "backend_a1b2c3",
"name": "My Jira MCP"
},
"effective_start_time": "2026-05-14T00:00:00.000000Z",
"effective_end_time": "2026-05-14T01:00:00.000000Z",
"max_exportable_time": "2026-05-14T10:45:00.000000Z",
"end_time_clamped": false,
"limit": 1000
}
For tenant-scoped export, the first line uses this scope shape:
{
"type": "export_started",
"schema_version": "v1",
"scope": {
"type": "tenant",
"tenant_id": "tenant_abc123",
"backend_count": 3
},
"effective_start_time": "2026-05-14T00:00:00.000000Z",
"effective_end_time": "2026-05-14T01:00:00.000000Z",
"max_exportable_time": "2026-05-14T10:45:00.000000Z",
"end_time_clamped": false,
"limit": 1000
}
scope
mcpgateway.tool_call
Each tool call row is emitted as one mcpgateway.tool_call event.
{
"type": "mcpgateway.tool_call",
"schema_version": "v1",
"cursor": "<opaque-cursor>",
"backend": {
"id": "backend_a1b2c3",
"name": "My Jira MCP"
},
"request": {
"id": "jsonrpc-request-id",
"log_id": 123,
"started_at": "2026-05-14T00:00:01.123456Z",
"completed_at": "2026-05-14T00:00:01.573456Z",
"duration_ms": 450
},
"auth": {
"mode": "token"
},
"tool": {
"name": "create_issue"
},
"guardrails": {
"input": {
"outcome": "allowed",
"is_blocked": false,
"predictions": []
},
"output": {
"outcome": "allowed",
"is_blocked": false,
"predictions": []
}
},
"payload": {
"tool_arguments": {},
"response_content": {}
},
"response": {
"success": true,
"error_message": null
},
"metadata": {
"user_email": "dev@company.com",
"agent": "cursor",
"extra_data": {}
}
}
Top-Level Fields
backend
request
auth
tool
guardrails
Each guardrail result uses this shape:
payload
The export redacts common credential fields and token patterns inside payloads, predictions, and metadata. Header-like objects are replaced with [REDACTED_HEADERS].
response
metadata
checkpoint
The final line on a successful response is a checkpoint.
{
"type": "checkpoint",
"schema_version": "v1",
"next_cursor": "<opaque-cursor>",
"rows": 1000,
"has_more": true,
"effective_end_time": "2026-05-14T01:00:00.000000Z",
"max_exportable_time": "2026-05-14T10:45:00.000000Z"
}
Errors
Errors are returned as NDJSON too.
{"type":"error","error":{"message":"<message>","code":"<code>"}}
Errors before streaming starts use HTTP status codes. Errors after streaming has started are emitted as an error event line because the HTTP response has already been committed.