microsoft/eventstream-consumption-cli

Update Check — ONCE PER SESSION (mandatory) The first time this skill is used in a session, run the check-updates skill before proceeding.

• GitHub Copilot CLI / VS Code: invoke the check-updates skill.
• Claude Code / Cowork / Cursor / Windsurf / Codex: compare local vs remote package.json version.
• Skip if the check was already performed earlier in this session.

CRITICAL NOTES

1. To find the workspace details (including its ID) from workspace name: list all workspaces and, then, use JMESPath filtering
2. To find the item details (including its ID) from workspace ID, item type, and item name: list all items of that type in that workspace and, then, use JMESPath filtering
3. Eventstream ≠ Eventhouse. Eventstream is a real-time event ingestion and routing pipeline. For KQL queries, use eventhouse-consumption-cli.

Eventstream Consumption — CLI Skill

Table of Contents

| Task | Reference | Notes | |---|---|---| | Finding Workspaces and Items in Fabric | COMMON-CLI.md § Finding Workspaces and Items in Fabric | Mandatory — READ link first [needed for finding workspace id by its name or item id by its name, item type, and workspace id] | | Fabric Topology & Key Concepts | COMMON-CORE.md § Fabric Topology & Key Concepts | | | Environment URLs | COMMON-CORE.md § Environment URLs | | | Authentication & Token Acquisition | COMMON-CORE.md § Authentication & Token Acquisition | Wrong audience = 401; read before any auth issue | | Core Control-Plane REST APIs | COMMON-CORE.md § Core Control-Plane REST APIs | Includes pagination, LRO polling, and rate-limiting patterns | | Gotchas, Best Practices & Troubleshooting | COMMON-CORE.md § Gotchas, Best Practices & Troubleshooting | | | Tool Selection Rationale | COMMON-CLI.md § Tool Selection Rationale | | | Authentication Recipes | COMMON-CLI.md § Authentication Recipes | az login flows and token acquisition | | Fabric Control-Plane API via az rest | COMMON-CLI.md § Fabric Control-Plane API via az rest | Always pass --resource; includes pagination and LRO helpers | | Gotchas & Troubleshooting (CLI-Specific) | COMMON-CLI.md § Gotchas & Troubleshooting (CLI-Specific) | az rest audience, shell escaping, token expiry | | Quick Reference | COMMON-CLI.md § Quick Reference | az rest template + token audience/tool matrix | | Listing and Discovering Eventstreams | EVENTSTREAM-CONSUMPTION-CORE.md § Listing and Discovering Eventstreams | List, Get, Search across workspaces | | Inspecting Eventstream Topology | EVENTSTREAM-CONSUMPTION-CORE.md § Inspecting Eventstream Topology | Decode base64 definition → trace graph flow | | Monitoring Eventstream Health | EVENTSTREAM-CONSUMPTION-CORE.md § Monitoring Eventstream Health | Retention and throughput checks | | Source and Destination Status | EVENTSTREAM-CONSUMPTION-CORE.md § Source and Destination Status | Validation checklist for sources and destinations | | Integration with Downstream Analytics | EVENTSTREAM-CONSUMPTION-CORE.md § Integration with Downstream Analytics | Eventhouse, Lakehouse, Activator, Real-Time Hub | | Gotchas and Troubleshooting Reference | EVENTSTREAM-CONSUMPTION-CORE.md § Gotchas and Troubleshooting Reference | 10 common issues with causes and fixes | | List Eventstreams | SKILL.md § List Eventstreams | | | Inspect Eventstream Topology | SKILL.md § Inspect Eventstream Topology | Decode and explore the graph | | Get Custom Endpoint Connection String | SKILL.md § Get Custom Endpoint Connection String | Retrieve Kafka/EH connection via Topology API | | Validate Eventstream Configuration | SKILL.md § Validate Eventstream Configuration | | | Gotchas, Rules, Troubleshooting | SKILL.md § Gotchas, Rules, Troubleshooting | MUST DO / AVOID / PREFER checklists |

---

List Eventstreams

List All Eventstreams in a Workspace

az rest --method GET \
  --url "https://api.fabric.microsoft.com/v1/workspaces/${WORKSPACE_ID}/eventstreams" \
  --resource "https://api.fabric.microsoft.com"

