AliveMCP

Guide · Data Serialization Integration

MCP Server Apache Arrow — columnar in-memory format, IPC streaming, DuckDB zero-copy

Three Apache Arrow behaviours trip up MCP server authors: Arrow buffers live off the JavaScript heap — a 500 MB Arrow table does not show up in process.memoryUsage().heapUsed because Arrow allocates its column buffers in WebAssembly linear memory or native allocations; if your MCP server loads large Arrow datasets in a loop, memory grows invisibly until the process is OOM-killed; always call table.delete() when using @apache-arrow/ts in environments that expose a GC handle, or scope table reads inside a bounded function to release references; Arrow's BigInt64 columns decode as bigint in JavaScript, which JSON.stringify cannot serialise — any Int64, Uint64, Timestamp[ns], or Duration column will throw TypeError: Do not know how to serialize a BigInt unless you convert each field with value.toString() or register a JSON.stringify replacer; and Arrow IPC has two wire formats — the stream format (starts with a continuation marker 0xFFFFFFFF) and the file format (starts with the magic bytes ARROW1); passing a file-format buffer to tableFromIPC works, but passing a stream-format buffer to a parser that expects file format will silently read zero batches.

TL;DR

Use apache-arrow (npm) to read Arrow IPC buffers from DuckDB, Python FastAPI, or Arrow Flight endpoints. Convert Int64 and Timestamp[ns] columns to strings before JSON.stringify. Use table.toArray() to get row objects, but convert to plain arrays before returning from tool handlers — Arrow row proxies are not JSON.stringify-safe. Health probe: perform a round-trip tableFromIPC(tableToIPC(sampleTable)) at startup.

Reading Arrow IPC from DuckDB in MCP tool handlers

DuckDB's Node.js bindings can return query results as Arrow IPC buffers via connection.arrowIPCStream(), which is significantly faster than row-by-row iteration for large result sets — DuckDB's columnar storage maps directly to Arrow's columnar format with zero data copying. The returned IPC stream can be piped into apache-arrow's RecordBatchReader for incremental consumption without loading the entire result into memory at once.

import { tableFromIPC, RecordBatchReader, Table, Int64, TimestampNanosecond } from 'apache-arrow';
import Database from 'duckdb-async';

const db = await Database.create(process.env.DUCKDB_PATH ?? ':memory:');

// Helper: safe JSON serialiser for Arrow decoded values
function arrowValueToJson(value: unknown): unknown {
  if (typeof value === 'bigint') return value.toString();
  if (value instanceof Date)     return value.toISOString();
  if (value instanceof Int8Array || value instanceof Uint8Array) {
    return Buffer.from(value).toString('base64');
  }
  return value;
}

function tableToJsonRows(table: Table): Record<string, unknown>[] {
  const schema = table.schema;
  const rows: Record<string, unknown>[] = [];

  for (let i = 0; i < table.numRows; i++) {
    const row: Record<string, unknown> = {};
    for (const field of schema.fields) {
      const col = table.getChild(field.name);
      const raw = col?.get(i);

      if (raw === null || raw === undefined) {
        row[field.name] = null;
        continue;
      }

      // Timestamp[ns] arrives as bigint nanoseconds since epoch
      if (field.type instanceof TimestampNanosecond || field.type.toString().includes('Timestamp')) {
        row[field.name] = new Date(Number(raw) / 1_000_000).toISOString();
      } else if (typeof raw === 'bigint') {
        row[field.name] = raw.toString();
      } else {
        row[field.name] = raw;
      }
    }
    rows.push(row);
  }
  return rows;
}

// MCP tool: run an analytics query via DuckDB + Arrow IPC (fast path)
server.tool(
  'query_analytics',
  {
    sql:      z.string().max(4000),
    max_rows: z.number().int().min(1).max(10000).default(1000),
  },
  async ({ sql, max_rows }) => {
    // Safety: only allow SELECT statements
    const trimmed = sql.trim().toLowerCase();
    if (!trimmed.startsWith('select') && !trimmed.startsWith('with')) {
      throw new Error('Only SELECT queries are allowed');
    }

    const conn = await db.connect();
    try {
      // arrowIPCStream returns an AsyncIterable of Uint8Array IPC buffers
      const stream = await conn.arrowIPCStream(
        `SELECT * FROM (${sql}) LIMIT ${max_rows}`
      );

      const batches: Uint8Array[] = [];
      for await (const chunk of stream) {
        batches.push(chunk);
      }

      // Concatenate IPC buffers and parse into Arrow Table
      const totalLength = batches.reduce((sum, b) => sum + b.byteLength, 0);
      const combined = new Uint8Array(totalLength);
      let offset = 0;
      for (const batch of batches) {
        combined.set(batch, offset);
        offset += batch.byteLength;
      }

      const table = tableFromIPC(combined);
      const rows = tableToJsonRows(table);

      return {
        content: [{
          type: 'text',
          text: JSON.stringify({
            row_count: rows.length,
            schema: table.schema.fields.map(f => ({ name: f.name, type: f.type.toString() })),
            rows,
          }),
        }],
      };
    } finally {
      await conn.close();
    }
  }
);

