MCP Tools for Grafana Loki — LogQL queries, log push, and label streaming

Loki architecture and label model for MCP tool authors

Loki's core design principle is "index-free log aggregation" — only labels are indexed, not log content. This means every query must start with a stream selector that specifies label matchers: {app="frontend",env="prod"}. Loki uses these labels to locate the compressed log chunks that could contain matching entries and then filters log line content within those chunks. A query without a stream selector is rejected with a parse error: unexpected end of stream selector error — this is the most common mistake when porting queries from Elasticsearch (which indexes all content) to Loki.

Loki supports three label matcher operators: = (exact match), != (negative exact match), =~ (regex match), !~ (negative regex). After the stream selector, log pipeline operators filter content: |= (contains string), != (does not contain), |~ (regex match on line content), !~ (negative regex). Parser stages (| json, | logfmt, | pattern) extract structured fields from log lines, enabling label filters on parsed values.

async function createLokiClient(config) {
  const baseHeaders = { 'Content-Type': 'application/json' };

  if (config.token) {
    // Grafana Cloud Loki: Bearer token auth
    baseHeaders['Authorization'] = `Bearer ${config.token}`;
  } else if (config.username && config.password) {
    // Grafana Cloud hosted Loki also accepts user:password (user ID : API key)
    const encoded = Buffer.from(`${config.username}:${config.password}`).toString('base64');
    baseHeaders['Authorization'] = `Basic ${encoded}`;
  }

  // Grafana Cloud URL format: https://{orgSlug}.grafana.net/loki/api/v1/
  // Self-hosted: http://loki:3100/loki/api/v1/

  async function get(path, params = {}) {
    const url = new URL(config.baseUrl + path);
    for (const [k, v] of Object.entries(params)) {
      if (v !== undefined) url.searchParams.set(k, v);
    }
    const response = await fetch(url.toString(), { headers: baseHeaders });
    if (!response.ok) {
      const text = await response.text();
      throw new Error(`Loki ${path} → ${response.status}: ${text}`);
    }
    return response.json();
  }

  async function post(path, body) {
    const response = await fetch(config.baseUrl + path, {
      method: 'POST',
      headers: baseHeaders,
      body: JSON.stringify(body),
    });
    if (!response.ok) {
      const text = await response.text();
      throw new Error(`Loki POST ${path} → ${response.status}: ${text}`);
    }
    return response.status === 204 ? null : response.json();
  }

  return { get, post };
}

const loki = await createLokiClient({
  baseUrl: 'https://logs-prod-us-central1.grafana.net',
  username: '12345',      // Grafana Cloud org ID (user ID)
  password: 'glc_xxx..', // Grafana Cloud API key
});

Log queries with query_range and instant query

The /loki/api/v1/query_range endpoint retrieves log entries over a time range. It returns a matrix of streams, each stream being a set of log entries with the same label set. The response format wraps results in data.result — an array of stream objects, each with a stream label map and a values array of [unix_nanoseconds_string, log_line] pairs.

The limit parameter controls how many log entries to return (default 100, max 5000). Time parameters are Unix timestamps in nanoseconds or RFC3339 strings. The direction parameter controls sort order: backward (default, newest first) or forward (oldest first). For MCP tools fetching recent errors, use backward with a limit. For tailing or replaying a time window in order, use forward.

async function queryLogs(loki, params) {
  const {
    logqlExpr,       // e.g. '{app="api"} |= "ERROR" | json | level="error"'
    startTime,       // RFC3339 or Unix nanoseconds string
    endTime,
    limit = 100,
    direction = 'backward', // 'backward' (newest first) or 'forward'
  } = params;

  const result = await loki.get('/loki/api/v1/query_range', {
    query: logqlExpr,
    start: startTime,
    end: endTime,
    limit: limit.toString(),
    direction,
  });

  if (result.data.resultType !== 'streams') {
    throw new Error(`Expected streams result, got ${result.data.resultType}`);
  }

  // Flatten streams into a sorted list of log entries
  const entries = [];
  for (const stream of result.data.result) {
    for (const [ts, line] of stream.values) {
      entries.push({
        timestamp: new Date(Number(BigInt(ts) / 1_000_000n)).toISOString(), // ns → ms → ISO
        labels: stream.stream,
        line,
      });
    }
  }

  // Sort by timestamp (streams may not be globally sorted)
  entries.sort((a, b) =>
    direction === 'backward'
      ? b.timestamp.localeCompare(a.timestamp)
      : a.timestamp.localeCompare(b.timestamp)
  );

  return {
    entries,
    streamCount: result.data.result.length,
    totalEntries: entries.length,
  };
}

