AliveMCP

Guide · Data Serialization Integration

MCP Server Apache Avro — Schema Registry, union types, schema evolution for Kafka pipelines

Three Apache Avro behaviours trip up MCP server authors consuming Kafka topics: Avro Kafka messages carry a 5-byte magic prefix before the payload — the first byte is always 0x00 (magic byte) followed by a 4-byte big-endian schema ID; decoding the raw Kafka message bytes directly with an Avro parser will fail because the parser sees the magic prefix as schema content; strip the first 5 bytes and fetch the schema by ID from the Schema Registry before decoding; nullable fields in Avro require a union type, not a simple optional marker — writing { "name": "user_email", "type": "string" } makes the field required and non-null; to allow null you must write { "name": "user_email", "type": ["null", "string"], "default": null } with null first in the union (so the default is valid); and schema compatibility mode determines which evolution operations are legal — Confluent Schema Registry defaults to BACKWARD compatibility which allows adding optional fields and deleting fields with defaults, but not adding required fields or changing existing types.

TL;DR

Use @kafkajs/confluent-schema-registry to decode Kafka+Avro messages in MCP tool handlers — it handles the magic-byte prefix and caches schemas by ID automatically. Always define nullable fields as ["null", "type"] union with "default": null as the first union branch. Cache the schema registry client at module scope, not per-request. Health probe: GET /subjects on the Schema Registry and verify it returns a list.

Decoding Kafka+Avro messages — magic byte prefix and Schema Registry lookup

Every Kafka message encoded with Confluent-compatible Avro serialisers starts with a 5-byte framing header: 0x00 (magic byte signals Confluent wire format) followed by a 4-byte big-endian integer that is the schema ID registered in the Schema Registry. The actual Avro binary payload follows these 5 bytes. The @kafkajs/confluent-schema-registry library handles this framing automatically when you call registry.decode(), and it caches the schema after the first lookup so subsequent messages with the same ID skip the HTTP fetch.

import { SchemaRegistry } from '@kafkajs/confluent-schema-registry';
import { Kafka } from 'kafkajs';

// Schema Registry client — one instance at module scope, not per-request
const registry = new SchemaRegistry({
  host: process.env.SCHEMA_REGISTRY_URL!,
  auth: {
    username: process.env.SCHEMA_REGISTRY_API_KEY!,
    password: process.env.SCHEMA_REGISTRY_API_SECRET!,
  },
  // Cache decoded schemas in memory — avoids repeated HTTP fetches
  // Default cache: 1000 schemas (enough for most topics)
});

const kafka = new Kafka({
  clientId: 'mcp-server',
  brokers: process.env.KAFKA_BROKERS!.split(','),
  ssl: true,
  sasl: {
    mechanism: 'plain',
    username: process.env.KAFKA_API_KEY!,
    password: process.env.KAFKA_API_SECRET!,
  },
});

// MCP tool: fetch latest N events from a Kafka topic
server.tool(
  'get_recent_events',
  { topic: z.string(), max_events: z.number().int().min(1).max(100).default(20) },
  async ({ topic, max_events }) => {
    const consumer = kafka.consumer({ groupId: `mcp-reader-${Date.now()}` });
    await consumer.connect();
    await consumer.subscribe({ topic, fromBeginning: false });

    const events: unknown[] = [];
    await consumer.run({
      eachMessage: async ({ message }) => {
        if (!message.value) return;
        try {
          // registry.decode() strips the 5-byte prefix, fetches schema by ID,
          // and decodes the Avro binary into a plain JS object
          const decoded = await registry.decode(message.value);
          events.push({
            offset: message.offset,
            timestamp: message.timestamp,
            key: message.key?.toString(),
            value: decoded,
          });
        } catch (err) {
          // Non-Avro messages (e.g. control records) silently skip
          console.warn('Failed to decode Avro message:', err);
        }
        if (events.length >= max_events) {
          await consumer.stop();
        }
      },
    });

    await consumer.disconnect();
    return {
      content: [{ type: 'text', text: JSON.stringify(events, null, 2) }],
    };
  }
);

