Azure Eventhub TS

MicrosoftOfficialCreated by Microsoft

Skill
Tool integration
Cloud and DevOps
MIT
No issues found

From the author

Build event streaming applications using Azure Event Hubs SDK for JavaScript (@azure/event-hubs). Use when implementing high-throughput event ingestion, real-time analytics, IoT telemetry, or event-driven architectures with partitioned consumers.

— the description by Microsoft

AutoGineer summary

Written by AutoGineer from the skill’s own files, not by the author

This skill provides the necessary patterns and interfaces to interact with Azure Event Hubs using the @azure/event-hubs SDK. It covers the core functionality for producing and consuming event streams, including batch processing, partition management, and checkpointing strategies using Azure Blob Storage.

What it does

Establishes connections to Azure Event Hubs using identity-based authentication.
Supports sending events individually or in batches to specific partitions.
Enables consumption of event streams with configurable batch sizes and wait times.
Provides mechanisms for persistent checkpointing to track event processing progress.
Offers event handlers for initialization, error management, and event processing.
Allows retrieval of metadata regarding Event Hub partitions and sequence numbers.

When to use it

When building event-driven architectures in TypeScript.
When implementing high-throughput data ingestion pipelines.
When you need to manage event consumption across multiple partitions.
When integrating real-time telemetry or streaming data into your application.

Advantages

Native support for TypeScript and Azure identity providers.
Flexible subscription patterns for handling event streams.
Built-in support for checkpointing to manage state across restarts.
Granular control over partition selection and event positioning.

Usage examples

“How do I send a batch of events to an Azure Event Hub?”: The agent provides a code snippet using EventHubProducerClient to create a batch, add events with body data, and send the batch to the hub.
“Show me how to consume events with checkpointing.”: The agent provides an example using EventHubConsumerClient configured with a BlobCheckpointStore to process events and update the checkpoint after each batch.

Install

Download the skill as a ZIP:
https://autogineer.com/agent-skills/microsoft/azure-eventhub-ts/download

Unzip it into the cross-client skills folder of your project:

mkdir -p .agents/skills && unzip azure-eventhub-ts.zip -d .agents/skills

Documentation by Microsoft

Azure Event Hubs SDK for TypeScript

High-throughput event streaming and real-time data ingestion.

Installation

npm install @azure/event-hubs @azure/identity

For checkpointing with consumer groups:

npm install @azure/eventhubs-checkpointstore-blob @azure/storage-blob

Environment Variables

EVENTHUB_NAMESPACE=<namespace>.servicebus.windows.net
EVENTHUB_NAME=my-eventhub
STORAGE_ACCOUNT_NAME=<storage-account>
STORAGE_CONTAINER_NAME=checkpoints
AZURE_TOKEN_CREDENTIALS=prod # Required only if DefaultAzureCredential is used in production

Authentication

import { EventHubProducerClient, EventHubConsumerClient } from "@azure/event-hubs";
import { DefaultAzureCredential, ManagedIdentityCredential } from "@azure/identity";

const fullyQualifiedNamespace = process.env.EVENTHUB_NAMESPACE!;
const eventHubName = process.env.EVENTHUB_NAME!;
// Local dev: DefaultAzureCredential. Production: set AZURE_TOKEN_CREDENTIALS=prod or AZURE_TOKEN_CREDENTIALS=<specific_credential>
const credential = new DefaultAzureCredential({requiredEnvVars: ["AZURE_TOKEN_CREDENTIALS"]});
// Or use a specific credential directly in production:
// See https://learn.microsoft.com/javascript/api/overview/azure/identity-readme?view=azure-node-latest#credential-classes
// const credential = new ManagedIdentityCredential();

// Producer
const producer = new EventHubProducerClient(fullyQualifiedNamespace, eventHubName, credential);

// Consumer
const consumer = new EventHubConsumerClient(
  "$Default", // Consumer group
  fullyQualifiedNamespace,
  eventHubName,
  credential
);

Core Workflow

Send Events

const producer = new EventHubProducerClient(namespace, eventHubName, credential);

// Create batch and add events
const batch = await producer.createBatch();
batch.tryAdd({ body: { temperature: 72.5, deviceId: "sensor-1" } });
batch.tryAdd({ body: { temperature: 68.2, deviceId: "sensor-2" } });

await producer.sendBatch(batch);
await producer.close();

Send to Specific Partition

// By partition ID
const batch = await producer.createBatch({ partitionId: "0" });

