AliveMCP

Guide · Event-Driven & Async Patterns

MCP Server CloudEvents — CNCF standard, specversion, SDK, event routing

Three CloudEvents behaviours catch MCP server authors off-guard: specversion must be the string "1.0" — older "0.3" events still circulate and fail validation against 1.0 schemas — CloudEvents 1.0 is the current stable spec; version 0.3 used a different attribute set (schemaurl replaced by dataschema, contenttype replaced by datacontenttype) and 0.3 events arriving at a 1.0 consumer will either fail validation or silently drop required attributes; always validate the specversion field before deserialising a received event and surface an explicit error rather than propagating a partially-parsed event; the type attribute is a reverse-DNS string by convention but this is not enforced by the spec, and omitting the convention breaks routing in brokers that use type prefix matching — EventBridge, Knative Eventing, and most event routers match event type values by prefix or exact string; using free-form names like "OrderCreated" instead of "com.example.orders.created" breaks prefix-based wildcard rules downstream; and binary content mode sends event attributes in HTTP headers, not in the body — most HTTP clients silently strip unknown headers on redirect — in binary content mode (the preferred HTTP binding for performance), event attributes like ce-id, ce-source, ce-type are sent as HTTP headers and only the data field goes in the body; proxies, load balancers, and CDNs that strip non-standard headers will deliver the body without its envelope, causing the receiver to see raw data with no event context.

TL;DR

Use the cloudevents npm package for type-safe event construction and parsing. Construct events with new CloudEvent({ specversion: '1.0', type, source, id, data }) — id must be unique per source (use crypto.randomUUID()). Emit as structured JSON with new HTTPBinding().toEvent() or emitBinary(). Parse received events with new HTTPBinding().toEvent(req) — it handles both binary and structured content modes automatically. Always validate specversion === '1.0' before processing. Use reverse-DNS type strings (e.g., com.example.order.created) to enable prefix routing in downstream brokers.

Creating and sending CloudEvents

