MCP Tools for Azure Service Bus — queues vs topics, DLQ, message lock renewal, AMQP SDK

Authentication — connection strings vs managed identity

Service Bus supports two authentication modes: Shared Access Signature (SAS) connection strings and Azure AD RBAC with managed identity. SAS connection strings embed a policy name and key — they are simple but require secret rotation. Managed identity (via DefaultAzureCredential) uses no secrets and is preferred for Azure-hosted MCP servers (App Service, Container Apps, AKS). The managed identity needs the Azure Service Bus Data Owner, Data Sender, or Data Receiver role on the namespace or individual queue/topic.

import {
  ServiceBusClient,
  ServiceBusAdministrationClient,
} from '@azure/service-bus';
import { DefaultAzureCredential } from '@azure/identity';

// Option 1: SAS connection string (simple, requires secret management)
const sbClientSAS = new ServiceBusClient(
  process.env.AZURE_SERVICE_BUS_CONNECTION_STRING!
);

// Option 2: Managed identity (no secrets, preferred for Azure-hosted apps)
const fullyQualifiedNamespace = `${process.env.SERVICE_BUS_NAMESPACE}.servicebus.windows.net`;
const credential = new DefaultAzureCredential();
const sbClient = new ServiceBusClient(fullyQualifiedNamespace, credential);

// Administration client — for creating/inspecting queues and topics
const adminClient = new ServiceBusAdministrationClient(
  fullyQualifiedNamespace,
  credential
);

// Graceful shutdown — close the client to drain AMQP connections
process.on('SIGTERM', async () => {
  await sbClient.close();
});

The DefaultAzureCredential chain tries: environment variables → workload identity (AKS) → managed identity (IMDS) → Azure CLI → Azure PowerShell → Azure Developer CLI. For local development, az login provides credentials via the Azure CLI fallback. In CI/CD, set AZURE_CLIENT_ID, AZURE_TENANT_ID, and AZURE_CLIENT_SECRET for service principal authentication.

Queues vs topics/subscriptions — when to use each

Queues implement point-to-point messaging — one sender, one or more competing consumers that each process different messages. Topics implement publish-subscribe — one sender, multiple subscriptions each receiving a copy of every message (filtered or unfiltered). MCP servers that need fan-out (notify multiple downstream systems about an agent action) use topics; MCP servers that dispatch work to a pool of workers use queues.

// Create a queue with configuration (idempotent — skip if exists)
async function ensureQueue(name: string): Promise {
  const exists = await adminClient.queueExists(name);
  if (!exists) {
    await adminClient.createQueue(name, {
      lockDuration: 'PT2M',           // 2 minute lock (increase for slow processors)
      maxDeliveryCount: 5,            // dead-letter after 5 failed attempts
      deadLetteringOnMessageExpiration: true,
      defaultMessageTimeToLive: 'P7D', // messages expire after 7 days
      requiresDuplicateDetection: true,
      duplicateDetectionHistoryTimeWindow: 'PT10M',
    });
  }
}

// Create a topic + subscriptions (Premium tier only)
async function ensureTopicWithSubscriptions(topicName: string): Promise {
  if (!await adminClient.topicExists(topicName)) {
    await adminClient.createTopic(topicName, {
      defaultMessageTimeToLive: 'P1D',
    });
  }
  // Subscription: all messages (no filter)
  if (!await adminClient.subscriptionExists(topicName, 'all-events')) {
    await adminClient.createSubscription(topicName, 'all-events', {
      lockDuration: 'PT1M',
      maxDeliveryCount: 3,
    });
  }
  // Subscription: filtered messages (SQL-like filter on message properties)
  if (!await adminClient.subscriptionExists(topicName, 'high-priority-only')) {
    await adminClient.createSubscription(topicName, 'high-priority-only', {
      lockDuration: 'PT1M',
      maxDeliveryCount: 3,
    });
    // Add a correlation filter on the subscription
    await adminClient.createRule(
      topicName,
      'high-priority-only',
      'priority-filter',
      { sqlExpression: "priority = 'high'" }  // matches messages with priority=high in ApplicationProperties
    );
    // Remove default true-filter that would also match all messages
    await adminClient.deleteRule(topicName, 'high-priority-only', '$Default');
  }
}