Returns an array of Eventstream items. Use JMESPath to filter by name:

az rest --method GET \
  --url "https://api.fabric.microsoft.com/v1/workspaces/${WORKSPACE_ID}/eventstreams" \
  --resource "https://api.fabric.microsoft.com" \
  --query "value[?displayName=='my-eventstream']"

Get Eventstream Details

az rest --method GET \
  --url "https://api.fabric.microsoft.com/v1/workspaces/${WORKSPACE_ID}/eventstreams/${EVENTSTREAM_ID}" \
  --resource "https://api.fabric.microsoft.com"

---

Inspect Eventstream Topology

Tip: The Topology API (GET .../eventstreams/{id}/topology) returns runtime status, error info, and node IDs without base64 decoding. Prefer it for operational inspection (health checks, connection retrieval). Use POST .../getDefinition (below) when you need the full authoring-time graph structure for topology modification.

Retrieve the Eventstream definition and decode it to inspect the full graph topology.

Step 1: Get the Definition

API Note: The Eventstream Definition API uses POST .../getDefinition, not GET .../definition. This follows the Fabric Items Definition pattern. See official docs.

az rest --method POST \
  --url "https://api.fabric.microsoft.com/v1/workspaces/${WORKSPACE_ID}/eventstreams/${EVENTSTREAM_ID}/getDefinition" \
  --resource "https://api.fabric.microsoft.com" \
  --body '{}'

Step 2: Decode the Topology

Extract the eventstream.json part's payload field and base64-decode it:

# Using jq + base64 (Linux; on macOS use base64 -D instead of -d)
az rest --method POST \
  --url "https://api.fabric.microsoft.com/v1/workspaces/${WORKSPACE_ID}/eventstreams/${EVENTSTREAM_ID}/getDefinition" \
  --resource "https://api.fabric.microsoft.com" \
  --body '{}' \
  | jq -r '.definition.parts[] | select(.path=="eventstream.json") | .payload' \
  | base64 -d | jq .