// By partition key (consistent hashing)
const batch = await producer.createBatch({ partitionKey: "device-123" });

Receive Events (Simple)

const consumer = new EventHubConsumerClient("$Default", namespace, eventHubName, credential);

const subscription = consumer.subscribe({
  processEvents: async (events, context) => {
    for (const event of events) {
      console.log(`Partition: ${context.partitionId}, Body: ${JSON.stringify(event.body)}`);
    }
  },
  processError: async (err, context) => {
    console.error(`Error on partition ${context.partitionId}: ${err.message}`);
  },
});

// Stop after some time
setTimeout(async () => {
  await subscription.close();
  await consumer.close();
}, 60000);

Receive with Checkpointing (Production)

import { EventHubConsumerClient } from "@azure/event-hubs";
import { ContainerClient } from "@azure/storage-blob";
import { BlobCheckpointStore } from "@azure/eventhubs-checkpointstore-blob";

const containerClient = new ContainerClient(
  `https://${storageAccount}.blob.core.windows.net/${containerName}`,
  credential
);

const checkpointStore = new BlobCheckpointStore(containerClient);

const consumer = new EventHubConsumerClient(
  "$Default",
  namespace,
  eventHubName,
  credential,
  checkpointStore
);

const subscription = consumer.subscribe({
  processEvents: async (events, context) => {
    for (const event of events) {
      console.log(`Processing: ${JSON.stringify(event.body)}`);
    }
    // Checkpoint after processing batch
    if (events.length > 0) {
      await context.updateCheckpoint(events[events.length - 1]);
    }
  },
  processError: async (err, context) => {
    console.error(`Error: ${err.message}`);
  },
});

Receive from Specific Position

const subscription = consumer.subscribe({
  processEvents: async (events, context) => { /* ... */ },
  processError: async (err, context) => { /* ... */ },
}, {
  startPosition: {
    // Start from beginning
    "0": { offset: "@earliest" },
    // Start from end (new events only)
    "1": { offset: "@latest" },
    // Start from specific offset
    "2": { offset: "12345" },
    // Start from specific time
    "3": { enqueuedOn: new Date("2024-01-01") },
  },
});

Event Hub Properties

// Get hub info
const hubProperties = await producer.getEventHubProperties();
console.log(`Partitions: ${hubProperties.partitionIds}`);

// Get partition info
const partitionProperties = await producer.getPartitionProperties("0");
console.log(`Last sequence: ${partitionProperties.lastEnqueuedSequenceNumber}`);

Batch Processing Options

const subscription = consumer.subscribe(
  {
    processEvents: async (events, context) => { /* ... */ },
    processError: async (err, context) => { /* ... */ },
  },
  {
    maxBatchSize: 100,           // Max events per batch
    maxWaitTimeInSeconds: 30,    // Max wait for batch
  }
);

Key Types

import {
  EventHubProducerClient,
  EventHubConsumerClient,
  EventData,
  ReceivedEventData,
  PartitionContext,
  Subscription,
  SubscriptionEventHandlers,
  CreateBatchOptions,
  EventPosition,
} from "@azure/event-hubs";

import { BlobCheckpointStore } from "@azure/eventhubs-checkpointstore-blob";

Event Properties

// Send with properties
const batch = await producer.createBatch();
batch.tryAdd({
  body: { data: "payload" },
  properties: {
    eventType: "telemetry",
    deviceId: "sensor-1",
  },
  contentType: "application/json",
  correlationId: "request-123",
});

// Access in receiver
consumer.subscribe({
  processEvents: async (events, context) => {
    for (const event of events) {
      console.log(`Type: ${event.properties?.eventType}`);
      console.log(`Sequence: ${event.sequenceNumber}`);
      console.log(`Enqueued: ${event.enqueuedTimeUtc}`);
      console.log(`Offset: ${event.offset}`);
    }
  },
});

Error Handling

consumer.subscribe({
  processEvents: async (events, context) => {
    try {
      for (const event of events) {
        await processEvent(event);
      }
      await context.updateCheckpoint(events[events.length - 1]);
    } catch (error) {
      // Don't checkpoint on error - events will be reprocessed
      console.error("Processing failed:", error);
    }
  },
  processError: async (err, context) => {
    if (err.name === "MessagingError") {
      // Transient error - SDK will retry
      console.warn("Transient error:", err.message);
    } else {
      // Fatal error
      console.error("Fatal error:", err);
    }
  },
});

Best Practices