Service Bus Standard tier supports only queues — topics require Premium tier. If you're building on Standard and need fan-out, use multiple queues and send the same message to each queue explicitly. Premium tier also adds VNET integration, availability zones, and higher throughput tiers.

Sending messages — batching, messageId, and scheduling

Service Bus messages have a maximum size of 256 KB (Standard) or up to 100 MB with message sessions on Premium. Use createMessageBatch() to pack multiple messages into a single batch up to the maximum batch size — the batch is sent atomically. The messageId property enables duplicate detection: if the queue has requiresDuplicateDetection enabled, messages with the same messageId within the detection window are discarded without error.

const sender = sbClient.createSender('agent-tasks');

// Send a single message
await sender.sendMessages({
  body: { taskType: 'analyze', sourceId: 'src_123' },
  subject: 'analyze-task',            // acts as a routing label
  messageId: `analyze-src_123-${Date.now()}`,  // for deduplication
  timeToLive: 3600_000,               // milliseconds; overrides queue default TTL
  applicationProperties: {           // custom key-value pairs for filtering/routing
    priority: 'high',
    tenantId: 'tenant_abc',
  },
});

// Send a batch of messages atomically
const batch = await sender.createMessageBatch();
const messages = items.map(item => ({
  body: item,
  messageId: `item-${item.id}`,
}));

for (const msg of messages) {
  if (!batch.tryAddMessage(msg)) {
    // Batch is full — send current batch and start a new one
    await sender.sendMessages(batch);
    const newBatch = await sender.createMessageBatch();
    newBatch.tryAddMessage(msg);
  }
}
await sender.sendMessages(batch);

// Schedule a message to appear in the queue at a specific time
const scheduledEnqueueTime = new Date(Date.now() + 5 * 60 * 1000); // 5 minutes from now
const [sequenceNumber] = await sender.scheduleMessages(
  [{ body: { taskType: 'reminder', userId: 'usr_456' }, subject: 'reminder' }],
  scheduledEnqueueTime
);
// Cancel: await sender.cancelScheduledMessages([sequenceNumber]);

await sender.close();

Session-enabled queues (set requiresSession: true at queue creation) group messages by sessionId property and ensure all messages in a session are processed in order by a single consumer. This is useful for MCP tools that dispatch ordered sequences of operations for the same entity — for example, a series of steps in an agent workflow that must execute sequentially for a given user.

Receiving messages — lock renewal and completion

Messages are locked when received — the lock prevents other consumers from processing the same message. The lock duration is set at queue creation time (default 60 seconds). If processing takes longer than the lock duration, Service Bus makes the message visible to other consumers. Call receiver.renewMessageLock(message) before the lock expires to extend it. After successful processing, call receiver.completeMessage(message) to remove it from the queue. On failure, call receiver.abandonMessage(message) to immediately re-enqueue it, or receiver.deadLetterMessage(message, { reason }) to send it to the DLQ.

const receiver = sbClient.createReceiver('agent-tasks');
const LOCK_RENEWAL_INTERVAL_MS = 30_000;  // renew every 30s (half of 60s lock duration)

async function processMessagesWithLockRenewal(): Promise {
  const messages = await receiver.receiveMessages(10, { maxWaitTimeInMs: 5000 });

  for (const message of messages) {
    // Start lock renewal timer — renew every 30 seconds
    const renewalTimer = setInterval(async () => {
      try {
        await receiver.renewMessageLock(message);
      } catch {
        clearInterval(renewalTimer);  // message lock lost — stop renewing
      }
    }, LOCK_RENEWAL_INTERVAL_MS);

    try {
      // Process the message (may take longer than the lock duration)
      await processTask(message.body);

      // Success — complete (remove from queue)
      clearInterval(renewalTimer);
      await receiver.completeMessage(message);

    } catch (err) {
      clearInterval(renewalTimer);
      const deliveryCount = message.deliveryCount ?? 0;

      if (deliveryCount >= 4) {
        // Exhausted retries — send to DLQ with reason
        await receiver.deadLetterMessage(message, {
          deadLetterReason: 'MaxRetriesExceeded',
          deadLetterErrorDescription: String(err),
        });
      } else {
        // Retry: abandon puts message back in queue immediately
        await receiver.abandonMessage(message);
        // Or defer delivery: receiver.deferMessage(message) — requires sequenceNumber to retrieve later
      }
    }
  }
}

