MCP Tools for InfluxDB — Flux queries, write API, and bucket management

InfluxDB v2 data model and authentication

InfluxDB v2 organizes data in a hierarchy: Organization → Bucket → Measurement → Tags + Fields + Timestamp. A measurement is the top-level grouping (analogous to a table name), tags are indexed string key-value pairs that appear in every data point for that series, and fields are the actual numeric or string measurements. Each data point has exactly one timestamp.

Tags are immutable after writing — you cannot update a tag value for existing data. If you need to correct a tag, you must delete and re-write the affected time range. This is a common gotcha for MCP tools that need to enrich or correct historical data: what looks like an "update" is actually a delete-and-reinsert operation that may violate retention policy constraints if the data is older than the bucket's retention period.

InfluxDB v2 uses token-based authentication. All-access tokens (created at setup) have full read/write access to all buckets in the organization. For MCP tools, create scoped tokens with only the necessary bucket permissions — read-only tokens for query tools, write-only tokens for data ingestion tools, and separate tokens per bucket if your MCP tool serves multiple tenants.

async function createInfluxClient(config) {
  const headers = {
    'Authorization': `Token ${config.token}`,
    'Content-Type': 'application/json',
  };

  const orgId = config.orgId; // InfluxDB v2 org ID (hex string, not org name)

  async function query(fluxScript) {
    const response = await fetch(`${config.baseUrl}/api/v2/query?org=${orgId}`, {
      method: 'POST',
      headers: {
        ...headers,
        'Content-Type': 'application/vnd.flux',
        'Accept': 'application/csv', // InfluxDB returns annotated CSV
      },
      body: fluxScript,
    });

    if (!response.ok) {
      const body = await response.json().catch(() => ({ message: response.statusText }));
      throw new Error(`InfluxDB query error: ${body.message ?? body.error}`);
    }

    return parseAnnotatedCSV(await response.text());
  }

  async function write(bucket, lineProtocolLines, precision = 'ns') {
    const body = Array.isArray(lineProtocolLines)
      ? lineProtocolLines.join('\n')
      : lineProtocolLines;

    const response = await fetch(
      `${config.baseUrl}/api/v2/write?org=${orgId}&bucket=${encodeURIComponent(bucket)}&precision=${precision}`,
      {
        method: 'POST',
        headers: { ...headers, 'Content-Type': 'text/plain; charset=utf-8' },
        body,
      }
    );

    if (!response.ok) {
      const errBody = await response.json().catch(() => ({}));
      throw new Error(`InfluxDB write error: ${errBody.message ?? response.status}`);
    }
    // 204 No Content on success
  }

  async function getRequest(path) {
    const response = await fetch(`${config.baseUrl}${path}`, { headers });
    if (!response.ok) throw new Error(`InfluxDB GET ${path} → ${response.status}`);
    return response.json();
  }

  return { query, write, getRequest, orgId };
}

const influx = await createInfluxClient({
  baseUrl: 'https://us-east-1-1.aws.cloud2.influxdata.com',
  token: 'my-influx-token==',
  orgId: 'a1b2c3d4e5f67890',
});

Flux queries: range, filter, aggregations, and pivoting

Flux is a functional data scripting language. Every Flux query begins with a data source declaration (from(bucket: "...")) followed by a range() call that restricts the time window, then one or more transformation functions piped with |>. Unlike SQL, Flux operates on table streams — a query produces a stream of tables, where each table has a consistent schema (same field and tag columns). Functions like filter() reduce rows, aggregateWindow() groups by time buckets, pivot() reshapes columns, and join() combines multiple streams.

A common Flux pattern for MCP tools is aggregateWindow to downsample time series for dashboards — returning one data point per minute or hour rather than every raw measurement. The createEmpty: true option fills in empty time buckets with null values, making it easy to detect gaps in data collection. A bucket with gaps in expected data is often a symptom of a failing collector or a downed scrape target that should trigger an AliveMCP alert.

// Query recent metrics for a specific measurement and field
async function queryTimeSeries(influx, params) {
  const {
    bucket,
    measurement,
    field,
    tags = {},           // e.g. { host: 'web-01', region: 'us-east' }
    start = '-1h',       // Flux relative range: -1h, -30m, -7d, or absolute RFC3339
    stop = 'now()',
    aggregateEvery,      // e.g. '1m', '5m', '1h' — omit for raw data
    aggregateFn = 'mean',
  } = params;

  // Build tag filter expressions
  const tagFilters = Object.entries(tags)
    .map(([k, v]) => `r["${k}"] == "${v}"`)
    .join(' and ');
  const tagFilterExpr = tagFilters ? ` and ${tagFilters}` : '';

  let flux;
  if (aggregateEvery) {
    flux = `
from(bucket: "${bucket}")
  |> range(start: ${start}, stop: ${stop})
  |> filter(fn: (r) => r._measurement == "${measurement}"
      and r._field == "${field}"${tagFilterExpr})
  |> aggregateWindow(
      every: ${aggregateEvery},
      fn: ${aggregateFn},
      createEmpty: true
     )
  |> yield(name: "${aggregateFn}")`;
  } else {
    flux = `
from(bucket: "${bucket}")
  |> range(start: ${start}, stop: ${stop})
  |> filter(fn: (r) => r._measurement == "${measurement}"
      and r._field == "${field}"${tagFilterExpr})
  |> yield(name: "raw")`;
  }

  return influx.query(flux);
}