# PowerShell (Windows)
$def = az rest --method POST `
  --url "https://api.fabric.microsoft.com/v1/workspaces/$WORKSPACE_ID/eventstreams/$EVENTSTREAM_ID/getDefinition" `
  --resource "https://api.fabric.microsoft.com" `
  --body '{}' | ConvertFrom-Json
$payload = ($def.definition.parts | Where-Object { $_.path -eq 'eventstream.json' }).payload
[Text.Encoding]::UTF8.GetString([Convert]::FromBase64String($payload)) | ConvertFrom-Json | ConvertTo-Json -Depth 10

Step 3: Summarize the Topology

After decoding, count and list each node type:

| Metric | Path in decoded JSON | |--------|---------------------| | Sources | .sources[] \| .name, .type | | Destinations | .destinations[] \| .name, .type | | Operators | .operators[] \| .name, .type | | Streams | .streams[] \| .name, .type |

---

Get Custom Endpoint Connection String

The POST .../getDefinition endpoint returns empty properties for Custom Endpoint sources. To retrieve the Kafka/Event Hub connection info, use the Topology API /connection endpoint.

Important: This endpoint requires Eventstream.ReadWrite.All permission scope (not just Read).

Step 1: Get the Topology to Find the Source ID

az rest --method GET \
  --url "https://api.fabric.microsoft.com/v1/workspaces/${WORKSPACE_ID}/eventstreams/${EVENTSTREAM_ID}/topology" \
  --resource "https://api.fabric.microsoft.com"

From the response, find the Custom Endpoint source node and extract its id:

# Extract the sourceId for a Custom Endpoint source (use name filter if multiple exist)
SOURCE_ID=$(az rest --method GET \
  --url "https://api.fabric.microsoft.com/v1/workspaces/${WORKSPACE_ID}/eventstreams/${EVENTSTREAM_ID}/topology" \
  --resource "https://api.fabric.microsoft.com" \
  | jq -r '[.sources[] | select(.type=="CustomEndpoint")] | if length == 0 then error("No Custom Endpoint sources found in this Eventstream") elif length > 1 then error("Multiple Custom Endpoint sources found — filter by .name") else .[0].id end') \
  || { echo "Failed to resolve Custom Endpoint source ID"; exit 1; }

if [ -z "$SOURCE_ID" ]; then echo "SOURCE_ID is empty — check topology output"; exit 1; fi

# PowerShell — extract sourceId for Custom Endpoint (fails clearly if multiple exist)
$topology = az rest --method GET `
  --url "https://api.fabric.microsoft.com/v1/workspaces/$WORKSPACE_ID/eventstreams/$EVENTSTREAM_ID/topology" `
  --resource "https://api.fabric.microsoft.com" | ConvertFrom-Json
$customSources = @($topology.sources | Where-Object { $_.type -eq 'CustomEndpoint' })
if ($customSources.Count -eq 0) { throw "No Custom Endpoint sources found in this Eventstream" }
if ($customSources.Count -gt 1) { throw "Multiple Custom Endpoint sources found. Filter by name: $($customSources.name -join ', ')" }
$sourceId = $customSources[0].id

Step 2: Get the Connection Details

⚠️ Security: This endpoint returns access keys and connection strings. Get explicit user confirmation before calling it. Redact primaryKey, secondaryKey, primaryConnectionString, and secondaryConnectionString from any displayed output unless the user explicitly asks for secret values in a secure context. Avoid logging raw credentials; store securely and rotate as needed.

az rest --method GET \
  --url "https://api.fabric.microsoft.com/v1/workspaces/${WORKSPACE_ID}/eventstreams/${EVENTSTREAM_ID}/sources/${SOURCE_ID}/connection" \
  --resource "https://api.fabric.microsoft.com"

az rest --method GET `
  --url "https://api.fabric.microsoft.com/v1/workspaces/$WORKSPACE_ID/eventstreams/$EVENTSTREAM_ID/sources/$sourceId/connection" `
  --resource "https://api.fabric.microsoft.com" | ConvertFrom-Json

Expected Response

{
  "fullyQualifiedNamespace": "namespace.servicebus.windows.net",
  "eventHubName": "es_<guid>",
  "accessKeys": {
    "primaryKey": "...",
    "secondaryKey": "...",
    "primaryConnectionString": "Endpoint=sb://namespace.servicebus.windows.net/;...",
    "secondaryConnectionString": "..."
  }
}

Kafka Producer Configuration

Use the response to configure a Kafka producer:

| Setting | Value | |---------|-------| | bootstrap_servers | {fullyQualifiedNamespace}:9093 | | topic | {eventHubName} | | security_protocol | SASL_SSL | | sasl_mechanism | PLAIN | | sasl_plain_username | $ConnectionString (fixed literal — not a variable) | | sasl_plain_password | {primaryConnectionString} |

Limitation: The /connection endpoint is only supported for Custom Endpoint sources (returns Kafka/Event Hub credentials). Other source types (Event Hub, IoT Hub, etc.) store their connection configuration (e.g., dataConnectionId, consumerGroup) directly in the decoded definition properties.

---

Validate Eventstream Configuration

Check key configuration aspects of a decoded Eventstream topology:

Source Validation Checklist

| Check | How | |-------|-----| | Source type is API-supported | Compare against 25 known type enums | | Cloud connection exists | Verify dataConnectionId GUID resolves | | Consumer group set | Required for Event Hub, IoT Hub, Kafka sources | | Serialization matches source | inputSerialization.type = Json, Csv, or Avro |

Destination Validation Checklist

| Check | How | |-------|-----| | Destination type is valid | Must be Lakehouse, Eventhouse, Activator, or CustomEndpoint | | Target item accessible | Verify workspaceId + itemId resolve via GET | | Input wired | inputNodes array must not be empty | | Eventhouse direct ingestion | connectionName and mappingRuleName set |

EventstreamProperties Validation

Decode eventstreamProperties.json and check:

• retentionTimeInDays is within 1–90
• eventThroughputLevel is Low, Medium, or High

---

Gotchas, Rules, Troubleshooting

MUST DO

• Always pass --resource https://api.fabric.microsoft.com with az rest calls
• Always use JMESPath filtering to resolve workspace name → ID and item name → ID
• Always base64-decode the definition payload before inspecting topology (not needed for the Topology API — that returns JSON directly)
• For Custom Endpoint connection details, use the Topology API — POST .../getDefinition returns empty properties; call GET .../topology to get the sourceId, then GET .../sources/{sourceId}/connection
• Use POST for definition endpoints — POST .../getDefinition (not GET), POST .../updateDefinition (not PUT). See official docs.
• Handle pagination — check for continuationUri in list responses
• Poll LRO responses — Get Definition may return 202 Accepted

PREFER

• Decode topology JSON into structured output for readable summaries
• Use jq (bash) or ConvertFrom-Json (PowerShell) for parsing
• Validate configurations before reporting issues to users
• Cross-reference destinations with downstream skills (eventhouse, sqldw, spark)

AVOID

• Do NOT confuse Eventstream with Eventhouse — they are separate Fabric workloads
• Do NOT hardcode workspace or item IDs — always discover them via the API
• Do NOT assume all source types appear in API enums — preview sources exist only in the UI
• Do NOT modify Eventstream topology with this consumption skill — use eventstream-authoring-cli for writes
• Do NOT attempt to query event data through the Eventstream API — use downstream skills (eventhouse-consumption-cli, sqldw-consumption-cli) for querying landed data

---

Examples

Platform note — examples use PowerShell. Always write the JSON body to a temp file via [IO.File]::WriteAllText() (no BOM) and pass --body "@$file" to az rest, rather than inline --body "..." which cmd.exe can mangle. Use -Compress with ConvertTo-Json to avoid newline issues. The one safe inline exception is --body '{}' for empty bodies. For large workspaces, check for continuationUri in list responses to handle pagination.

Example 1: List All Eventstreams in a Workspace

Prompt: "List all Eventstreams in my analytics workspace showing their names and IDs."

# 1. Discover workspace ID
$wsId = (az rest --method get `
  --url "https://api.fabric.microsoft.com/v1/workspaces" `
  --resource "https://api.fabric.microsoft.com" `
  --query "value[?displayName=='analytics'] | [0].id" -o tsv)
