AliveMCP

Guide · Event-Driven & Async Patterns

MCP Server Dapr — pub/sub, state management, sidecar API, CloudEvents envelope

Three Dapr behaviours catch MCP server authors off-guard: Dapr is a sidecar — your code never calls Kafka or Redis directly, it calls http://localhost:3500 — Dapr's runtime sidecar runs alongside your MCP server process and exposes a uniform HTTP/gRPC API for pub/sub, state, service invocation, and bindings; your tool handlers swap their infrastructure SDK imports for simple HTTP calls to the Dapr sidecar, and Dapr handles the actual connection to Kafka, Redis, RabbitMQ, or any other backing component; pub/sub messages arrive wrapped in a CloudEvents envelope, not as raw payloads — when you subscribe to a Dapr topic, incoming events are delivered as CloudEvents JSON with data nested inside a wrapper that includes specversion, type, source, id, and time; you must extract event.data to get your original payload, or set the subscription to rawPayload: true to bypass the envelope; and state ETags enable optimistic concurrency but only work if you read before you write — Dapr state management returns an ETag with every GET; if you include that ETag in a subsequent POST with concurrency: first-write, the write fails with 409 if another process updated the key between your read and write — the pattern is read → modify → write-with-ETag, not just write.

TL;DR

Install @dapr/dapr (or use raw HTTP to http://localhost:${DAPR_HTTP_PORT}). Publish with client.pubsub.publish(pubSubName, topic, data). Subscribe by registering a POST endpoint at /dapr/subscribe or using server.pubsub.subscribe(). Read/write state with client.state.get(storeName, key) and client.state.save() — always store the returned ETag and pass it back for optimistic updates. Use GET /v1.0/healthz/outbound on the sidecar for health probes. Set DAPR_HTTP_PORT (default 3500) and DAPR_GRPC_PORT (default 50001) from environment, never hardcode.

Dapr sidecar API and client setup

Dapr's programming model separates your application logic from infrastructure concerns. Your MCP server connects to a Dapr sidecar process running on localhost; the sidecar holds the component configuration (which Kafka cluster, which Redis instance, which secret store) and handles retries, circuit-breaking, and observability. The @dapr/dapr SDK wraps the sidecar's HTTP/gRPC API and is the recommended integration point.

import { DaprClient, DaprServer, CommunicationProtocolEnum } from '@dapr/dapr';
import { z } from 'zod';

// Read from env — injected by Dapr CLI or Kubernetes
const DAPR_HOST     = process.env.DAPR_HOST      ?? 'http://localhost';
const DAPR_HTTP_PORT = process.env.DAPR_HTTP_PORT ?? '3500';

const daprClient = new DaprClient({
  daprHost: DAPR_HOST,
  daprPort: DAPR_HTTP_PORT,
  communicationProtocol: CommunicationProtocolEnum.HTTP,
});

// Component names come from Dapr component YAML, not connection strings
const PUBSUB_NAME  = process.env.DAPR_PUBSUB_COMPONENT  ?? 'pubsub';   // e.g. Redis Streams, Kafka
const STATE_STORE  = process.env.DAPR_STATE_COMPONENT   ?? 'statestore'; // e.g. Redis, CosmosDB

// MCP tool — publish a domain event
server.tool(
  'publish_event',
  {
    topic:   z.string().min(1),
    payload: z.record(z.unknown()),
  },
  async ({ topic, payload }) => {
    await daprClient.pubsub.publish(PUBSUB_NAME, topic, payload);
    return {
      content: [{
        type: 'text',
        text: JSON.stringify({ published: true, topic }),
      }],
    };
  }
);

// MCP tool — read state with ETag for optimistic updates
server.tool(
  'get_state',
  { key: z.string().min(1) },
  async ({ key }) => {
    // getStateAndEtag returns { data, etag } — always capture the etag
    const { data, etag } = await daprClient.state.getStateAndEtag(STATE_STORE, key);
    if (data === undefined || data === null) {
      return { content: [{ type: 'text', text: JSON.stringify({ found: false, key }) }] };
    }
    return {
      content: [{
        type: 'text',
        // Return etag to the agent so it can pass it back on write
        text: JSON.stringify({ found: true, key, data, etag }),
      }],
    };
  }
);

The component name (e.g., 'pubsub') is defined in a Dapr component YAML file, not in your application code. This is the key insight: the same MCP server code can run against Redis Streams in development and Kafka in production by changing only the Dapr component YAML, with no code changes. Never import Kafka or Redis SDKs directly if you are using Dapr — let the sidecar handle the protocol translation.

State management — ETag-based optimistic concurrency

Dapr state management supports three concurrency modes: first-write (ETag-based optimistic locking), last-write (last caller wins, no ETag check), and the default (no concurrency control). For MCP servers where multiple tool calls from different agent sessions may update the same key concurrently, first-write prevents lost updates. The workflow is always: read the current value and its ETag, modify the value, write back with the original ETag — the Dapr sidecar rejects the write with HTTP 409 Conflict if another writer modified the key between your read and write.

// MCP tool — update state with optimistic concurrency
server.tool(
  'update_state',
  {
    key:  z.string().min(1),
    data: z.record(z.unknown()),
    etag: z.string().optional(), // omit for create-or-overwrite (last-write)
  },
  async ({ key, data, etag }) => {
    if (etag) {
      // first-write mode — fails with 409 if ETag doesn't match
      try {
        await daprClient.state.saveWithEtag(STATE_STORE, key, data, etag, {
          concurrency: 'first-write',
          consistency: 'strong',
        });
        return { content: [{ type: 'text', text: JSON.stringify({ saved: true, key }) }] };
      } catch (err: any) {
        if (err?.message?.includes('409') || err?.statusCode === 409) {
          return {
            content: [{
              type: 'text',
              text: JSON.stringify({
                error: 'conflict',
                message: 'State was modified by another process. Re-read with get_state and retry.',
                key,
              }),
            }],
          };
        }
        throw err;
      }
    }

    // No ETag — last-write wins (suitable for idempotent or new keys)
    await daprClient.state.save(STATE_STORE, [{ key, value: data }]);
    return { content: [{ type: 'text', text: JSON.stringify({ saved: true, key }) }] };
  }
);

// MCP tool — transactional state update (multiple keys, atomic)
server.tool(
  'bulk_state_update',
  {
    operations: z.array(z.object({
      key:      z.string().min(1),
      value:    z.record(z.unknown()),
      operationType: z.enum(['upsert', 'delete']),
    })).min(1).max(25),
  },
  async ({ operations }) => {
    await daprClient.state.transaction(STATE_STORE, operations.map(op => ({
      operation: op.operationType,
      request:   { key: op.key, value: op.value },
    })));
    return {
      content: [{
        type: 'text',
        text: JSON.stringify({ transacted: true, count: operations.length }),
      }],
    };
  }
);

Pub/sub subscription and CloudEvents envelope

When subscribing to Dapr topics, incoming messages arrive as CloudEvents JSON. The original payload lives inside the data field of the CloudEvent wrapper. Dapr adds envelope fields (specversion, type, source, id, time, topic, pubsubname) that are useful for routing, deduplication, and debugging. If your subscription endpoint does not return a success response within the timeout, Dapr retries the delivery according to the component's retry policy — always return { status: 'SUCCESS' } only after your handler completes, not before.

import { DaprServer } from '@dapr/dapr';
import express from 'express';

const daprServer = new DaprServer({
  serverHost: '127.0.0.1',
  serverPort: process.env.APP_PORT ?? '3000',
  daprHost:   DAPR_HOST,
  daprPort:   DAPR_HTTP_PORT,
  communicationProtocol: CommunicationProtocolEnum.HTTP,
});

// Subscribe to a Dapr topic — handler receives the CloudEvents envelope
await daprServer.pubsub.subscribe(PUBSUB_NAME, 'user-events', async (cloudEvent) => {
  // cloudEvent = { specversion, type, source, id, time, data, topic, pubsubname }
  const payload = cloudEvent.data;  // your original published payload

  // Idempotency check — use cloudEvent.id (UUID) as a dedup key
  const dedupKey = `dedup:${cloudEvent.id}`;
  const { data: existing } = await daprClient.state.getStateAndEtag(STATE_STORE, dedupKey);
  if (existing) {
    return { status: 'SUCCESS' };  // already processed — acknowledge without re-processing
  }

  // Process the event
  await processUserEvent(payload);

  // Mark as processed
  await daprClient.state.save(STATE_STORE, [{
    key:   dedupKey,
    value: { processed_at: new Date().toISOString() },
    metadata: { ttlInSeconds: '86400' },  // auto-expire dedup key after 24 hours
  }]);

  return { status: 'SUCCESS' };
  // Return { status: 'RETRY' } to ask Dapr to retry this message
  // Return { status: 'DROP' }  to discard it (DLQ) — for poison messages
});

// Raw subscription — bypass CloudEvents envelope for binary/custom formats
await daprServer.pubsub.subscribeWithOptions(PUBSUB_NAME, 'raw-events', {
  metadata: { rawPayload: 'true' },  // receive raw bytes, no CloudEvents wrapping
}, async (rawData) => {
  const payload = Buffer.from(rawData as string, 'base64');
  // ... process binary payload
  return { status: 'SUCCESS' };
});

await daprServer.start();

Service invocation and output bindings

Dapr service invocation lets an MCP server call another Dapr-enabled service by its app-id without knowing its address — Dapr handles service discovery, mTLS, and retries. Output bindings provide one-way fire-and-forget integration with external systems (SMTP, Twilio, AWS S3, cron triggers) without requiring a dedicated SDK for each. Both patterns keep your MCP server code infrastructure-agnostic.

// MCP tool — invoke another Dapr service by app-id
server.tool(
  'invoke_service',
  {
    app_id:  z.string().min(1),
    method:  z.string().min(1),
    payload: z.record(z.unknown()).optional(),
  },
  async ({ app_id, method, payload }) => {
    // Dapr resolves the service address — no hostname/port needed in code
    const result = await daprClient.invoker.invoke(
      app_id,
      method,
      HttpMethod.POST,
      payload ?? {}
    );
    return {
      content: [{ type: 'text', text: JSON.stringify(result) }],
    };
  }
);

// MCP tool — trigger an output binding (send email, write S3, etc.)
server.tool(
  'trigger_binding',
  {
    binding_name: z.string().min(1),
    operation:    z.string().min(1),  // e.g. 'create', 'post', 'publish'
    data:         z.record(z.unknown()),
    metadata:     z.record(z.string()).optional(),
  },
  async ({ binding_name, operation, data, metadata }) => {
    await daprClient.binding.send(binding_name, operation, data, metadata ?? {});
    return {
      content: [{ type: 'text', text: JSON.stringify({ triggered: true, binding: binding_name, operation }) }],
    };
  }
);

// Health probe — check that the Dapr sidecar and all its outbound components are healthy
server.resource('dapr_health', 'event://dapr/health', async () => {
  const DAPR_BASE = `${DAPR_HOST}:${DAPR_HTTP_PORT}`;
  try {
    // /v1.0/healthz/outbound checks sidecar readiness including component connections
    const res = await fetch(`${DAPR_BASE}/v1.0/healthz/outbound`);
    const isHealthy = res.status === 204;
    return {
      contents: [{
        uri: 'event://dapr/health',
        text: JSON.stringify({
          status:       isHealthy ? 'ok' : 'degraded',
          sidecar_http: res.status,
          checked_at:   new Date().toISOString(),
        }),
      }],
    };
  } catch (err: any) {
    return {
      contents: [{
        uri: 'event://dapr/health',
        text: JSON.stringify({ status: 'error', message: err.message }),
      }],
    };
  }
});

The health endpoint /v1.0/healthz/outbound (204 on success) differs from /v1.0/healthz (200 on sidecar process alive). Use the outbound variant for MCP server startup probes — it verifies that the sidecar has successfully connected to all configured components before your server starts accepting tool calls. A sidecar that is alive but cannot reach its Kafka broker or Redis instance will return 500 from /outbound.