// Detect data gaps: find time buckets with no data in last 2 hours
async function detectDataGaps(influx, bucket, measurement, field, expectedIntervalMinutes) {
  const flux = `
from(bucket: "${bucket}")
  |> range(start: -2h)
  |> filter(fn: (r) => r._measurement == "${measurement}" and r._field == "${field}")
  |> aggregateWindow(
      every: ${expectedIntervalMinutes}m,
      fn: count,
      createEmpty: true
     )
  |> filter(fn: (r) => r._value == 0 or r._value == null)
  |> yield(name: "gaps")`;

  const gaps = await influx.query(flux);
  return {
    gapCount: gaps.length,
    gaps: gaps.map((r) => ({ time: r._time, bucket, measurement, field })),
  };
}

Writing measurements with line protocol

InfluxDB's line protocol is a compact text format for writing time series data. Each line represents one data point: measurement,tag1=val1,tag2=val2 field1=1.0,field2="str" timestamp. Tag values and measurement names with special characters (spaces, commas, equals signs) must be escaped with a backslash. Field string values are quoted with double quotes; field numeric values are unquoted. Integer field values require a trailing i suffix: count=42i.

The timestamp is optional — if omitted, InfluxDB uses server time. When providing timestamps, match the precision parameter sent in the write request URL: precision=ns expects nanosecond integers, precision=ms expects millisecond integers. Mismatching precision is a silent error: writing a millisecond timestamp with precision=ns results in a data point timestamped at approximately 1970-01-01T00:00:00.001Z. Always explicitly specify precision and derive it from Date.now() with the correct multiplier.

function toLineProtocol(params) {
  const { measurement, tags, fields, timestamp } = params;

  // Escape special characters in measurement name and tag values
  const escapeName = (s) => String(s).replace(/[, =]/g, '\\$&');
  const escapeTagValue = (s) => String(s).replace(/[, =]/g, '\\$&');

  // Build tag set (sorted for consistent series key)
  const tagStr = Object.entries(tags ?? {})
    .sort(([a], [b]) => a.localeCompare(b))
    .map(([k, v]) => `${escapeName(k)}=${escapeTagValue(v)}`)
    .join(',');

  // Build field set — strings need quotes, integers need 'i' suffix
  const fieldStr = Object.entries(fields)
    .map(([k, v]) => {
      if (typeof v === 'string') return `${escapeName(k)}="${v.replace(/["\\]/g, '\\$&')}"`;
      if (Number.isInteger(v)) return `${escapeName(k)}=${v}i`;
      return `${escapeName(k)}=${v}`;
    })
    .join(',');

  if (!fieldStr) throw new Error('At least one field is required in line protocol');

  // Timestamp in nanoseconds (multiply ms by 1e6)
  const ts = timestamp ? (timestamp instanceof Date
    ? (BigInt(timestamp.getTime()) * 1_000_000n).toString()
    : String(timestamp)
  ) : '';

  const tagPart = tagStr ? `,${tagStr}` : '';
  return `${escapeName(measurement)}${tagPart} ${fieldStr}${ts ? ` ${ts}` : ''}`;
}

// Write MCP tool execution metrics to InfluxDB
async function writeMCPMetrics(influx, bucket, toolName, metrics) {
  const lines = metrics.map((m) =>
    toLineProtocol({
      measurement: 'mcp_tool_execution',
      tags: {
        tool: toolName,
        status: m.status, // 'success' | 'error' | 'timeout'
        env: m.env ?? 'prod',
      },
      fields: {
        duration_ms: m.durationMs,
        result_count: m.resultCount ?? 0,
        error_message: m.errorMessage ?? '',
      },
      timestamp: m.timestamp ?? new Date(),
    })
  );

  await influx.write(bucket, lines, 'ns');
}

// Batch write for high-throughput ingestion (respect the 50MB write limit)
async function batchWrite(influx, bucket, lines, batchSize = 5000) {
  for (let i = 0; i < lines.length; i += batchSize) {
    const batch = lines.slice(i, i + batchSize);
    await influx.write(bucket, batch);
    // Brief pause between batches to avoid rate limiting
    if (i + batchSize < lines.length) {
      await new Promise((r) => setTimeout(r, 100));
    }
  }
}

Bucket and organization management

InfluxDB v2 organizes data into buckets, each with a configurable retention period. When a bucket's retention period is exceeded, InfluxDB automatically deletes data older than the retention boundary — this is called the retention enforcement cycle. MCP tools that need to store historical data must either use a bucket with a long or infinite retention period, or implement their own archival workflow before data ages out.