The CloudEvents spec defines a minimal set of required attributes: specversion, id, source, and type. The id must be unique per source — two events from the same source with the same id are considered duplicates and can be deduplicated by brokers. The source is a URI that identifies the originating context (e.g., https://api.example.com/orders or /microservices/order-service). The type describes the event occurrence and conventionally uses reverse-DNS notation with a past-tense verb segment (e.g., com.example.order.created, com.example.user.password.reset).

import { CloudEvent, HTTP } from 'cloudevents';
import { z } from 'zod';
import crypto from 'crypto';

// MCP tool — publish a CloudEvent to an HTTP endpoint
server.tool(
  'publish_cloud_event',
  {
    endpoint:    z.string().url(),
    type:        z.string().min(1),  // e.g. 'com.example.order.created'
    source:      z.string().min(1),  // e.g. 'https://api.example.com/orders'
    data:        z.record(z.unknown()),
    subject:     z.string().optional(),     // e.g. the order ID — narrows the event
    dataschema:  z.string().url().optional(), // URI to JSON Schema for data
  },
  async ({ endpoint, type, source, data, subject, dataschema }) => {
    const event = new CloudEvent({
      specversion: '1.0',
      id:          crypto.randomUUID(),
      type,
      source,
      subject,
      dataschema,
      datacontenttype: 'application/json',
      data,
      time: new Date(),
    });

    // Structured content mode — single JSON body containing both attributes and data
    // Prefer structured mode when passing through HTTP proxies that strip headers
    const message = HTTP.structured(event);

    const resp = await fetch(endpoint, {
      method:  'POST',
      headers: message.headers as Record<string, string>,
      body:    JSON.stringify(message.body),
    });

    if (!resp.ok) {
      throw new Error(`CloudEvent delivery failed: HTTP ${resp.status} ${resp.statusText}`);
    }

    return {
      content: [{
        type: 'text',
        text: JSON.stringify({
          delivered: true,
          event_id:  event.id,
          endpoint,
          type:      event.type,
          source:    event.source,
        }),
      }],
    };
  }
);

// MCP tool — publish via binary content mode (attributes in headers, data in body)
server.tool(
  'publish_cloud_event_binary',
  {
    endpoint: z.string().url(),
    type:     z.string().min(1),
    source:   z.string().min(1),
    data:     z.record(z.unknown()),
  },
  async ({ endpoint, type, source, data }) => {
    const event = new CloudEvent({
      specversion: '1.0',
      id:          crypto.randomUUID(),
      type,
      source,
      datacontenttype: 'application/json',
      data,
    });

    // Binary content mode — ce-* HTTP headers + JSON body
    const message = HTTP.binary(event);

    const resp = await fetch(endpoint, {
      method:  'POST',
      headers: message.headers as Record<string, string>,
      body:    JSON.stringify(message.body),
    });

    return {
      content: [{
        type: 'text',
        text: JSON.stringify({ delivered: resp.ok, status: resp.status, event_id: event.id }),
      }],
    };
  }
);

Receiving and validating CloudEvents

The cloudevents SDK's HTTP.toEvent() function parses both binary and structured content modes from an incoming HTTP request. It handles the content-mode detection automatically by inspecting the Content-Type header: structured mode uses application/cloudevents+json; binary mode uses the original datacontenttype with event attributes in ce-* headers. After parsing, always validate the specversion field and check that required attributes are present before processing.

import express from 'express';
import { HTTP, CloudEvent } from 'cloudevents';

const app = express();
app.use(express.raw({ type: '*/*' }));  // preserve raw body for cloudevents parsing

// CloudEvent receiver endpoint — called by brokers (Knative, EventBridge Pipes, etc.)
app.post('/events', async (req, res) => {
  let event: CloudEvent<unknown>;

  try {
    // HTTP.toEvent handles both binary and structured content modes
    event = HTTP.toEvent({
      headers: req.headers,
      body:    req.body,
    });
  } catch (err: any) {
    res.status(400).json({ error: 'invalid_cloud_event', message: err.message });
    return;
  }

  // Validate specversion
  if (event.specversion !== '1.0') {
    res.status(422).json({
      error:   'unsupported_specversion',
      message: `Expected specversion 1.0, got ${event.specversion}`,
    });
    return;
  }

  // Validate required attributes
  if (!event.type || !event.source || !event.id) {
    res.status(422).json({ error: 'missing_required_attributes' });
    return;
  }

  // Route by type prefix
  const routed = await routeEvent(event);
  if (!routed) {
    // Return 200 even for unknown types — the broker should not retry unknown types
    res.status(200).json({ status: 'ignored', type: event.type });
    return;
  }

  res.status(200).json({ status: 'processed', event_id: event.id });
});

async function routeEvent(event: CloudEvent<unknown>): Promise<boolean> {
  const type = event.type;

  if (type.startsWith('com.example.order.')) {
    await handleOrderEvent(event);
    return true;
  }
  if (type.startsWith('com.example.user.')) {
    await handleUserEvent(event);
    return true;
  }
  return false;
}

// MCP tool — parse and inspect a raw CloudEvent payload
server.tool(
  'parse_cloud_event',
  { raw_event: z.string().min(1) },
  async ({ raw_event }) => {
    let parsed: Record<string, unknown>;
    try {
      parsed = JSON.parse(raw_event);
    } catch {
      return { content: [{ type: 'text', text: JSON.stringify({ error: 'invalid_json' }) }] };
    }

    const issues: string[] = [];
    if (!parsed['specversion'])    issues.push('missing specversion');
    if (parsed['specversion'] !== '1.0') issues.push(`specversion '${parsed['specversion']}' is not 1.0`);
    if (!parsed['id'])             issues.push('missing id');
    if (!parsed['source'])         issues.push('missing source');
    if (!parsed['type'])           issues.push('missing type');

    const isReversDns = typeof parsed['type'] === 'string' &&
      /^[a-z][a-z0-9]*(\.[a-z][a-z0-9]*)+/.test(parsed['type'] as string);

    return {
      content: [{
        type: 'text',
        text: JSON.stringify({
          valid:         issues.length === 0,
          issues,
          specversion:   parsed['specversion'],
          id:            parsed['id'],
          source:        parsed['source'],
          type:          parsed['type'],
          type_follows_reverse_dns_convention: isReversDns,
          time:          parsed['time'],
          datacontenttype: parsed['datacontenttype'],
          subject:       parsed['subject'],
          dataschema:    parsed['dataschema'],
          has_data:      'data' in parsed,
        }, null, 2),
      }],
    };
  }
);

Event batching and CloudEvent schemas

CloudEvents supports batch delivery via the application/cloudevents-batch+json content type — a JSON array of CloudEvent objects. Batch mode reduces HTTP overhead when publishing many events to the same endpoint. Some brokers (Knative, Azure Event Grid) deliver batches to subscribers; your MCP server's receiver endpoint must handle both single-event and batch payloads by inspecting the Content-Type header before parsing.

// Publish a batch of CloudEvents
server.tool(
  'publish_event_batch',
  {
    endpoint: z.string().url(),
    events: z.array(z.object({
      type:    z.string().min(1),
      source:  z.string().min(1),
      data:    z.record(z.unknown()),
      subject: z.string().optional(),
    })).min(1).max(100),
  },
  async ({ endpoint, events }) => {
    const cloudEvents = events.map(e =>
      new CloudEvent({
        specversion:     '1.0',
        id:              crypto.randomUUID(),
        type:            e.type,
        source:          e.source,
        subject:         e.subject,
        datacontenttype: 'application/json',
        data:            e.data,
        time:            new Date(),
      })
    );

    // Batch message — Content-Type: application/cloudevents-batch+json
    const message = HTTP.batch(cloudEvents);

    const resp = await fetch(endpoint, {
      method:  'POST',
      headers: message.headers as Record<string, string>,
      body:    JSON.stringify(message.body),
    });

    return {
      content: [{
        type: 'text',
        text: JSON.stringify({
          delivered:   resp.ok,
          status:      resp.status,
          event_count: cloudEvents.length,
          ids:         cloudEvents.map(e => e.id),
        }),
      }],
    };
  }
);

// Receiver that handles both single and batch CloudEvents
app.post('/events/batch', async (req, res) => {
  const contentType = req.headers['content-type'] ?? '';
  let events: CloudEvent<unknown>[];

  if (contentType.includes('application/cloudevents-batch+json')) {
    // Batch mode
    const body = JSON.parse(req.body.toString());
    events = body.map((raw: unknown) => HTTP.toEvent({ headers: {}, body: raw }));
  } else {
    // Single event
    events = [HTTP.toEvent({ headers: req.headers, body: req.body })];
  }

  const results = await Promise.allSettled(events.map(e => routeEvent(e)));
  const failed  = results.filter(r => r.status === 'rejected').length;

  res.status(200).json({ processed: events.length, failed });
});