eventstream-authoring-cli

检查更新 — 每个会话仅执行一次（必须） 首次在会话中使用此技能时，继续操作前先运行 check-updates 技能。

• GitHub Copilot CLI / VS Code：调用 check-updates 技能。
• Claude Code / Cowork / Cursor / Windsurf / Codex：比较本地与远程的 package.json 版本。
• 如果本会话中此前已执行过检查，则跳过。

重要说明

1. 若要通过工作区名称查找工作区详情（包括其 ID）：列出所有工作区，然后使用 JMESPath 进行筛选
2. 若要通过工作区 ID、项目类型和项目名称查找项目详情（包括其 ID）：列出该工作区中该类型的所有项目，然后使用 JMESPath 进行筛选
3. Eventstream ≠ Eventhouse。Eventstream 是用于实时事件引入和路由的管道。对于 KQL 数据库操作，请使用 eventhouse-authoring-cli 或 eventhouse-consumption-cli。

Eventstream 创作 — CLI 技能

目录

任务	参考	备注
在 Fabric 中查找工作区和项目	COMMON-CLI.md § Finding Workspaces and Items in Fabric	必读——请先阅读链接内容 [用于按名称查找工作区 ID，或按名称、项目类型和工作区 ID 查找项目 ID]
Fabric 拓扑和关键概念	COMMON-CORE.md § Fabric Topology & Key Concepts
环境 URL	COMMON-CORE.md § Environment URLs
身份验证和令牌获取	COMMON-CORE.md § Authentication & Token Acquisition	受众错误会导致 401；遇到任何身份验证问题前请先阅读
核心控制平面 REST API	COMMON-CORE.md § Core Control-Plane REST APIs	包含分页、LRO 轮询和速率限制模式
注意事项、最佳实践和故障排查	COMMON-CORE.md § Gotchas, Best Practices & Troubleshooting
工具选择依据	COMMON-CLI.md § Tool Selection Rationale
身份验证方法	COMMON-CLI.md § Authentication Recipes	az login 流程和令牌获取
通过 az rest 使用 Fabric 控制平面 API	COMMON-CLI.md § Fabric Control-Plane API via az rest	始终传入 --resource；包含分页和 LRO 辅助工具
注意事项和故障排查（特定于 CLI）	COMMON-CLI.md § Gotchas & Troubleshooting (CLI-Specific)	az rest 受众、Shell 转义、令牌过期
快速参考	COMMON-CLI.md § Quick Reference	az rest 模板和令牌受众/工具矩阵
Eventstream 资源模型	EVENTSTREAM-AUTHORING-CORE.md § Eventstream Resource Model	请先阅读——由源、运算符、流和目标组成的基于图的拓扑
源配置	EVENTSTREAM-AUTHORING-CORE.md § Source Configuration	25 种 API 支持的源类型及各自属性
转换运算符	EVENTSTREAM-AUTHORING-CORE.md § Transformation Operators	8 种运算符类型：Filter、Aggregate、GroupBy、Join、ManageFields、Union、Expand、SQL
目标配置	EVENTSTREAM-AUTHORING-CORE.md § Destination Configuration	4 种 API 支持的目标类型及节点架构
流类型	EVENTSTREAM-AUTHORING-CORE.md § Stream Types	DefaultStream（自动）和 DerivedStream（由运算符生成）
Eventstream 生命周期（REST API）	EVENTSTREAM-AUTHORING-CORE.md § Eventstream Lifecycle (REST API)	CRUD 和 Definition 端点
项目定义和部署	EVENTSTREAM-AUTHORING-CORE.md § Item Definitions and Deployment	eventstream.json 的 Base64 编码模式
注意事项和限制	EVENTSTREAM-AUTHORING-CORE.md § Gotchas and Limitations	最多 11 个自定义端点、Base64 编码、命名限制
创建 Eventstream	SKILL.md § Create an Eventstream
部署完整拓扑	SKILL.md § Deploy Full Topology	端到端流程：构建拓扑 JSON → Base64 编码 → 提交定义
更新 Eventstream 拓扑	SKILL.md § Update Eventstream Topology
删除 Eventstream	SKILL.md § Delete an Eventstream
注意事项、规则和故障排查	SKILL.md § Gotchas, Rules, Troubleshooting	必须执行/避免/优先选择检查清单

---

创建 Eventstream

创建一个空的 Eventstream 项目，然后通过定义 API 配置其源、目标和运算符。