// Example: fetch the last 50 error log lines from the api app in the last 15 minutes
const now = Date.now();
const fifteenMinutesAgo = now - 15 * 60 * 1000;

const errors = await queryLogs(loki, {
  logqlExpr: '{app="api",env="prod"} |= "ERROR"',
  startTime: new Date(fifteenMinutesAgo).toISOString(),
  endTime: new Date(now).toISOString(),
  limit: 50,
  direction: 'backward',
});

Metric queries: deriving time series from log streams

LogQL metric queries wrap a log stream selector in a rate or count function to produce time series values — useful for MCP tools that need to answer "how many errors per minute in the last hour?" without loading all log lines. The /loki/api/v1/query_range endpoint handles both log queries and metric queries depending on the LogQL expression.

Common metric functions: rate({...}[1m]) (per-second rate of log entries over a 1-minute window), count_over_time({...}[5m]) (absolute count in each 5-minute window), sum(rate({...}[1m])) (aggregate rate across all streams), topk(5, rate({...}[1m])) (top 5 streams by rate). The range vector selector (the [1m] in brackets) must be at least as large as the step parameter in the query, or you'll get gaps in the time series. When step is omitted, Loki computes it as (end - start) / 250.

async function getErrorRate(loki, app, env, lookbackMinutes = 60, stepSeconds = 60) {
  const end = new Date().toISOString();
  const start = new Date(Date.now() - lookbackMinutes * 60 * 1000).toISOString();

  // Metric query: per-second error rate, aggregated across all matching streams
  const logqlExpr = `sum(rate({app="${app}",env="${env}"} |= "error" [1m])) by (app)`;

  const result = await loki.get('/loki/api/v1/query_range', {
    query: logqlExpr,
    start,
    end,
    step: `${stepSeconds}s`,
  });

  if (result.data.resultType !== 'matrix') {
    throw new Error(`Expected matrix result for metric query, got ${result.data.resultType}`);
  }

  return result.data.result.map((series) => ({
    labels: series.metric,
    // Each value: [unix_timestamp_seconds_float, value_string]
    dataPoints: series.values.map(([ts, val]) => ({
      time: new Date(ts * 1000).toISOString(),
      errorsPerSecond: parseFloat(val),
    })),
  }));
}

Pushing logs from MCP tools to Loki

MCP tools that generate their own logs can push directly to Loki using the push API. The push endpoint at /loki/api/v1/push accepts a JSON body with a streams array. Each stream has a stream object (the label set) and a values array of [timestamp, line] pairs. Timestamps must be Unix nanoseconds expressed as strings — not milliseconds, not numbers. Sending milliseconds silently results in entries dated to 1970 (epoch + milliseconds interpreted as nanoseconds).

Loki enforces log ordering within a stream: entries must be pushed in ascending timestamp order for a given label set. Pushing out-of-order entries returns a 422 Unprocessable Entity with entry out of order. If your MCP tool processes batches that may contain out-of-order logs, sort entries by timestamp before pushing, or use different label values (like adding a unique request ID label) to route out-of-order entries into separate streams.

async function pushLogsToLoki(loki, entries) {
  // entries: Array<{ labels: Record, timestamp: Date, line: string }>

  // Group entries by label set (streams)
  const streamMap = new Map();

  for (const entry of entries) {
    const labelKey = JSON.stringify(Object.fromEntries(Object.entries(entry.labels).sort()));
    if (!streamMap.has(labelKey)) {
      streamMap.set(labelKey, { labels: entry.labels, values: [] });
    }
    // Timestamp must be nanoseconds as a string
    const tsNs = (BigInt(entry.timestamp.getTime()) * 1_000_000n).toString();
    streamMap.get(labelKey).values.push([tsNs, entry.line]);
  }

  // Sort each stream's values by timestamp (Loki requires ascending order per stream)
  const streams = Array.from(streamMap.values()).map((stream) => ({
    stream: stream.labels,
    values: stream.values.sort((a, b) => (BigInt(a[0]) < BigInt(b[0]) ? -1 : 1)),
  }));

  await loki.post('/loki/api/v1/push', { streams });
}

// Example: push MCP tool execution logs
await pushLogsToLoki(loki, [
  {
    labels: { app: 'my-mcp-tool', tool: 'search_codebase', env: 'prod' },
    timestamp: new Date(),
    line: JSON.stringify({ level: 'info', message: 'Tool invoked', query: 'foo', results: 12 }),
  },
]);

Label discovery and stream exploration

