Data Pipeline Integration for MCP Servers — CDC, Kafka, and keeping tool data fresh

Three data freshness strategies

Before choosing CDC, understand all three options and their tradeoffs:

Use query-on-demand when: the source database can absorb the load, tool call latency is acceptable, and data must always be current (financial balances, reservation status). Use scheduled refresh when: moderate staleness is acceptable (minutes), source DB load is a concern, and you want minimal operational overhead. Use CDC when: you need near-real-time freshness (<10 seconds), tool calls must be fast, or the source database cannot absorb high query concurrency from the MCP server.

PostgreSQL CDC with logical replication

PostgreSQL logical replication streams committed changes from the WAL (write-ahead log) to a subscriber. The MCP server subscribes to changes in the tables it cares about, maintaining a local SQLite or PostgreSQL materialized view.

Setting up logical replication

-- On the source PostgreSQL server (requires wal_level = logical in postgresql.conf)
-- Create a publication for the tables the MCP server needs
CREATE PUBLICATION mcp_server_feed
  FOR TABLE products, inventory_levels, deployments;

-- The MCP server subscribes and creates a logical replication slot

// MCP server: subscribe to logical replication changes
// Using pg-logical-replication npm package
import { LogicalReplicationService, PgoutputPlugin } from 'pg-logical-replication';

const service = new LogicalReplicationService({
  host: process.env.PGHOST_SOURCE,
  database: process.env.PGDATABASE_SOURCE,
  user: process.env.PGREPLICATION_USER,
  password: process.env.PGREPLICATION_PASSWORD,
});

const plugin = new PgoutputPlugin({
  protoVersion: 1,
  publicationNames: ['mcp_server_feed'],
});

// Local materialized store (SQLite or local PostgreSQL)
const localDb = new Database('./local_mirror.db');

service.on('data', async (lsn, log) => {
  if (log.tag === 'insert' || log.tag === 'update') {
    // Upsert into local mirror
    await upsertRecord(localDb, log.relation.name, log.new);
    await recordFreshness(log.relation.name);
  }
  if (log.tag === 'delete') {
    await deleteRecord(localDb, log.relation.name, log.key);
  }
});

// Track freshness per table
const tableFreshness: Map = new Map();

async function recordFreshness(tableName: string) {
  tableFreshness.set(tableName, new Date());
  await localDb.run(
    'INSERT OR REPLACE INTO _pipeline_health (table_name, last_updated_at) VALUES (?, ?)',
    [tableName, new Date().toISOString()]
  );
}

// Start replication from last confirmed LSN (persisted across restarts)
const lastLsn = await getLastConfirmedLsn(localDb);
await service.subscribe(plugin, 'mcp_server_slot', lastLsn);

Initial snapshot before streaming

A logical replication slot only streams changes that occur after the slot was created. Before starting to stream, take a full snapshot of the source tables to populate the local mirror:

// Startup: snapshot + stream pattern
async function initializeMirror() {
  const slotExists = await checkSlotExists();

  if (!slotExists) {
    // Take consistent snapshot while creating slot
    const snapshotName = await createReplicationSlotWithSnapshot();
    await bulkLoadFromSnapshot(snapshotName);
    console.log('Initial snapshot loaded');
  } else {
    console.log('Resuming from existing slot');
  }

  // Start streaming from last confirmed LSN
  await startLogicalReplication();
}

Kafka consumer integration with KafkaJS

For organizations where source data already flows through Kafka (via Debezium CDC, application-level event publishing, or ETL pipelines), the MCP server consumes from the relevant Kafka topics and maintains its local view.

import { Kafka, logLevel } from 'kafkajs';

const kafka = new Kafka({
  clientId: `mcp-server-${process.env.HOSTNAME}`,
  brokers: process.env.KAFKA_BROKERS!.split(','),
  logLevel: logLevel.WARN,
});

const consumer = kafka.consumer({
  groupId: 'mcp-server-inventory',   // one group per MCP server service
  sessionTimeout: 30_000,
  heartbeatInterval: 3_000,
});

await consumer.connect();
await consumer.subscribe({
  topics: ['inventory.products', 'inventory.stock_levels'],
  fromBeginning: false,   // start from latest — snapshot handles historical state
});