// Subscribe-based processor (handles lock renewal automatically)
receiver.subscribe({
  processMessage: async (message) => {
    await processTask(message.body);
    // completeMessage is called automatically when processMessage returns without error
  },
  processError: async (args) => {
    console.error('Receiver error:', args.error);
  },
}, {
  autoCompleteMessages: true,   // complete on processMessage success, abandon on error
  maxConcurrentCalls: 5,        // process up to 5 messages simultaneously
  maxAutoLockRenewalDurationInMs: 5 * 60 * 1000,  // auto-renew for up to 5 minutes
});

await receiver.close();

The subscribe API with autoCompleteMessages: true and maxAutoLockRenewalDurationInMs is the simplest approach for most MCP use cases — it handles lock renewal automatically and completes or abandons messages based on whether processMessage resolved or threw. Use manual lock management only when you need fine-grained control over completion vs dead-lettering vs deferral.

Dead-letter queue — draining and reprocessing

Every Service Bus queue and subscription has a built-in dead-letter sub-queue accessible as queueName/$DeadLetterQueue. Messages land in the DLQ when they exceed maxDeliveryCount, when they expire (if deadLetteringOnMessageExpiration is true), or when receiver.deadLetterMessage() is called explicitly. DLQ messages are not automatically retried — they require manual inspection and reprocessing.

// Read from DLQ without removing (peek)
async function inspectDLQ(queueName: string, maxMessages: number = 10) {
  const dlqReceiver = sbClient.createReceiver(
    queueName,
    { subQueueType: 'deadLetter' }
  );
  const messages = await dlqReceiver.peekMessages(maxMessages);
  const summary = messages.map(m => ({
    messageId: m.messageId,
    deadLetterReason: m.deadLetterReason,
    deadLetterErrorDescription: m.deadLetterErrorDescription,
    deliveryCount: m.deliveryCount,
    enqueuedAt: m.enqueuedTimeUtc,
    body: m.body,
  }));
  await dlqReceiver.close();
  return summary;
}

// Reprocess DLQ messages — move back to main queue after fixing the issue
async function reprocessDLQ(queueName: string): Promise {
  const dlqReceiver = sbClient.createReceiver(queueName, { subQueueType: 'deadLetter' });
  const sender = sbClient.createSender(queueName);
  let reprocessed = 0;

  const messages = await dlqReceiver.receiveMessages(50, { maxWaitTimeInMs: 2000 });
  for (const msg of messages) {
    try {
      // Re-send the original message body back to the main queue
      await sender.sendMessages({
        body: msg.body,
        subject: msg.subject,
        messageId: `retry-${msg.messageId}`,
        applicationProperties: msg.applicationProperties,
      });
      await dlqReceiver.completeMessage(msg);  // remove from DLQ
      reprocessed++;
    } catch {
      await dlqReceiver.abandonMessage(msg);  // leave in DLQ
    }
  }
  await dlqReceiver.close();
  await sender.close();
  return reprocessed;
}

// Queue metrics — includes DLQ count
async function getQueueMetrics(queueName: string) {
  const props = await adminClient.getQueueRuntimeProperties(queueName);
  return {
    activeMessages: props.activeMessageCount,
    deadLetterMessages: props.deadLetterMessageCount,
    scheduledMessages: props.scheduledMessageCount,
    transferredMessages: props.transferMessageCount,
    sizeBytes: props.sizeInBytes,
  };
}

Monitor the deadLetterMessageCount metric as a key operational signal — a growing DLQ count indicates persistent processing failures that require attention. AliveMCP can alert on queue metric thresholds by polling the Service Bus management API. Set up an alert when deadLetterMessageCount exceeds a threshold (e.g., 10 messages) to catch systematic failures before they accumulate into large backlogs.

Health probe — namespace connectivity and queue metrics

Use the administration client to verify Service Bus namespace connectivity and check queue health. adminClient.getQueueRuntimeProperties() makes a single management API call that confirms both authentication and queue accessibility. The response includes active, dead-letter, and scheduled message counts — all essential for operational health monitoring.

async function checkServiceBusHealth(queueName: string): Promise