Use checkpointing - Always checkpoint in production for exactly-once processing
Batch sends - Use createBatch() for efficient sending
Partition keys - Use partition keys to ensure ordering for related events
Consumer groups - Use separate consumer groups for different processing pipelines
Handle errors gracefully - Don't checkpoint on processing failures
Close clients - Always close producer/consumer when done
Monitor lag - Track lastEnqueuedSequenceNumber vs processed sequence

Files

ATTRIBUTION.mdAdded by AutoGineer970 B
LICENSELicence file added by AutoGineer1.1 KiB
SKILL.md7.6 KiB
references/checkpointing.md11.3 KiB
references/event-processing.md11.0 KiB

references/event-processing.md

Event Processing Reference

Receiving and processing events with Azure Event Hubs using the @azure/event-hubs TypeScript SDK.

Overview

Event Hubs provides high-throughput event streaming. This reference covers the EventHubConsumerClient, subscription patterns, event handlers, and processing strategies.

Key Interfaces

import {
  EventHubConsumerClient,
  ReceivedEventData,
  PartitionContext,
  SubscriptionEventHandlers,
  SubscribeOptions,
  Subscription
} from "@azure/event-hubs";

// ReceivedEventData - structure of received events
interface ReceivedEventData {
  body: any;
  contentType?: string;
  correlationId?: string | number | Buffer;
  enqueuedTimeUtc: Date;
  messageId?: string | number | Buffer;
  offset: string;
  partitionKey: string | null;
  properties?: Record<string, any>;
  sequenceNumber: number;
  systemProperties?: Record<string, any>;
}

// PartitionContext - context for event handlers
interface PartitionContext {
  readonly consumerGroup: string;
  readonly eventHubName: string;
  readonly fullyQualifiedNamespace: string;
  readonly partitionId: string;
  readonly lastEnqueuedEventProperties?: LastEnqueuedEventProperties;
  updateCheckpoint(eventData: ReceivedEventData): Promise<void>;
}

// Handler types
type ProcessEventsHandler = (
  events: ReceivedEventData[],
  context: PartitionContext
) => Promise<void>;

type ProcessErrorHandler = (
  error: Error,
  context: PartitionContext
) => Promise<void>;

type ProcessInitializeHandler = (
  context: PartitionContext
) => Promise<void>;

type ProcessCloseHandler = (
  reason: CloseReason,
  context: PartitionContext
) => Promise<void>;

import { EventHubConsumerClient, earliestEventPosition } from "@azure/event-hubs";
import { DefaultAzureCredential } from "@azure/identity";

const fullyQualifiedNamespace = "my-namespace.servicebus.windows.net";
const eventHubName = "my-event-hub";
const consumerGroup = "$Default";

const client = new EventHubConsumerClient(
  consumerGroup,
  fullyQualifiedNamespace,
  eventHubName,
  new DefaultAzureCredential()
);

const subscription = client.subscribe({
  processEvents: async (events, context) => {
    for (const event of events) {
      console.log(`Partition: ${context.partitionId}`);
      console.log(`Body: ${JSON.stringify(event.body)}`);
      console.log(`Sequence: ${event.sequenceNumber}`);
      console.log(`Enqueued: ${event.enqueuedTimeUtc}`);
    }
  },
  processError: async (err, context) => {
    console.error(`Error on partition ${context.partitionId}: ${err.message}`);
  }
}, {
  startPosition: earliestEventPosition
});

// Stop after some time
setTimeout(async () => {
  await subscription.close();
  await client.close();
}, 60000);

Full Subscription with All Handlers

import {
  EventHubConsumerClient,
  ReceivedEventData,
  PartitionContext,
  CloseReason,
  earliestEventPosition,
  latestEventPosition
} from "@azure/event-hubs";

const client = new EventHubConsumerClient(
  "$Default",
  connectionString,
  eventHubName
);

const subscription = client.subscribe({
  processInitialize: async (context: PartitionContext) => {
    console.log(`Started receiving from partition ${context.partitionId}`);
  },

  processEvents: async (events: ReceivedEventData[], context: PartitionContext) => {
    if (events.length === 0) {
      console.log(`No events received. Waiting...`);
      return;
    }

    console.log(`Received ${events.length} events from partition ${context.partitionId}`);
    
    for (const event of events) {
      // Process each event
      await processEvent(event, context);
    }

    // Checkpoint after processing batch
    const lastEvent = events[events.length - 1];
    await context.updateCheckpoint(lastEvent);
  },

  processError: async (err: Error, context: PartitionContext) => {
    console.error(`Error on partition ${context.partitionId}:`);
    console.error(err.message);
  },

  processClose: async (reason: CloseReason, context: PartitionContext) => {
    console.log(`Stopped receiving from partition ${context.partitionId}`);
    console.log(`Reason: ${reason}`);
  }
}, {
  startPosition: earliestEventPosition,
  maxBatchSize: 100,
  maxWaitTimeInSeconds: 30
});