if (-not $wsId) { throw "Workspace 'analytics' not found" }

# 2. List Eventstreams (handles pagination)
$allItems = @()
$resp = az rest --method get `
  --url "https://api.fabric.microsoft.com/v1/workspaces/$wsId/eventstreams" `
  --resource "https://api.fabric.microsoft.com" | ConvertFrom-Json
$allItems += $resp.value
while ($resp.continuationUri) {
    $resp = az rest --method get `
      --url $resp.continuationUri `
      --resource "https://api.fabric.microsoft.com" | ConvertFrom-Json
    $allItems += $resp.value
}
$allItems | Select-Object displayName, id, description | Format-Table

Example 2: Inspect Eventstream Topology

Prompt: "Show me the topology of my SensorIngestion Eventstream — all sources, operators, and destinations."

# 1. Discover workspace + Eventstream IDs
$wsId = (az rest --method get `
  --url "https://api.fabric.microsoft.com/v1/workspaces" `
  --resource "https://api.fabric.microsoft.com" `
  --query "value[?displayName=='analytics'] | [0].id" -o tsv)
if (-not $wsId) { throw "Workspace 'analytics' not found" }

$esId = (az rest --method get `
  --url "https://api.fabric.microsoft.com/v1/workspaces/$wsId/eventstreams" `
  --resource "https://api.fabric.microsoft.com" `
  --query "value[?displayName=='SensorIngestion'] | [0].id" -o tsv)
if (-not $esId) { throw "Eventstream 'SensorIngestion' not found" }

# 2. Get topology (returns JSON directly — no base64 decoding needed)
$topo = az rest --method get `
  --url "https://api.fabric.microsoft.com/v1/workspaces/$wsId/eventstreams/$esId/topology" `
  --resource "https://api.fabric.microsoft.com" | ConvertFrom-Json

# 3. Summarize nodes (filter nulls from arrays)
Write-Host "Sources:"
@($topo.sources | Where-Object { $_ -ne $null }) | ForEach-Object {
    Write-Host "  - $($_.name) (type: $($_.type), id: $($_.id))"
}
Write-Host "Operators:"
@($topo.operators | Where-Object { $_ -ne $null }) | ForEach-Object {
    Write-Host "  - $($_.name) (type: $($_.type))"
}
Write-Host "Destinations:"
@($topo.destinations | Where-Object { $_ -ne $null }) | ForEach-Object {
    Write-Host "  - $($_.name) (type: $($_.type))"
}
Write-Host "Streams:"
@($topo.streams | Where-Object { $_ -ne $null }) | ForEach-Object {
    Write-Host "  - $($_.name) (type: $($_.type))"
}