Consuming Arrow IPC streams from Python FastAPI backends

Python data services built with FastAPI often expose Arrow IPC via a streaming HTTP response — the endpoint returns application/vnd.apache.arrow.stream content-type and streams record batches as they become available. This pattern is common in ML serving pipelines where a Python model produces predictions as an Arrow batch. The Node.js MCP server can consume this stream incrementally without buffering the full response.

import { RecordBatchReader } from 'apache-arrow';

// Python backend returns Arrow IPC stream — consume incrementally
async function fetchArrowStream(url: string, params: Record<string, string> = {}): Promise<Record<string, unknown>[]> {
  const query = new URLSearchParams(params).toString();
  const resp = await fetch(`${url}?${query}`, {
    headers: { Accept: 'application/vnd.apache.arrow.stream' },
    signal: AbortSignal.timeout(30_000),
  });

  if (!resp.ok) {
    throw new Error(`Arrow stream endpoint returned ${resp.status}: ${await resp.text()}`);
  }

  const contentType = resp.headers.get('content-type') ?? '';
  if (!contentType.includes('arrow')) {
    throw new Error(`Expected Arrow IPC stream, got: ${contentType}`);
  }

  // ReadableStream → AsyncIterable<Uint8Array> for RecordBatchReader
  const reader = await RecordBatchReader.from(resp.body!);

  const allRows: Record<string, unknown>[] = [];
  for await (const batch of reader) {
    // batch is a RecordBatch — convert to row objects batch by batch
    const schema = batch.schema;
    for (let i = 0; i < batch.numRows; i++) {
      const row: Record<string, unknown> = {};
      for (const field of schema.fields) {
        const col = batch.getChildAt(schema.fields.indexOf(field));
        const raw = col?.get(i);
        // Convert bigint and Timestamps
        if (typeof raw === 'bigint') {
          row[field.name] = raw.toString();
        } else if (raw instanceof Date) {
          row[field.name] = raw.toISOString();
        } else {
          row[field.name] = raw ?? null;
        }
      }
      allRows.push(row);
    }
  }

  return allRows;
}

// MCP tool: call ML prediction endpoint and return results
server.tool(
  'predict_churn',
  { user_ids: z.array(z.string()).max(500) },
  async ({ user_ids }) => {
    const rows = await fetchArrowStream(
      `${process.env.ML_SERVICE_URL}/predict/churn`,
      { user_ids: user_ids.join(',') }
    );

    return {
      content: [{
        type: 'text',
        text: JSON.stringify({ predictions: rows, count: rows.length }),
      }],
    };
  }
);

Arrow schema inspection — understanding column types before conversion

Before converting an Arrow Table to JSON rows, inspecting the schema tells you which columns need special handling. Arrow has many numeric types (Int8 through Int64, Uint8 through Uint64, Float16, Float32, Float64) and temporal types (Date32, Date64, Timestamp[s], Timestamp[ms], Timestamp[us], Timestamp[ns]) that need different conversion logic. Only Int64, Uint64, and nanosecond timestamps arrive as bigint; millisecond timestamps arrive as number (JavaScript-safe integers up to ~285 million years).

import {
  Table, DataType, Type,
  Int64, Uint64,
  TimestampSecond, TimestampMillisecond, TimestampMicrosecond, TimestampNanosecond,
  Date32, Date64,
  Utf8, LargeUtf8,
  Binary, LargeBinary,
  List, FixedSizeList,
} from 'apache-arrow';