Bucket IDs (hex strings) are the stable identifier for API requests; bucket names can change. For MCP tools that reference specific buckets, store both the ID and the name, and fall back to looking up the ID by name if needed. The GET /api/v2/buckets?name=<name>&org=<orgId> endpoint returns a single bucket by name — use this for name-to-ID resolution rather than listing all buckets and filtering in the client.

async function manageBuckets(influx) {
  // List all buckets in the org
  async function listBuckets() {
    const result = await influx.getRequest(
      `/api/v2/buckets?orgID=${influx.orgId}&limit=100`
    );
    return result.buckets.map((b) => ({
      id: b.id,
      name: b.name,
      orgID: b.orgID,
      retentionRules: b.retentionRules,
      // retentionRules[0].everySeconds === 0 means infinite retention
      retentionPeriodSeconds: b.retentionRules?.[0]?.everySeconds ?? 0,
      type: b.type, // 'user' (normal) or 'system' (internal InfluxDB buckets like _monitoring)
      createdAt: b.createdAt,
    }));
  }

  // Resolve bucket name to ID
  async function getBucketByName(name) {
    const result = await influx.getRequest(
      `/api/v2/buckets?orgID=${influx.orgId}&name=${encodeURIComponent(name)}`
    );
    if (!result.buckets?.length) throw new Error(`Bucket '${name}' not found`);
    return result.buckets[0];
  }

  // Check bucket write status via a test write
  async function checkBucketWritable(bucketName) {
    try {
      const testLine = toLineProtocol({
        measurement: '_health_probe',
        tags: { source: 'mcp_health' },
        fields: { test: 1 },
      });
      await influx.write(bucketName, [testLine], 'ns');
      return { writable: true };
    } catch (err) {
      return { writable: false, error: err.message };
    }
  }

  return { listBuckets, getBucketByName, checkBucketWritable };
}

// InfluxDB health check — use this as the AliveMCP probe
async function checkInfluxHealth(baseUrl, token) {
  const response = await fetch(`${baseUrl}/health`, {
    headers: { Authorization: `Token ${token}` },
  });
  const body = await response.json();
  // { name, message, checks: [], status: 'pass' | 'fail', version, commit }
  return {
    healthy: body.status === 'pass',
    status: body.status,
    version: body.version,
    message: body.message,
    checks: body.checks, // Sub-check results for clustered InfluxDB
  };
}

AliveMCP integration for InfluxDB monitoring

InfluxDB's /health endpoint returns a structured JSON response with status pass or fail. For clustered InfluxDB (Enterprise or InfluxDB Cloud Dedicated), the checks array includes sub-check results for each cluster component. Register this endpoint directly with AliveMCP — it does not require authentication in most deployments and is designed as a machine-readable health probe.

Beyond the health endpoint, instrument your MCP tools that write to InfluxDB to track write latency and failure rate. Write failures often manifest as silent data gaps rather than errors — if the write request is lost in transit or the InfluxDB instance is temporarily full, the data point is simply missing from the time series. Use AliveMCP to probe a canary write-then-read workflow: write a known measurement, then query it back within a tolerance window, and alert if the round-trip fails.

// Annotated CSV parser for InfluxDB query results
function parseAnnotatedCSV(csv) {
  const rows = [];
  const lines = csv.split('\n');

  let columnHeaders = null;
  let dataTypes = null;

  for (const line of lines) {
    if (!line.trim() || line.startsWith('#group') || line.startsWith('#default')) continue;

    if (line.startsWith('#datatype')) {
      dataTypes = line.slice(1).split(',').map((s) => s.trim());
      continue;
    }

    const cells = line.split(',');

    if (!columnHeaders) {
      columnHeaders = cells.map((c) => c.trim());
      continue;
    }

    // Skip result/table metadata rows
    if (cells[0] === '' || cells[0] === ',') continue;

    const row = {};
    columnHeaders.forEach((col, i) => {
      const raw = cells[i]?.trim() ?? '';
      const dtype = dataTypes?.[i];
      if (dtype === 'double' || dtype === 'float') row[col] = parseFloat(raw);
      else if (dtype === 'long' || dtype === 'unsignedLong') row[col] = parseInt(raw, 10);
      else if (dtype === 'boolean') row[col] = raw === 'true';
      else row[col] = raw;
    });

    // Only include rows with actual measurement data
    if (row._time || row._value !== undefined) rows.push(row);
  }

  return rows;
}

Related guides

• MCP tools for Prometheus Alertmanager — alerts, silences, and inhibition rules
• MCP tools for Elasticsearch — Query DSL, cluster health, and index management
• MCP tools for Dynatrace — metrics API v2, problems, and entity queries
• Metrics collection for MCP servers — counters, gauges, histograms, and cardinality
• MCP server observability — metrics, logs, and traces for production MCP tools
• MCP server health check patterns — composite endpoints and failure classification