let lastMessageAt: Date | null = null;
const consumerLag: Map = new Map();

await consumer.run({
  eachMessage: async ({ topic, partition, message }) => {
    const event = JSON.parse(message.value!.toString());
    lastMessageAt = new Date();

    if (topic === 'inventory.products') {
      await upsertProduct(localDb, event);
    } else if (topic === 'inventory.stock_levels') {
      await upsertStockLevel(localDb, event);
    }

    // KafkaJS auto-commits offset after eachMessage resolves
    // Messages are processed at-least-once — handlers must be idempotent
  },
});

Consumer lag monitoring

Consumer lag (how many messages are behind the latest offset) is the primary freshness metric for Kafka-based pipelines. High lag means the MCP server's local view is stale:

// Fetch consumer lag from Kafka admin API
const admin = kafka.admin();
await admin.connect();

async function fetchConsumerLag(): Promise> {
  const offsets = await admin.fetchOffsets({ groupId: 'mcp-server-inventory', topics: ['inventory.products', 'inventory.stock_levels'] });
  const topicOffsets = await admin.fetchTopicOffsets('inventory.products');

  const lag: Map = new Map();
  for (const { topic, partitions } of offsets) {
    for (const { partition, offset } of partitions) {
      const latest = topicOffsets.find(t => t.partition === partition);
      if (latest) {
        lag.set(`${topic}:${partition}`, parseInt(latest.offset) - parseInt(offset));
      }
    }
  }
  return lag;
}

// Check lag every 30 seconds
setInterval(async () => {
  const lag = await fetchConsumerLag();
  for (const [key, messages] of lag) {
    consumerLag.set(key, messages);
    if (messages > 1000) {
      console.log(JSON.stringify({ level: 'warn', event: 'high_consumer_lag', key, messages }));
    }
  }
}, 30_000);

Debezium CDC → Kafka → MCP server

Debezium is an open-source CDC platform that captures row-level changes from database transaction logs and publishes them to Kafka topics. For PostgreSQL, Debezium uses logical replication (the same mechanism as direct CDC) but adds a managed connector, schema registry integration, and structured change events.

The MCP server is a Kafka consumer — it does not interact with Debezium directly:

Debezium change events have a structured envelope:

// Debezium change event structure (PostgreSQL connector)
interface DebeziumEvent {
  payload: {
    op: 'c' | 'u' | 'd' | 'r';  // create, update, delete, read (snapshot)
    before: Record | null;
    after: Record | null;
    source: {
      table: string;
      db: string;
      ts_ms: number;  // source DB commit timestamp — use for freshness tracking
    };
    ts_ms: number;    // Debezium processing timestamp
  };
}

// Process Debezium event in MCP server consumer
async function processDebeziumEvent(event: DebeziumEvent) {
  const { op, after, before, source } = event.payload;

  // Track source DB timestamp for accurate freshness measurement
  const sourceTs = new Date(source.ts_ms);
  const pipelineLag = Date.now() - source.ts_ms;  // end-to-end CDC lag in ms

  if (op === 'c' || op === 'u' || op === 'r') {
    await upsertRecord(localDb, source.table, after!);
  } else if (op === 'd') {
    await deleteRecord(localDb, source.table, before!);
  }

  // Record freshness using source DB timestamp (not processing timestamp)
  tableFreshness.set(source.table, sourceTs);
}

Data freshness monitoring with AliveMCP

A protocol probe confirms the MCP server accepts connections. It does not confirm the local materialized view is current. Data freshness must be monitored separately, via the health endpoint.

Freshness check in /health