// Introspect Arrow schema and generate conversion rules
function buildConversionMap(table: Table): Map<string, 'bigint_to_string' | 'ts_ns' | 'ts_us' | 'ts_ms' | 'ts_s' | 'date32' | 'date64' | 'binary_to_base64' | 'pass'> {
  const map = new Map<string, string>();

  for (const field of table.schema.fields) {
    const t = field.type;
    if (t instanceof Int64 || t instanceof Uint64) {
      map.set(field.name, 'bigint_to_string');
    } else if (t instanceof TimestampNanosecond) {
      map.set(field.name, 'ts_ns');   // bigint ns since epoch
    } else if (t instanceof TimestampMicrosecond) {
      map.set(field.name, 'ts_us');   // bigint us since epoch
    } else if (t instanceof TimestampMillisecond) {
      map.set(field.name, 'ts_ms');   // number ms since epoch (safe)
    } else if (t instanceof TimestampSecond) {
      map.set(field.name, 'ts_s');    // number s since epoch
    } else if (t instanceof Date32 || t instanceof Date64) {
      map.set(field.name, t instanceof Date32 ? 'date32' : 'date64');
    } else if (t instanceof Binary || t instanceof LargeBinary) {
      map.set(field.name, 'binary_to_base64');
    } else {
      map.set(field.name, 'pass');
    }
  }

  return map as Map<string, 'bigint_to_string' | 'ts_ns' | 'ts_us' | 'ts_ms' | 'ts_s' | 'date32' | 'date64' | 'binary_to_base64' | 'pass'>;
}

function convertArrowValue(value: unknown, rule: string): unknown {
  if (value === null || value === undefined) return null;
  switch (rule) {
    case 'bigint_to_string': return (value as bigint).toString();
    case 'ts_ns':  return new Date(Number(value as bigint) / 1_000_000).toISOString();
    case 'ts_us':  return new Date(Number(value as bigint) / 1_000).toISOString();
    case 'ts_ms':  return new Date(value as number).toISOString();
    case 'ts_s':   return new Date((value as number) * 1000).toISOString();
    case 'date32': return new Date((value as number) * 86400000).toISOString().slice(0, 10);
    case 'date64': return new Date(value as number).toISOString().slice(0, 10);
    case 'binary_to_base64': return Buffer.from(value as Uint8Array).toString('base64');
    default: return value;
  }
}

function arrowTableToRows(table: Table): Record<string, unknown>[] {
  const conversionMap = buildConversionMap(table);
  const rows: Record<string, unknown>[] = [];

  for (let i = 0; i < table.numRows; i++) {
    const row: Record<string, unknown> = {};
    for (const field of table.schema.fields) {
      const col = table.getChild(field.name);
      row[field.name] = convertArrowValue(col?.get(i), conversionMap.get(field.name) ?? 'pass');
    }
    rows.push(row);
  }
  return rows;
}

Health probe — round-trip IPC verification at startup

Arrow IPC deserialisation failures are rare but silent when they occur — a truncated buffer or mismatched schema version produces an empty table rather than a thrown error in some code paths. Run a startup probe that encodes a known small table to IPC and decodes it back, verifying row count and a sample value match.

import { tableFromIPC, tableToIPC, makeTable, vectorFromArray, Float64, Utf8 as ArrowUtf8 } from 'apache-arrow';

async function verifyArrowIpc(): Promise<void> {
  // Build a tiny reference table
  const source = makeTable({
    id:    vectorFromArray([1, 2, 3], new Int32Array(0).constructor as unknown as Int32ArrayConstructor),
    name:  vectorFromArray(['alpha', 'beta', 'gamma']),
    score: vectorFromArray([0.1, 0.9, 0.5]),
  });

  // Encode to IPC stream format and decode back
  const ipcBuffer = tableToIPC(source, 'stream');
  const roundTripped = tableFromIPC(ipcBuffer);

  if (roundTripped.numRows !== 3) {
    throw new Error(`Arrow IPC round-trip failed: expected 3 rows, got ${roundTripped.numRows}`);
  }

  const firstId = roundTripped.getChild('id')?.get(0);
  if (firstId !== 1) {
    throw new Error(`Arrow IPC round-trip value mismatch: expected id[0]=1, got ${firstId}`);
  }

  console.log('Arrow IPC round-trip verification passed');
}

// Register as MCP resource for AliveMCP uptime monitoring
server.resource('arrow_health', 'arrow://health', async () => {
  try {
    await verifyArrowIpc();
    return {
      contents: [{ uri: 'arrow://health', text: JSON.stringify({ status: 'ok', ipc_roundtrip: true }) }],
    };
  } catch (err) {
    return {
      contents: [{ uri: 'arrow://health', text: JSON.stringify({ status: 'error', error: String(err) }) }],
    };
  }
});