Example 3: Check Retention and Throughput Settings

Prompt: "What are the retention and throughput settings for my SensorIngestion Eventstream?"

Note: Retention and throughput settings are stored in the eventstreamProperties.json part of the definition (not eventstream.json). If this part is absent, the Eventstream uses platform defaults.

# 1. Discover workspace + Eventstream IDs
$wsId = (az rest --method get `
  --url "https://api.fabric.microsoft.com/v1/workspaces" `
  --resource "https://api.fabric.microsoft.com" `
  --query "value[?displayName=='analytics'] | [0].id" -o tsv)
if (-not $wsId) { throw "Workspace 'analytics' not found" }

$esId = (az rest --method get `
  --url "https://api.fabric.microsoft.com/v1/workspaces/$wsId/items?type=Eventstream" `
  --resource "https://api.fabric.microsoft.com" `
  --query "value[?displayName=='SensorIngestion'] | [0].id" -o tsv)
if (-not $esId) { throw "Eventstream 'SensorIngestion' not found" }

# 2. Get definition (handles LRO if API returns 202)
$respJson = az rest --method post `
  --url "https://api.fabric.microsoft.com/v1/workspaces/$wsId/eventstreams/$esId/getDefinition" `
  --resource "https://api.fabric.microsoft.com" `
  --headers "Content-Type=application/json" `
  --body '{}'
if ($LASTEXITCODE -ne 0 -or -not $respJson) { throw "getDefinition request failed" }
$resp = $respJson | ConvertFrom-Json

if ($resp.definition) {
    $def = $resp  # Synchronous — got definition immediately
} elseif ($resp.operationId) {
    # Asynchronous (LRO) — poll operation status
    $def = $null
    for ($i = 0; $i -lt 12; $i++) {
        Start-Sleep -Seconds 5
        $status = (az rest --method get `
          --url "https://api.fabric.microsoft.com/v1/operations/$($resp.operationId)" `
          --resource "https://api.fabric.microsoft.com" | ConvertFrom-Json)
        if ($status.status -eq 'Succeeded') {
            $def = (az rest --method get `
              --url "https://api.fabric.microsoft.com/v1/operations/$($resp.operationId)/result" `
              --resource "https://api.fabric.microsoft.com" | ConvertFrom-Json)
            break
        } elseif ($status.status -in @('Failed', 'Cancelled')) {
            throw "getDefinition LRO $($status.status): $($status.error.message)"
        }
    }
    if (-not $def.definition) { throw "getDefinition LRO timed out (last status: $($status.status))" }
} else {
    throw "getDefinition returned neither a definition nor an operationId"
}

# 3. Decode eventstreamProperties.json part (holds retention + throughput)
$propsPart = $def.definition.parts | Where-Object { $_.path -eq 'eventstreamProperties.json' } | Select-Object -First 1
if ($propsPart) {
    $props = [Text.Encoding]::UTF8.GetString(
      [Convert]::FromBase64String($propsPart.payload)) | ConvertFrom-Json
    Write-Host "Retention: $($props.retentionTimeInDays) days"
    Write-Host "Throughput Level: $($props.eventThroughputLevel)"
} else {
    # Fall back to topology-level properties (older format)
    $esPart = $def.definition.parts | Where-Object { $_.path -eq 'eventstream.json' } | Select-Object -First 1
    if (-not $esPart) { throw "eventstream.json part not found in definition" }
    $topology = [Text.Encoding]::UTF8.GetString(
      [Convert]::FromBase64String($esPart.payload)) | ConvertFrom-Json
    if ($topology.properties.retentionTimeInDays) {
        Write-Host "Retention: $($topology.properties.retentionTimeInDays) days"
    } else {
        Write-Host "Retention: (platform default — not explicitly configured)"
    }
    if ($topology.properties.eventThroughputLevel) {
        Write-Host "Throughput Level: $($topology.properties.eventThroughputLevel)"
    } else {
        Write-Host "Throughput Level: (platform default — not explicitly configured)"
    }
}