步骤 1：创建项目

az rest --method POST \
  --url "https://api.fabric.microsoft.com/v1/workspaces/${WORKSPACE_ID}/eventstreams" \
  --resource "https://api.fabric.microsoft.com" \
  --headers "Content-Type=application/json" \
  --body '{"displayName": "my-eventstream", "description": "IoT sensor pipeline"}'

将返回的 id 保存为 EVENTSTREAM_ID。

步骤 2：构建拓扑

构建包含源、流、运算符和目标的 eventstream.json 拓扑。每个节点都通过 inputNodes 引用其上游节点。

建议以编程方式构建 JSON，以避免序列化错误。关键规则：

• 拓扑必须且只能有一个 DefaultStream — 所有源都通过 inputNodes 接入该流
• 运算符通过 inputNodes[].name 引用其输入
• DerivedStreams 必须在 properties 中包含 inputSerialization
• 目标引用其输入流或运算符

步骤 3：部署定义

对拓扑 JSON 进行 Base64 编码，并通过定义 API 提交。完整的请求体结构请参见项目定义和部署。

---

部署完整拓扑

若要通过一次 API 调用部署包含拓扑的完整 Eventstream，请使用 Create Item with Definition 端点：

# 1. Build eventstream.json content (topology)
TOPOLOGY_JSON='{"compatibilityLevel":"1.1","sources":[...],"streams":[...],"operators":[...],"destinations":[...]}'

# 2. Build eventstreamProperties.json (optional — controls retention and throughput)
PROPERTIES_JSON='{"retentionTimeInDays":1,"eventThroughputLevel":"Low"}'

# 3. Base64-encode both (no line wraps)
TOPOLOGY_B64=$(echo -n "$TOPOLOGY_JSON" | base64 -w 0)
PROPERTIES_B64=$(echo -n "$PROPERTIES_JSON" | base64 -w 0)

# 4. Submit via Items API
az rest --method POST \
  --url "https://api.fabric.microsoft.com/v1/workspaces/${WORKSPACE_ID}/items" \
  --resource "https://api.fabric.microsoft.com" \
  --headers "Content-Type=application/json" \
  --body "{
    \"displayName\": \"my-eventstream\",
    \"type\": \"Eventstream\",
    \"definition\": {
      \"parts\": [
        {
          \"path\": \"eventstream.json\",
          \"payload\": \"${TOPOLOGY_B64}\",
          \"payloadType\": \"InlineBase64\"
        },
        {
          \"path\": \"eventstreamProperties.json\",
          \"payload\": \"${PROPERTIES_B64}\",
          \"payloadType\": \"InlineBase64\"
        }
      ]
    }
  }"

注意： 如果省略 eventstreamProperties.json，API 会应用默认值：retentionTimeInDays: 1、eventThroughputLevel: "Low"。如需控制保留时间（1–90 天）和吞吐量，请显式包含该文件。

在 Windows（PowerShell）上，请使用 [Convert]::ToBase64String([Text.Encoding]::UTF8.GetBytes($json)) 进行 Base64 编码。

---

更新 Eventstream 拓扑

1. 获取当前定义：POST /v1/workspaces/{wsId}/eventstreams/{esId}/getDefinition
2. 解码 base64 编码的 eventstream.json 载荷
3. 修改拓扑（添加／移除／更新节点）
4. 重新编码为 base64
5. 提交：POST /v1/workspaces/{wsId}/eventstreams/{esId}/updateDefinition

API 注意事项：Eventstream 定义 API 使用带有动作动词（getDefinition、updateDefinition）的 POST，而不是对 /definition 资源使用 GET／PUT。这遵循 Fabric Items 定义模式。请参见官方文档。

对于长时间运行的操作，更新定义 API 会返回 202 Accepted。轮询 Location 标头中的 URL，直到操作完成。

添加筛选运算符

⚠️ 重要： 筛选运算符条件中的 column 和 value 使用嵌套对象，而不是裸字符串。若使用 "column": "temperature"，而不是下方的对象形式，API 将静默拒绝请求。

{
  "name": "FilterHighTemp",
  "type": "Filter",
  "inputNodes": [{"name": "my-stream"}],
  "properties": {
    "conditions": [{
      "column": {
        "node": null,
        "columnName": "temperature",
        "columnPath": null,
        "expressionType": "ColumnReference"
      },
      "operatorType": "GreaterThan",
      "value": {
        "dataType": "Float",
        "value": "30.0",
        "expressionType": "Literal"
      }
    }]
  }
}