// Producing Avro-encoded messages from an MCP tool
server.tool(
  'publish_event',
  { topic: z.string(), subject: z.string(), payload: z.record(z.unknown()) },
  async ({ topic, subject, payload }) => {
    const producer = kafka.producer();
    await producer.connect();

    // Fetch the latest schema version for the subject
    const { id } = await registry.getLatestSchemaId(subject);
    // Encode adds the 5-byte magic prefix with the schema ID automatically
    const encoded = await registry.encode(id, payload);

    await producer.send({
      topic,
      messages: [{ value: encoded }],
    });
    await producer.disconnect();
    return { content: [{ type: 'text', text: `Published 1 message to ${topic} with schema ID ${id}` }] };
  }
);

Union types — the correct way to model nullable fields

Avro has no concept of "optional" as a field modifier. Every field is required unless you explicitly use a union that includes null. The convention for nullable fields is a union of ["null", "type"] with a default of null — the default must always match the first branch of the union. If you place the non-null type first (["string", "null"]) the default must be a string, making the field effectively required in backward-compatible consumers.

// Correct Avro schema for a user event with nullable fields
const userEventSchema = {
  type: 'record',
  name: 'UserEvent',
  namespace: 'com.example.events',
  fields: [
    // Required field — no union, no default
    { name: 'event_id', type: 'string' },

    // Required enum — must have a zero-value by convention
    {
      name: 'event_type',
      type: {
        type: 'enum',
        name: 'UserEventType',
        symbols: ['CREATED', 'UPDATED', 'DELETED', 'SUSPENDED'],
      },
    },

    // Nullable string — union with null FIRST so default: null is valid
    { name: 'display_name', type: ['null', 'string'], default: null },

    // Nullable record — nested message that may not exist
    {
      name: 'address',
      type: [
        'null',
        {
          type: 'record',
          name: 'Address',
          fields: [
            { name: 'street',  type: 'string' },
            { name: 'city',    type: 'string' },
            { name: 'country', type: 'string' },
            // Nullable optional field inside nested record
            { name: 'postal_code', type: ['null', 'string'], default: null },
          ],
        },
      ],
      default: null,
    },

    // Logical type: timestamp-millis maps to a JavaScript Date
    {
      name: 'created_at',
      type: { type: 'long', logicalType: 'timestamp-millis' },
    },

    // Array field — use empty array as default for optional lists
    {
      name: 'tags',
      type: { type: 'array', items: 'string' },
      default: [],
    },

    // Map field — string keys, any-type values encoded as string
    {
      name: 'metadata',
      type: { type: 'map', values: 'string' },
      default: {},
    },
  ],
};

// When decoded, union fields arrive wrapped: { string: 'John' } not just 'John'
// @kafkajs/confluent-schema-registry unwraps unions automatically by default
// If using avsc directly you must unwrap manually:
function unwrapUnion(value: unknown): unknown {
  if (value === null) return null;
  if (typeof value === 'object' && value !== null) {
    const keys = Object.keys(value as object);
    if (keys.length === 1 && typeof (value as Record<string, unknown>)[keys[0]] !== 'undefined') {
      // Single-key object is an Avro union wrapper — unwrap to the inner value
      return (value as Record<string, unknown>)[keys[0]];
    }
  }
  return value;
}

Schema evolution compatibility modes

Confluent Schema Registry enforces compatibility rules per subject on every POST /subjects/{subject}/versions call. The default is BACKWARD, meaning new consumers can read messages written by old producers. Understanding what each mode allows is essential before evolving a schema that MCP tool handlers depend on — a 409 Conflict from the registry when registering a new schema version means the change violates the configured compatibility mode.

// Schema compatibility modes and what they allow:
//
// BACKWARD (default): new schema can read data written by old schema
//   ✅ Add optional field (union with null, default null)
//   ✅ Delete field that had a default value
//   ❌ Add required field (no default)
//   ❌ Rename a field (field order and name both matter in Avro)
//   ❌ Change a field's type (even int → long is not always backward compat)
//
// FORWARD: old schema can read data written by new schema
//   ✅ Delete optional field (old readers use default when field missing)
//   ✅ Add required field (old readers ignore unknown fields)
//   ❌ Delete required field (old readers expect it, new writers don't send it)
//
// FULL: both backward AND forward compatible
//   ✅ Add optional field with default
//   ✅ Delete optional field with default
//   ❌ Add/remove required fields
//   ❌ Change types
//
// NONE: no compatibility check — use only in dev/test environments