const client = new EventHubConsumerClient(
  "$Default",
  connectionString,
  eventHubName
);

// Get partition IDs
const partitionIds = await client.getPartitionIds();
console.log(`Partitions: ${partitionIds.join(", ")}`);

// Subscribe to specific partition
const subscription = client.subscribe(
  partitionIds[0], // First partition only
  {
    processEvents: async (events, context) => {
      console.log(`Partition ${context.partitionId}: ${events.length} events`);
    },
    processError: async (err, context) => {
      console.error(err);
    }
  },
  { startPosition: earliestEventPosition }
);

Start Position Options

import { 
  earliestEventPosition, 
  latestEventPosition,
  EventPosition 
} from "@azure/event-hubs";

// Start from beginning (all historical events)
const fromBeginning = { startPosition: earliestEventPosition };

// Start from end (new events only)
const fromEnd = { startPosition: latestEventPosition };

// Start from specific offset
const fromOffset: SubscribeOptions = {
  startPosition: { offset: "12345" }
};

// Start from specific sequence number
const fromSequence: SubscribeOptions = {
  startPosition: { sequenceNumber: 1000 }
};

// Start from specific time
const fromTime: SubscribeOptions = {
  startPosition: { enqueuedOn: new Date("2024-01-01T00:00:00Z") }
};

// Different positions per partition
const perPartition: SubscribeOptions = {
  startPosition: {
    "0": earliestEventPosition,
    "1": latestEventPosition,
    "2": { offset: "5000" }
  }
};

SubscribeOptions Reference

interface SubscribeOptions {
  /** Starting position for reading events */
  startPosition?: EventPosition | Record<string, EventPosition>;
  
  /** Max events per batch (default: varies) */
  maxBatchSize?: number;
  
  /** Max wait time in seconds for a batch */
  maxWaitTimeInSeconds?: number;
  
  /** Track last enqueued event info (for lag monitoring) */
  trackLastEnqueuedEventProperties?: boolean;
  
  /** Owner level for exclusive access */
  ownerLevel?: number;
}

Event Processing Patterns

Sequential Processing

const subscription = client.subscribe({
  processEvents: async (events, context) => {
    for (const event of events) {
      await processEventSequentially(event);
    }
    await context.updateCheckpoint(events[events.length - 1]);
  },
  processError: async (err, context) => {
    console.error(err);
  }
});

Parallel Processing within Batch

const subscription = client.subscribe({
  processEvents: async (events, context) => {
    // Process events in parallel
    await Promise.all(
      events.map(event => processEventAsync(event))
    );
    
    // Checkpoint after all complete
    if (events.length > 0) {
      await context.updateCheckpoint(events[events.length - 1]);
    }
  },
  processError: async (err, context) => {
    console.error(err);
  }
});

With Error Handling per Event

const subscription = client.subscribe({
  processEvents: async (events, context) => {
    const results = await Promise.allSettled(
      events.map(event => processEvent(event))
    );
    
    const failures = results.filter(r => r.status === "rejected");
    if (failures.length > 0) {
      console.error(`${failures.length} events failed processing`);
      // Decide: checkpoint anyway or skip?
    }
    
    // Checkpoint even with some failures (at-least-once)
    if (events.length > 0) {
      await context.updateCheckpoint(events[events.length - 1]);
    }
  },
  processError: async (err, context) => {
    console.error(err);
  }
});

Monitoring Consumer Lag

const subscription = client.subscribe({
  processEvents: async (events, context) => {
    // Check lag if tracking enabled
    if (context.lastEnqueuedEventProperties) {
      const lastEnqueued = context.lastEnqueuedEventProperties.sequenceNumber;
      const lastProcessed = events.length > 0 
        ? events[events.length - 1].sequenceNumber 
        : 0;
      
      const lag = lastEnqueued - lastProcessed;
      console.log(`Partition ${context.partitionId} lag: ${lag} events`);
    }
    
    // Process events...
  },
  processError: async (err, context) => {
    console.error(err);
  }
}, {
  trackLastEnqueuedEventProperties: true
});