所有运算符条件字段必须采用以下结构：

• column → 包含 {node, columnName, columnPath, expressionType: "ColumnReference"} 的对象
• value → 包含 {dataType, value, expressionType: "Literal"} 的对象
• operatorType → 字符串：Equals、NotEquals、GreaterThan、GreaterThanOrEquals、LessThan、LessThanOrEquals、Contains、DoesNotContain、StartsWith、DoesNotStartWith、EndsWith、DoesNotEndWith、IsEmpty、IsNull、IsNotNull、IsNotNullOrEmpty
• dataType → BigInt、Float、Nvarchar(max)、DateTime、Bit

所有引用列的运算符（Filter、Aggregate、GroupBy、Join、ManageFields）都采用相同的嵌套对象模式。

---

删除 Eventstream

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

成功时返回 200 OK。

---

注意事项、规则和故障排除

必须

• 提交定义前，始终对 eventstream.json 负载进行 Base64 编码
• 调用 az rest 时，始终传入 --resource https://api.fabric.microsoft.com
• 始终使用 JMESPath 筛选，将工作区名称解析为 ID，将项名称解析为 ID
• 始终使用嵌套对象引用运算符的列/值 — "column": {"columnName": "x", "expressionType": "ColumnReference", ...}，绝不要使用 "column": "x"（API 会静默拒绝裸字符串）
• 每个拓扑中只能有一个 DefaultStream — 所有源都连接到它（API 会拒绝多个 DefaultStream）
• 轮询 LRO 响应 — Update Definition 会返回带有 Location 标头的 202 Accepted

建议

• 以编程方式构建拓扑 JSON，而不是手动拼接字符串
• 测试和原型设计时使用 SampleData 源类型
• 显式设置 retentionTimeInDays，不要依赖默认值
• 在源配置中引用云连接之前，先验证连接
• 使用 DerivedStreams，让运算符的输出可在 Real-Time Hub 中使用

避免

• 不要在定义负载中使用原始 JSON — 必须进行 Base64 编码
• 不要在 Eventstream 显示名称中使用下划线或句点（会导致 SQL 运算符失效）
• 不要在用户定义的拓扑节点名称（源、运算符、DerivedStreams、目标）中使用连字符、下划线、句点或空格 — 只允许使用字母数字 PascalCase（例如，使用 FilterTemperature，而不是 filter-temperature 或 filter_temperature）。例外：DefaultStream 名称由平台自动生成为 {eventstreamName}-stream，可以包含连字符 — 不要重命名
• CustomEndpoint 源与 CustomEndpoint/Eventhouse 直接引入目标的总数不要超过 11 个
• 不要混淆 Eventstream 和 Eventhouse — 它们是 Fabric 中彼此独立的工作负载
• 不要硬编码工作区或项 ID — 始终通过 API 查找

---

示例

平台说明 — 示例使用 PowerShell。始终通过 [IO.File]::WriteAllText()（不含 BOM）将 JSON 正文写入临时文件，并向 az rest 传入 --body "@$file"，不要使用内联 --body "..."，因为 cmd.exe 可能会破坏内容。对 ConvertTo-Json 使用 -Compress，以避免换行问题。唯一安全的内联例外是用于空正文的 --body '{}'。

示例 1：创建包含源的 Eventstream

提示词：“在我的开发工作区中创建一个名为 SensorIngestion、包含示例数据源的 Eventstream。”

# 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=='dev'] | [0].id" -o tsv)
if (-not $wsId) { throw "Workspace 'dev' not found" }

# 2. Create empty Eventstream
$esBody = @{ displayName = "SensorIngestion"; description = "IoT sensor pipeline" } | ConvertTo-Json -Compress
$bodyFile = Join-Path ([IO.Path]::GetTempPath()) "es_create.json"
[IO.File]::WriteAllText($bodyFile, $esBody, [System.Text.UTF8Encoding]::new($false))
$created = az rest --method post `
  --url "https://api.fabric.microsoft.com/v1/workspaces/$wsId/eventstreams" `
  --resource "https://api.fabric.microsoft.com" `
  --headers "Content-Type=application/json" `
  --body "@$bodyFile" | ConvertFrom-Json

# 3. Get the created Eventstream ID from response
$esId = $created.id
if (-not $esId) { throw "Eventstream creation did not return an ID" }