Before querying Loki, MCP tools can discover available labels and their values to build valid stream selectors dynamically. The labels API returns all label names seen in the configured time range (default: last 6 hours). The label values API returns all values for a specific label within the time range and an optional stream selector to scope the values.

async function discoverLokiStreams(loki, options = {}) {
  const { start, end, streamSelector } = options;

  // Get all label names
  const labelsResult = await loki.get('/loki/api/v1/labels', {
    ...(start && { start }),
    ...(end && { end }),
  });

  // Get values for key labels
  const labelValuePromises = ['app', 'env', 'service', 'job', 'namespace'].map(async (label) => {
    try {
      const valuesResult = await loki.get(`/loki/api/v1/label/${label}/values`, {
        ...(start && { start }),
        ...(end && { end }),
        ...(streamSelector && { query: streamSelector }),
      });
      return { label, values: valuesResult.data };
    } catch {
      return { label, values: [] };
    }
  });

  const labelValues = await Promise.all(labelValuePromises);

  return {
    allLabels: labelsResult.data,
    commonLabelValues: Object.fromEntries(
      labelValues.filter((l) => l.values.length > 0).map((l) => [l.label, l.values])
    ),
  };
}

// Get series metadata (unique label combinations) matching a selector
async function getStreamSeries(loki, streamSelector, start, end) {
  const result = await loki.get('/loki/api/v1/series', {
    match: streamSelector,
    start,
    end,
  });
  // Returns array of label set objects
  return result.data;
}

Loki readiness and AliveMCP integration

Loki's readiness model differs from a simple HTTP health check. The /ready endpoint reports whether the instance has joined the Loki ring — the distributed hash ring used by the ingester component to claim log stream ownership. A Loki instance that is starting up or that has been partitioned from the ring returns 404 Not Ready on the /ready endpoint even while /loki/api/v1/labels returns cached results. This makes the /ready endpoint the authoritative liveness check — use it for AliveMCP probes, not just a root path check.

For complete observability of a Loki deployment, combine three probes: the /ready endpoint (ingester ring membership), a /loki/api/v1/labels query (confirms the query path is functional — the querier and query-frontend components), and a push followed by an immediate query (confirms the write path and ingestion pipeline are functioning end-to-end). The last test can be done with a low-cardinality synthetic stream using a purpose="health_probe" label.

async function lokiHealthProbe(baseUrl, authHeaders) {
  const results = {
    ingesterReady: false,
    queryPathHealthy: false,
    writePipelineHealthy: false,
  };

  // 1. Ingester ring membership
  try {
    const readyResp = await fetch(`${baseUrl}/ready`, { headers: authHeaders });
    results.ingesterReady = readyResp.ok; // 200 = ready, 404/503 = not ready
    results.ingesterStatus = readyResp.status;
  } catch (err) {
    results.ingesterError = err.message;
  }

  // 2. Query path: labels endpoint (uses querier + query-frontend)
  try {
    const labelsResp = await fetch(`${baseUrl}/loki/api/v1/labels`, { headers: authHeaders });
    results.queryPathHealthy = labelsResp.ok;
  } catch (err) {
    results.queryError = err.message;
  }

  // 3. Write path: push a synthetic probe entry then query it back
  if (results.ingesterReady) {
    try {
      const probeTs = (BigInt(Date.now()) * 1_000_000n).toString();
      const probeId = Math.random().toString(36).slice(2);
      const probeBody = {
        streams: [{
          stream: { purpose: 'health_probe', probe_id: probeId },
          values: [[probeTs, `AliveMCP health probe ${probeId}`]],
        }],
      };
      const pushResp = await fetch(`${baseUrl}/loki/api/v1/push`, {
        method: 'POST',
        headers: { ...authHeaders, 'Content-Type': 'application/json' },
        body: JSON.stringify(probeBody),
      });
      results.writePipelineHealthy = pushResp.status === 204;
    } catch (err) {
      results.writeError = err.message;
    }
  }

  results.healthy = results.ingesterReady && results.queryPathHealthy;
  return results;
}

Register the /ready probe with AliveMCP at 30-second intervals. Set up a separate AliveMCP probe for the labels endpoint at 60-second intervals — this catches degraded query frontend performance without false-positive alerting from the ingester check.

Related guides

• MCP tools for Elasticsearch — Query DSL, cluster health, and search_after pagination
• MCP tools for Prometheus Alertmanager — alerts, silences, and inhibition rules
• Structured logging for MCP servers — JSON fields, correlation IDs, and log levels
• Log aggregation for MCP servers — centralized logging, retention, and search
• MCP server observability — metrics, logs, and traces for production MCP tools
• MCP server health check patterns — composite endpoints and failure classification