Example 4: Get Custom Endpoint Connection Metadata

Prompt: "Get the Kafka connection metadata for the Custom Endpoint source in my SensorIngestion Eventstream."

Security: The connection endpoint returns access keys. Get user confirmation before calling it and avoid logging raw credentials.

# 1. Discover workspace + Eventstream IDs (omitted for brevity)

# 2. Get topology to find Custom Endpoint source ID
$topo = az rest --method get `
  --url "https://api.fabric.microsoft.com/v1/workspaces/$wsId/eventstreams/$esId/topology" `
  --resource "https://api.fabric.microsoft.com" | ConvertFrom-Json

$ceSource = @($topo.sources | Where-Object { $_.type -eq 'CustomEndpoint' }) | Select-Object -First 1
if (-not $ceSource) { throw "No CustomEndpoint source found in this Eventstream" }
$sourceId = $ceSource.id

# 3. Get connection metadata
$conn = az rest --method get `
  --url "https://api.fabric.microsoft.com/v1/workspaces/$wsId/eventstreams/$esId/sources/$sourceId/connection" `
  --resource "https://api.fabric.microsoft.com" | ConvertFrom-Json

Write-Host "Fully Qualified Namespace: $($conn.fullyQualifiedNamespace)"
Write-Host "Event Hub Name:            $($conn.eventHubName)"
Write-Host "Kafka Bootstrap Server:    $($conn.fullyQualifiedNamespace):9093"

Example 5: Validate Eventstream Configuration

Prompt: "Check if my SensorIngestion Eventstream has any configuration issues."

# 1. Discover workspace + Eventstream IDs (omitted for brevity)

# 2. Get topology
$topo = az rest --method get `
  --url "https://api.fabric.microsoft.com/v1/workspaces/$wsId/eventstreams/$esId/topology" `
  --resource "https://api.fabric.microsoft.com" | ConvertFrom-Json

# 3. Validate sources
@($topo.sources | Where-Object { $_ -ne $null }) | ForEach-Object {
    $src = $_
    Write-Host "Source: $($src.name) (type: $($src.type))"
    if ($src.properties.dataConnectionId) {
        Write-Host "  Cloud connection: $($src.properties.dataConnectionId)"
    }
    if ($src.type -in @('AzureEventHub','AzureEventHubExtended','AzureIoTHub','ConfluentCloud','ApacheKafka','AmazonMSKKafka') -and
        -not $src.properties.consumerGroupName) {
        Write-Host "  WARNING: No consumer group set (required for $($src.type))"
    }
}

# 4. Validate destinations
@($topo.destinations | Where-Object { $_ -ne $null }) | ForEach-Object {
    $dst = $_
    Write-Host "Destination: $($dst.name) (type: $($dst.type))"
    if (-not $dst.inputNodes -or $dst.inputNodes.Count -eq 0) {
        Write-Host "  WARNING: No input wired — destination will receive no events"
    }
    if ($dst.type -eq 'Eventhouse' -and -not $dst.properties.tableName) {
        Write-Host "  WARNING: No target table configured"
    }
    if ($dst.type -eq 'Eventhouse' -and $dst.properties.dataIngestionMode -eq 'DirectIngestion') {
        if (-not $dst.properties.connectionName -or -not $dst.properties.mappingRuleName) {
            Write-Host "  WARNING: DirectIngestion requires connectionName and mappingRuleName"
        }
    }
}

# 5. Check node count limits
$ceCount = @($topo.sources | Where-Object { $_.type -eq 'CustomEndpoint' }).Count +
           @($topo.destinations | Where-Object { $_.type -eq 'CustomEndpoint' }).Count +
           @($topo.destinations | Where-Object { $_.type -eq 'Eventhouse' -and
             $_.properties.dataIngestionMode -eq 'DirectIngestion' }).Count
Write-Host "CustomEndpoint + DI count: $ceCount / 11 limit"
if ($ceCount -gt 11) {
    Write-Host "WARNING: Exceeds limit of 11 CustomEndpoint + DirectIngestion nodes"
}