# 4. Build topology — DefaultStream uses inputNodes (not parentName)
$topology = @{
    compatibilityLevel = "1.0"
    sources = @(@{
        name = "SampleSource"
        type = "SampleData"
        properties = @{ type = "Bicycles" }
    })
    streams = @(@{
        name = "SensorIngestion-stream"
        type = "DefaultStream"
        properties = @{}
        inputNodes = @(@{ name = "SampleSource" })
    })
    operators = @()
    destinations = @()
}
$topologyJson = $topology | ConvertTo-Json -Depth 10 -Compress
$topologyB64 = [Convert]::ToBase64String([Text.Encoding]::UTF8.GetBytes($topologyJson))

# 5. Deploy definition
$defBody = @{
    definition = @{
        parts = @(@{
            path = "eventstream.json"
            payload = $topologyB64
            payloadType = "InlineBase64"
        })
    }
} | ConvertTo-Json -Depth 5 -Compress
$defFile = Join-Path ([IO.Path]::GetTempPath()) "es_def.json"
[IO.File]::WriteAllText($defFile, $defBody, [System.Text.UTF8Encoding]::new($false))
# updateDefinition returns 202 Accepted (LRO). Use Invoke-WebRequest to capture headers.
$token = (az account get-access-token --resource "https://api.fabric.microsoft.com" --query accessToken -o tsv)
$ps5 = @{}; if ($PSVersionTable.PSVersion.Major -lt 6) { $ps5.UseBasicParsing = $true }
$response = Invoke-WebRequest @ps5 -Method Post `
  -Uri "https://api.fabric.microsoft.com/v1/workspaces/$wsId/eventstreams/$esId/updateDefinition" `
  -Headers @{ Authorization = "Bearer $token" } `
  -ContentType "application/json; charset=utf-8" `
  -Body (Get-Content $defFile -Raw -Encoding UTF8)

if ($response.StatusCode -eq 202) {
    $location = $response.Headers['Location']
    if ($location -is [array]) { $location = $location[0] }
    if (-not $location) { throw "LRO response missing Location header" }
    $ra = $response.Headers['Retry-After']
    if ($ra -is [array]) { $ra = $ra[0] }
    $retryAfter = if ($ra) { [int]$ra } else { 5 }
    $succeeded = $false
    for ($i = 0; $i -lt 12; $i++) {
        Start-Sleep -Seconds $retryAfter
        $poll = Invoke-RestMethod -Uri $location -Headers @{ Authorization = "Bearer $token" }
        if ($poll.status -eq 'Succeeded') { $succeeded = $true; Write-Host "Update succeeded"; break }
        elseif ($poll.status -in @('Failed', 'Cancelled')) {
            throw "updateDefinition LRO $($poll.status): $($poll.error.message)"
        }
    }
    if (-not $succeeded) { throw "updateDefinition LRO timed out" }
}

示例 2：添加包含 DerivedStream 的筛选运算符

提示词：“在我的 SensorIngestion Eventstream 中添加筛选器，仅保留 No_Bikes > 5 的事件，并将筛选后的输出作为 DerivedStream 暴露。”

重要：仅添加 Filter 运算符节点不会重定向 DefaultStream。 要让筛选后的输出可供使用，请通过 inputNodes 将 DerivedStream（或目标）连接到 筛选器的输出。

# 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=='dev'] | [0].id" -o tsv)
if (-not $wsId) { throw "Workspace 'dev' 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 current definition (handles LRO via Location header)
$token = (az account get-access-token --resource "https://api.fabric.microsoft.com" --query accessToken -o tsv)
$ps5 = @{}; if ($PSVersionTable.PSVersion.Major -lt 6) { $ps5.UseBasicParsing = $true }
$response = Invoke-WebRequest @ps5 -Method Post `
  -Uri "https://api.fabric.microsoft.com/v1/workspaces/$wsId/eventstreams/$esId/getDefinition" `
  -Headers @{ Authorization = "Bearer $token"; "Content-Type" = "application/json" } `
  -Body '{}'

if ($response.StatusCode -eq 202) {
    $location = $response.Headers['Location']
    if ($location -is [array]) { $location = $location[0] }
    if (-not $location) { throw "LRO response missing Location header" }
    $ra = $response.Headers['Retry-After']
    if ($ra -is [array]) { $ra = $ra[0] }
    $retryAfter = if ($ra) { [int]$ra } else { 5 }
    $def = $null
    for ($i = 0; $i -lt 12; $i++) {
        Start-Sleep -Seconds $retryAfter
        $poll = Invoke-RestMethod -Uri $location -Headers @{ Authorization = "Bearer $token" }
        if ($poll.status -eq 'Succeeded') {
            $def = Invoke-RestMethod -Uri "$location/result" `
              -Headers @{ Authorization = "Bearer $token" }
            break
        } elseif ($poll.status -in @('Failed', 'Cancelled')) {
            throw "getDefinition LRO $($poll.status): $($poll.error.message)"
        }
    }
    if (-not $def -or -not $def.definition) { throw "getDefinition LRO timed out (last status: $(if ($poll) { $poll.status } else { 'unknown' }))" }
} else {
    $def = $response.Content | ConvertFrom-Json
}