Event Properties Access

const subscription = client.subscribe({
  processEvents: async (events, context) => {
    for (const event of events) {
      // Standard properties
      console.log(`Body: ${event.body}`);
      console.log(`Partition Key: ${event.partitionKey}`);
      console.log(`Sequence Number: ${event.sequenceNumber}`);
      console.log(`Offset: ${event.offset}`);
      console.log(`Enqueued Time: ${event.enqueuedTimeUtc}`);
      
      // Custom properties (set by producer)
      if (event.properties) {
        console.log(`Event Type: ${event.properties.eventType}`);
        console.log(`Device ID: ${event.properties.deviceId}`);
      }
      
      // System properties
      if (event.systemProperties) {
        console.log(`System Props: ${JSON.stringify(event.systemProperties)}`);
      }
      
      // Content type
      if (event.contentType) {
        console.log(`Content Type: ${event.contentType}`);
      }
    }
  },
  processError: async (err, context) => {
    console.error(err);
  }
});

Graceful Shutdown

let subscription: Subscription;

async function start() {
  const client = new EventHubConsumerClient(
    "$Default",
    connectionString,
    eventHubName
  );

  subscription = client.subscribe({
    processEvents: async (events, context) => {
      // Process events...
    },
    processError: async (err, context) => {
      console.error(err);
    }
  });

  // Handle shutdown signals
  process.on("SIGINT", async () => {
    console.log("Shutting down...");
    await subscription.close();
    await client.close();
    process.exit(0);
  });

  process.on("SIGTERM", async () => {
    console.log("Shutting down...");
    await subscription.close();
    await client.close();
    process.exit(0);
  });
}

start().catch(console.error);

Best Practices

Always handle empty batches — processEvents may receive empty arrays
Checkpoint after processing — Not before, to avoid data loss
Use consumer groups — Separate groups for different processing pipelines
Monitor lag — Enable trackLastEnqueuedEventProperties
Handle errors gracefully — Don't crash on individual event failures
Close resources — Always close subscription and client on shutdown
Set appropriate batch size — Balance throughput vs latency

Requirements and compatibility

Works with: Claude Code, Claude apps, OpenAI Codex, GitHub Copilot, Cursor, Gemini CLI, Kiro, OpenCode, Goose, Amp, Roo Code, JetBrains Junie, Factory Droid, Devin (ex-Windsurf), Mistral Vibe, OpenHands, Letta

“Works with” means the tool reads the open Agent Skills format; a skill written for one tool may use features others lack.

Needs: Needs Node.js

Security

No issues found by our static scan· No issues found

A static scan is not a guarantee. Skills are third-party content: review a skill before you let an agent use it.

Attribution and licence

Created by

Microsoft

Publisher

Microsoft (Official)

Source

microsoft/skills · .github/plugins/azure-sdk-typescript/skills/azure-eventhub-ts · commit 0696600

Licence

MIT License (redistribution allowed)

LICENSE (added by AutoGineer from LICENSE)
ATTRIBUTION.md (written by AutoGineer)

Synced

2026-11-09

AutoGineer is not affiliated with or endorsed by Microsoft or Microsoft.

Download ZIP

The ZIP holds the skill’s files, its licence files and an ATTRIBUTION.md. Download azure-eventhub-ts.zip

From the author

AutoGineer summary

What it does

When to use it

Advantages

Usage examples

Install

Documentation by Microsoft

Azure Event Hubs SDK for TypeScript

Installation

Environment Variables

Authentication

Core Workflow

Send Events

Send to Specific Partition

Receive Events (Simple)

Receive with Checkpointing (Production)

Receive from Specific Position

Event Hub Properties

Batch Processing Options

Key Types

Event Properties

Error Handling

Best Practices

Files

references/event-processing.md

Event Processing Reference

Overview

Key Interfaces

Basic Subscribe Pattern

Full Subscription with All Handlers

Subscribe to Specific Partition

Start Position Options

SubscribeOptions Reference

Event Processing Patterns

Sequential Processing

Parallel Processing within Batch

With Error Handling per Event

Monitoring Consumer Lag

Event Properties Access

Graceful Shutdown

Best Practices

See Also

Requirements and compatibility

Security

Attribution and licence

Related skills

Modal

Cloud Infrastructure Network Engineer

Deployment Validation Cloud Architect

Airflow Dag Patterns

Cost Optimization

Hybrid Cloud Networking

Download ZIP