// Data freshness check in /health endpoint
app.get('/health', async (req, res) => {
  const checks: Record = {};
  let status = 'ok';

  // Check freshness per table
  const FRESHNESS_THRESHOLDS: Record = {
    products: 300,           // 5 minutes acceptable staleness
    inventory_levels: 60,    // 1 minute — agents act on this data
    deployments: 30,         // 30 seconds — near-real-time required
  };

  for (const [tableName, thresholdSeconds] of Object.entries(FRESHNESS_THRESHOLDS)) {
    const lastUpdated = tableFreshness.get(tableName);
    if (!lastUpdated) {
      checks[`${tableName}_freshness`] = 'never_updated';
      status = 'degraded';
      continue;
    }

    const ageSec = (Date.now() - lastUpdated.getTime()) / 1000;
    checks[`${tableName}_freshness`] = {
      last_updated_at: lastUpdated.toISOString(),
      age_seconds: Math.round(ageSec),
      status: ageSec < thresholdSeconds ? 'fresh' : 'stale',
    };

    if (ageSec > thresholdSeconds) {
      status = 'degraded';
    }
  }

  // Also check consumer lag if using Kafka
  let totalLag = 0;
  for (const lag of consumerLag.values()) totalLag += lag;
  checks.consumer_lag_total = totalLag;
  if (totalLag > 5000) {
    checks.consumer_lag_status = 'high';
    status = 'degraded';
  }

  res.status(status === 'ok' ? 200 : 503).json({ status, checks });
});

Point AliveMCP's custom health URL at this endpoint. AliveMCP probes every 60 seconds. When the CDC pipeline stalls (Debezium connector stopped, Kafka consumer group rebalancing stuck, PostgreSQL replication slot lagging), the freshness check exceeds the threshold and AliveMCP alerts. This catches data-quality failures that produce wrong answers without producing errors — the most dangerous class of failure for AI agent workflows.

Handling pipeline failures and backfills

Pipelines fail. Debezium connectors restart. Kafka brokers rebalance. The MCP server must handle these failures gracefully without serving data from an unboundedly-stale view.

Circuit breaker for stale data

// Tool handler checks data freshness before serving results
const MAX_STALENESS_SECONDS = 120;

async function checkFreshness(tableName: string): Promise {
  const lastUpdated = tableFreshness.get(tableName);
  if (!lastUpdated) {
    throw new Error(`data_not_yet_populated: ${tableName}`);
  }
  const ageSec = (Date.now() - lastUpdated.getTime()) / 1000;
  if (ageSec > MAX_STALENESS_SECONDS) {
    throw new Error(`data_stale: ${tableName} last updated ${Math.round(ageSec)}s ago`);
  }
}

server.tool('search_products', { query: z.string() }, async (args) => {
  try {
    await checkFreshness('products');
  } catch (err) {
    // Return error response instead of stale data
    return {
      content: [{
        type: 'text',
        text: JSON.stringify({ error: err.message, suggestion: 'try again in 30s' }),
      }],
      isError: true,
    };
  }

  const results = await localDb.all(
    'SELECT * FROM products WHERE name LIKE ? LIMIT 20',
    [`%${args.query}%`]
  );
  return { content: [{ type: 'text', text: JSON.stringify(results) }] };
});

Returning isError: true on data staleness is better than returning stale data silently. The AI agent receives a clear signal that the tool is temporarily unavailable and can retry later or use a fallback. Silently returning stale data causes the agent to make decisions based on incorrect information without knowing it.

Automatic backfill on pipeline gap detection

// Trigger a full table refresh when pipeline gap exceeds threshold
const BACKFILL_THRESHOLD_SECONDS = 300;  // 5 minutes without update → backfill

setInterval(async () => {
  for (const [tableName, lastUpdated] of tableFreshness) {
    const ageSec = (Date.now() - lastUpdated.getTime()) / 1000;
    if (ageSec > BACKFILL_THRESHOLD_SECONDS) {
      console.log(JSON.stringify({
        level: 'warn',
        event: 'pipeline_gap_detected',
        table: tableName,
        age_seconds: ageSec,
        action: 'triggering_backfill',
      }));
      // Fall back to query-on-demand until pipeline recovers
      await backfillFromSource(tableName);
    }
  }
}, 60_000);

Frequently asked questions

Further reading

• Event-Driven Architecture for MCP Servers — pub/sub, webhooks, and real-time state
• Read Replica Routing for MCP Servers — write splitting and lag detection
• SQLite for MCP Servers — local storage patterns and WAL configuration
• MCP Server Health Checks — including data freshness in readiness probes
• Graceful Degradation for MCP Servers — partial responses when dependencies fail