// Check current compatibility setting for a subject
async function getCompatibility(subject: string): Promise<string> {
  const resp = await fetch(
    `${process.env.SCHEMA_REGISTRY_URL}/config/${subject}`,
    { headers: { Authorization: `Basic ${btoa(`${process.env.SCHEMA_REGISTRY_API_KEY}:${process.env.SCHEMA_REGISTRY_API_SECRET}`)}` } }
  );
  if (resp.status === 404) {
    // Subject has no override — uses global default (typically BACKWARD)
    const global = await fetch(`${process.env.SCHEMA_REGISTRY_URL}/config`, { headers: { /* same auth */ } });
    return ((await global.json()) as { compatibilityLevel: string }).compatibilityLevel;
  }
  return ((await resp.json()) as { compatibilityLevel: string }).compatibilityLevel;
}

// Test a candidate schema before registering (dry-run compatibility check)
async function testSchemaCompatibility(subject: string, candidateSchema: object): Promise<boolean> {
  const resp = await fetch(
    `${process.env.SCHEMA_REGISTRY_URL}/compatibility/subjects/${subject}/versions/latest`,
    {
      method: 'POST',
      headers: {
        'Content-Type': 'application/vnd.schemaregistry.v1+json',
        Authorization: `Basic ${btoa(`${process.env.SCHEMA_REGISTRY_API_KEY}:${process.env.SCHEMA_REGISTRY_API_SECRET}`)}`,
      },
      body: JSON.stringify({ schema: JSON.stringify(candidateSchema) }),
    }
  );
  const body = await resp.json() as { is_compatible: boolean };
  return body.is_compatible;
}

Health probe — Schema Registry connectivity check

The Schema Registry exposes a GET /subjects endpoint that returns a JSON array of subject names. A 200 response with a valid array confirms both connectivity and that the registry process is healthy. Cache the probe result for 30 seconds to avoid flooding the registry with health checks from all concurrent tool calls in a busy MCP session.

let registryHealthCache: { ok: boolean; checkedAt: number; subjects: number } | null = null;

async function checkSchemaRegistryHealth(): Promise<{ ok: boolean; subjects: number; latency_ms: number }> {
  const now = Date.now();
  if (registryHealthCache && now - registryHealthCache.checkedAt < 30_000) {
    return { ok: registryHealthCache.ok, subjects: registryHealthCache.subjects, latency_ms: 0 };
  }

  const start = Date.now();
  try {
    const resp = await fetch(`${process.env.SCHEMA_REGISTRY_URL}/subjects`, {
      signal: AbortSignal.timeout(5000),
      headers: {
        Authorization: `Basic ${btoa(`${process.env.SCHEMA_REGISTRY_API_KEY}:${process.env.SCHEMA_REGISTRY_API_SECRET}`)}`,
      },
    });
    const latency_ms = Date.now() - start;

    if (!resp.ok) {
      registryHealthCache = { ok: false, checkedAt: now, subjects: 0 };
      return { ok: false, subjects: 0, latency_ms };
    }

    const subjects = (await resp.json()) as string[];
    registryHealthCache = { ok: true, checkedAt: now, subjects: subjects.length };
    return { ok: true, subjects: subjects.length, latency_ms };
  } catch {
    registryHealthCache = { ok: false, checkedAt: now, subjects: 0 };
    return { ok: false, subjects: 0, latency_ms: Date.now() - start };
  }
}

// Register as MCP resource for AliveMCP uptime monitoring
server.resource('avro_health', 'avro://health', async () => {
  const health = await checkSchemaRegistryHealth();
  return {
    contents: [{
      uri: 'avro://health',
      text: JSON.stringify(health),
    }],
  };
});