# 3. Decode existing topology
$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

# 4. Add Filter operator (PascalCase name — no underscores or hyphens)
#    Column: expressionType + columnName; Value: expressionType + dataType + value
$filter = @{
    name = "FilterLowBikes"
    type = "Filter"
    inputNodes = @(@{ name = "SensorIngestion-stream" })
    properties = @{
        conditions = @(@{
            operatorType = "GreaterThan"
            column = @{
                expressionType = "ColumnReference"
                node = $null
                columnName = "No_Bikes"
                columnPath = $null
            }
            value = @{
                expressionType = "Literal"
                dataType = "BigInt"
                value = "5"
            }
        })
    }
}
$existingOps = @($topology.operators | Where-Object { $_ -ne $null })
$topology.operators = $existingOps + @($filter)

# 5. Add DerivedStream wired to filter output (makes filtered data available)
$derivedStream = @{
    name = "FilteredOutput"
    type = "DerivedStream"
    properties = @{
        inputSerialization = @{ type = "Json"; properties = @{ encoding = "UTF8" } }
    }
    inputNodes = @(@{ name = "FilterLowBikes" })
}
$existingStreams = @($topology.streams | Where-Object { $_ -ne $null })
$topology.streams = $existingStreams + @($derivedStream)

# 6. Re-encode and update
$topologyJson = $topology | ConvertTo-Json -Depth 10 -Compress
$topologyB64 = [Convert]::ToBase64String([Text.Encoding]::UTF8.GetBytes($topologyJson))
$esPart.payload = $topologyB64

$defBody = @{ definition = @{ parts = $def.definition.parts } } | ConvertTo-Json -Depth 5 -Compress
$defFile = Join-Path ([IO.Path]::GetTempPath()) "es_def.json"
[IO.File]::WriteAllText($defFile, $defBody, [System.Text.UTF8Encoding]::new($false))
# updateDefinition returns 202 Accepted (LRO). Use Invoke-WebRequest to capture headers.
$ps5 = @{}; if ($PSVersionTable.PSVersion.Major -lt 6) { $ps5.UseBasicParsing = $true }
$response = Invoke-WebRequest @ps5 -Method Post `
  -Uri "https://api.fabric.microsoft.com/v1/workspaces/$wsId/eventstreams/$esId/updateDefinition" `
  -Headers @{ Authorization = "Bearer $token" } `
  -ContentType "application/json; charset=utf-8" `
  -Body (Get-Content $defFile -Raw -Encoding UTF8)

if ($response.StatusCode -eq 202) {
    $location = $response.Headers['Location']
    if ($location -is [array]) { $location = $location[0] }
    if (-not $location) { throw "LRO response missing Location header" }
    $ra = $response.Headers['Retry-After']
    if ($ra -is [array]) { $ra = $ra[0] }
    $retryAfter = if ($ra) { [int]$ra } else { 5 }
    $succeeded = $false
    for ($i = 0; $i -lt 12; $i++) {
        Start-Sleep -Seconds $retryAfter
        $poll = Invoke-RestMethod -Uri $location -Headers @{ Authorization = "Bearer $token" }
        if ($poll.status -eq 'Succeeded') { $succeeded = $true; Write-Host "Update succeeded"; break }
        elseif ($poll.status -in @('Failed', 'Cancelled')) {
            throw "updateDefinition LRO $($poll.status): $($poll.error.message)"
        }
    }
    if (-not $succeeded) { throw "updateDefinition LRO timed out" }
}

示例 3：部署完整拓扑（以内联定义创建）

提示词：“创建一个名为 EventPipeline 的完整 Eventstream，其中包含 Custom Endpoint 源、高价值事件筛选器，以及用于筛选后输出的 DerivedStream。”

说明：此示例使用 Fabric Items API（POST /items）在单次调用中创建包含定义的 Eventstream，而不是先创建再更新。

# 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=='dev'] | [0].id" -o tsv)
if (-not $wsId) { throw "Workspace 'dev' not found" }

# 2. Build complete topology with filter + DerivedStream
$topology = @{
    compatibilityLevel = "1.0"
    sources = @(@{
        name = "CustomSource"
        type = "CustomEndpoint"
        properties = @{}
    })
    streams = @(
        @{
            name = "EventPipeline-stream"
            type = "DefaultStream"
            properties = @{}
            inputNodes = @(@{ name = "CustomSource" })
        }
        @{
            name = "FilteredEvents"
            type = "DerivedStream"
            properties = @{
                inputSerialization = @{ type = "Json"; properties = @{ encoding = "UTF8" } }
            }
            inputNodes = @(@{ name = "FilterPremium" })
        }
    )
    operators = @(@{
        name = "FilterPremium"
        type = "Filter"
        inputNodes = @(@{ name = "EventPipeline-stream" })
        properties = @{
            conditions = @(@{
                operatorType = "GreaterThan"
                column = @{
                    expressionType = "ColumnReference"
                        node = $null
                        columnName = "Amount"
                        columnPath = $null
                    }
                value = @{
                    expressionType = "Literal"
                    dataType = "BigInt"
                    value = "100"
                }
            })
        }
    })
    destinations = @()
}

# 3. Create with inline definition (single API call)
$topologyJson = $topology | ConvertTo-Json -Depth 10 -Compress
$topologyB64 = [Convert]::ToBase64String([Text.Encoding]::UTF8.GetBytes($topologyJson))

$body = @{
    displayName = "EventPipeline"
    type = "Eventstream"
    definition = @{
        parts = @(@{
            path = "eventstream.json"
            payload = $topologyB64
            payloadType = "InlineBase64"
        })
    }
} | ConvertTo-Json -Depth 5 -Compress
$bodyFile = Join-Path ([IO.Path]::GetTempPath()) "es_create_full.json"
[IO.File]::WriteAllText($bodyFile, $body, [System.Text.UTF8Encoding]::new($false))

# Create-with-definition returns 202 Accepted (LRO). Use Invoke-WebRequest to capture headers.
$token = (az account get-access-token --resource "https://api.fabric.microsoft.com" --query accessToken -o tsv)
$ps5 = @{}; if ($PSVersionTable.PSVersion.Major -lt 6) { $ps5.UseBasicParsing = $true }
$response = Invoke-WebRequest @ps5 -Method Post `
  -Uri "https://api.fabric.microsoft.com/v1/workspaces/$wsId/items" `
  -Headers @{ Authorization = "Bearer $token" } `
  -ContentType "application/json; charset=utf-8" `
  -Body (Get-Content $bodyFile -Raw -Encoding UTF8)

if ($response.StatusCode -eq 202) {
    $location = $response.Headers['Location']
    if ($location -is [array]) { $location = $location[0] }
    if (-not $location) { throw "LRO response missing Location header" }
    $ra = $response.Headers['Retry-After']
    if ($ra -is [array]) { $ra = $ra[0] }
    $retryAfter = if ($ra) { [int]$ra } else { 10 }
    $succeeded = $false
    for ($i = 0; $i -lt 12; $i++) {
        Start-Sleep -Seconds $retryAfter
        $poll = Invoke-RestMethod -Uri $location -Headers @{ Authorization = "Bearer $token" }
        if ($poll.status -eq 'Succeeded') { $succeeded = $true; Write-Host "Create succeeded"; break }
        elseif ($poll.status -in @('Failed', 'Cancelled')) {
            throw "Create LRO $($poll.status): $($poll.error.message)"
        }
    }
    if (-not $succeeded) { throw "Create LRO timed out" }
} else {
    Write-Host "Created: $(($response.Content | ConvertFrom-Json).displayName)"
}

示例 4：删除 Eventstream

提示词：“从我的开发工作区中删除 SensorIngestion Eventstream。”

# 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=='dev'] | [0].id" -o tsv)
if (-not $wsId) { throw "Workspace 'dev' 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. Delete
az rest --method delete `
  --url "https://api.fabric.microsoft.com/v1/workspaces/$wsId/eventstreams/$esId" `
  --resource "https://api.fabric.microsoft.com"