关于 dataflows-consumption-cli
dataflows-consumption-cli 是面向 Fabric Dataflows Gen2 的只读命令行技能，供 Agent 探索数据流、参数、定义和刷新作业。它以 az rest 为主要接口，并提供查询与排查流程。
主要功能
- 01列出工作区和数据流
- 02发现参数与数据连接
- 03解码并检查数据流定义
- 04查询刷新历史与作业状态
- 05执行查询并可视化结果
适用场景
- 01盘点工作区中的数据流
- 02检查 Power Query M 代码和连接
- 03排查数据流刷新失败
- 04分析查询结果并生成图表
MCP 相关
这是一个命令行技能，而非 MCP 服务；内容指导 Agent 使用 az、curl、jq 等工具访问 Fabric REST API。
安装
复制下面的命令，在终端中运行。
npx skills add https:/安装后在 Claude Code 中输入/dataflows-consumption-cli确认已加载。
包含的文件
除 SKILL.md 外，安装时一起下载的参考资料和脚本。
references/chart-visualization.mdconsumption-cli-quickref.mddiscovery-queries.mdscript-templates.mdreferences/charts/bar_chart.pyline_chart.py详细介绍
更新检查——每个会话仅检查一次（必须执行） 首次在一个会话中使用此技能时，必须先运行 check-updates 技能，然后再继续。
- GitHub Copilot CLI / VS Code：调用
check-updates技能。- Claude Code / Cowork / Cursor / Windsurf / Codex：比较本地与远程的 package.json 版本。
- 如果本会话中已执行过检查，则跳过。
重要说明
- 若要根据工作区名称查找工作区详情（包括其 ID）：先列出所有工作区，然后使用 JMESPath 进行筛选
- 若要根据名称查找数据流：列出工作区中的所有数据流，并在客户端按
displayName筛选——服务端不支持按名称筛选getDefinition是 POST，而不是 GET——即使它读取的是数据
范围界定——只读（必须遵守） 此技能严格只读。绝不能创建、更新或删除数据流或其定义——这意味着不得调用
Delete Dataflow、Create Dataflow、updateDefinition，也不得进行其他会修改或破坏数据的调用（例如针对数据流执行az rest --method delete/put/patch，或执行会创建／覆盖数据的 POST）。 仅允许使用下文明确说明的只读 POST 操作getDefinition和executeQuery。 如果用户要求删除、创建、修改或持久化数据流，必须在最终回复中明确拒绝该修改操作，并仅将dataflows-authoring-cli列为后续交接选项。不要调用该技能、不要为写入操作预先执行发现，也不要在同一轮中继续写入工作流。 只有在用户明确确认单独的创作请求后，才能开始修改操作。
dataflows-consumption-cli — 通过 CLI 使用 Dataflows Gen2
目录
工具栈
| 工具 | 作用 | 安装 |
|---|---|---|
az CLI | 主要工具：身份验证（az login）、通过 az rest 调用 Fabric REST API | 大多数开发环境中已预装 |
curl | 用于 REST 调用的备用 HTTP 客户端 | 已预装 |
jq | 解析 JSON 响应、提取字段、格式化输出 | 已预装或可轻松安装 |
base64 | 解码 base64 格式的定义部分 | bash 内置；PowerShell 使用 [Convert]::FromBase64String |
bash/pwsh | 脚本执行 | 已预装 |
Agent 检查 — 首次操作前验证：
az account show >/dev/null 2>&1 || echo "RUN: az login" command -v jq >/dev/null 2>&1 || echo "INSTALL: apt-get install jq OR brew install jq"
连接
解析工作区 ID 和 Dataflow ID
请参阅 COMMON-CLI.md 中的“在 Fabric 中查找工作区和项目”：
# Find workspace ID by name
WS_ID=$(az rest --method get \
--resource "https://api.fabric.microsoft.com" \
--url "https://api.fabric.microsoft.com/v1/workspaces" \
--query "value[?displayName=='My Workspace'].id" --output tsv)
# Find dataflow ID by name within workspace
DF_ID=$(az rest --method get \
--resource "https://api.fabric.microsoft.com" \
--url "https://api.fabric.microsoft.com/v1/workspaces/$WS_ID/dataflows" \
--query "value[?displayName=='Sales Data Pipeline'].id" --output tsv)
可复用的连接变量
# Set once at script top
WS_ID="<workspaceId>"
DF_ID="<dataflowId>"
API="https://api.fabric.microsoft.com/v1"
AZ="az rest --resource https://api.fabric.microsoft.com"
Agent 式探索（“与我的 Dataflow 对话”）
发现顺序
按以下顺序运行，以全面探索工作区中的 Dataflow。更多模式请参阅 references/discovery-queries.md。
# 1. List workspaces → find target
az rest --method get --resource "https://api.fabric.microsoft.com" \
--url "$API/workspaces" --query "value[].{name:displayName, id:id}" -o table
# 2. List dataflows → enumerate all
az rest --method get --resource "https://api.fabric.microsoft.com" \
--url "$API/workspaces/$WS_ID/dataflows" \
--query "value[].{name:displayName, id:id, desc:description}" -o table
# 3. Get dataflow properties
az rest --method get --resource "https://api.fabric.microsoft.com" \
--url "$API/workspaces/$WS_ID/dataflows/$DF_ID"
# 4. Discover parameters
# Note: the /parameters endpoint returns DataflowNotParametricError (an HTTP 4xx)
# for a non-parametric dataflow (no Power Query parameters). Treat that as
# "this dataflow has no parameters" and report it plainly — do NOT surface the
# raw error. Optionally confirm by checking mashup.pq for `IsParameterQuery`.
az rest --method get --resource "https://api.fabric.microsoft.com" \
--url "$API/workspaces/$WS_ID/dataflows/$DF_ID/parameters" \
--query "value[].{name:name, type:type, required:isRequired, default:defaultValue}" -o table
# 5. Get definition → decode mashup.pq
RESPONSE=$(az rest --method post --resource "https://api.fabric.microsoft.com" \
--url "$API/workspaces/$WS_ID/dataflows/$DF_ID/getDefinition")
echo "$RESPONSE" | jq -r '.definition.parts[] | select(.path=="mashup.pq") | .payload' | base64 --decode
# 6. Check job history
az rest --method get --resource "https://api.fabric.microsoft.com" \
--url "$API/workspaces/$WS_ID/dataflows/$DF_ID/jobs/instances" \
--query "value[].{status:status, type:invokeType, start:startTimeUtc, end:endTimeUtc, error:failureReason}" -o table
Agent 式工作流
- 发现 → 运行步骤 1–3，列出并识别 Dataflow。
- 参数 → 通过步骤 4 了解输入和默认值。
- 定义 → 通过步骤 5 检查 M 查询、连接和暂存配置。
- 监控 → 通过步骤 6 查看刷新历史记录和错误模式。
- 迭代 → 深入检查特定查询或连接详情。
- 呈现 → 总结发现，或生成可复用脚本（请参阅 script-templates.md）。
注意事项、规则和故障排除
完整的平台注意事项请参阅：DATAFLOWS-CONSUMPTION-CORE.md“注意事项和故障排除参考”，以及 COMMON-CLI.md“注意事项和故障排除（CLI 专属）”。
必须执行
- 始终先运行
az login—az rest使用当前活动会话。没有会话会导致难以理解的错误。 - 始终指定
--resource "https://api.fabric.microsoft.com"— 受众不正确会导致 401 错误。 - 处理分页 — 使用
continuationToken重复发送请求，直到该令牌缺失或为 null。 - 处理
getDefinition的 LRO — 可能会返回带有Location标头的202 Accepted；轮询直至完成。 - 检查前先解码 base64 — 定义部分采用 base64 编码。
- 对
getDefinition使用 POST — 它不是 GET 端点。
避免
- 硬编码 GUID — 始终使用先列出、再筛选的模式进行发现。
- 假设
getDefinition使用 GET — 它使用 POST（常见错误）。 - 忽略分页 — 列表端点可能只返回部分结果。
- 过于频繁地轮询 — 对于 429 响应，请遵循
Retry-After标头。 - 期望 Viewer 角色可以调用
getDefinition— 需要 Read+Write（Contributor+）。
优先采用
- 优先使用
az rest，而不是原始curl— 它会自动处理身份验证。 - 采用先列出、再筛选的模式 — Dataflow 不支持服务器端名称筛选。
- 对作业轮询采用指数退避 — 5s → 10s → 20s → 最多 30s。
- 使用
jq解析响应 — 比 shell 字符串操作更简洁。 - 使用 JMESPath
--query，直接在az rest中简单提取字段。 - 使用环境变量（
WS_ID、DF_ID、API）以便复用脚本。
故障排除
| 症状 | 原因 | 修复方法 |
|---|---|---|
401 Unauthorized | 令牌已过期或 audience 错误 | az login；确保使用 --resource "https://api.fabric.microsoft.com" |
在 getDefinition 上出现 403 Forbidden | Viewer 角色（只读） | 需要 Contributor 角色或更高权限（读写） |
404 Not Found | 工作区或 dataflow ID 错误 | 通过 List Dataflows API 重新发现 |
getDefinition 返回 202 | 定义较大或服务端负载较高 | 轮询 Location 标头 URL，直到操作完成 |
在 /parameters 上出现 DataflowNotParametricError（4xx） | Dataflow 没有 Power Query 参数（非参数化） | 这是预期情况——报告“此 dataflow 没有参数”；不要显示原始错误。可选通过 mashup.pq 的 IsParameterQuery 确认 |
queryMetadata.json 中的 connections 数组为空或不存在 | 查询使用内联／字面量源（例如 #table()），因此没有可绑定的外部连接 | 这是预期情况——报告“没有外部数据源连接（内联源）”；不要捏造绑定 |
| Base64 解码后显示乱码 | 编码内容中有 BOM | 解码时剥除 UTF-8 BOM（\xEF\xBB\xBF） |
429 TooManyRequests | 请求频率受限 | 遵循 Retry-After 标头；实现指数退避 |
| 列表中有重复结果 | 重用了过期的 continuationToken | 始终使用最新响应中的令牌 |
OperationNotSupportedForItem | 项目类型错误 | 通过 Get Item 确认项目类型为 Dataflow |
示例
示例 1：列出工作区中的所有 Dataflow
提示词：“列出此工作区中的所有 dataflow。”
Agent 响应：
az rest --method get \
--url "https://api.fabric.microsoft.com/v1/workspaces/${WS_ID}/dataflows" \
--resource "https://api.fabric.microsoft.com" \
--query "value[].{Name:displayName, Id:id, Type:type}" -o table
示例 2：解码 Dataflow 定义
提示词：“向我展示此 dataflow 中的 Power Query M 代码。”
Agent 响应：
# Step 1: Request definition (POST returns 202 with Location header).
# az rest cannot expose response headers, so capture Location via curl + az-acquired token.
TOKEN=$(az account get-access-token --resource "https://api.fabric.microsoft.com" --query accessToken -o tsv)
LOCATION=$(curl -sS -X POST \
-H "Authorization: Bearer $TOKEN" -H "Content-Length: 0" \
"https://api.fabric.microsoft.com/v1/workspaces/${WS_ID}/dataflows/${DF_ID}/getDefinition" \
-o /dev/null -D - | tr -d '\r' | grep -i "^location:" | awk '{print $2}')
# Step 2: Poll until definition is ready
DEF=$(az rest --method get --url "${LOCATION}" \
--resource "https://api.fabric.microsoft.com")
# Step 3: Decode mashup.pq to see the Power Query M code
echo "$DEF" | python3 -c "
import json, base64, sys
parts = json.load(sys.stdin)['definition']['parts']
for p in parts:
if p['path'] == 'mashup.pq':
print(base64.b64decode(p['payload']).decode('utf-8'))
"
示例 3：查看刷新作业历史记录
提示词：“向我展示此 dataflow 最近的刷新历史记录。”
Agent 响应：
# Get recent job instances for a dataflow
az rest --method get \
--url "https://api.fabric.microsoft.com/v1/workspaces/${WS_ID}/dataflows/${DF_ID}/jobs/instances?limit=5" \
--resource "https://api.fabric.microsoft.com" \
--query "value[].{Status:status, Start:startTimeUtc, End:endTimeUtc, Id:id}" -o table
示例 4：从定义中发现参数
提示词：“此 dataflow 接受哪些参数？”
Agent 响应：
# After decoding the definition (see Example 2), extract parameters:
echo "$DEF" | python3 -c "
import json, base64, sys
parts = json.load(sys.stdin)['definition']['parts']
for p in parts:
if p['path'] == 'queryMetadata.json':
meta = json.loads(base64.b64decode(p['payload']).decode('utf-8'))
for qname, qmeta in meta.get('queriesMetadata', {}).items():
if qmeta.get('queryGroupId') == 'parameters' or 'IsParameterQuery' in str(qmeta):
print(f'Parameter: {qname}')
"
查询评估
执行数据流中的单个查询并检查结果。响应是原始 Apache Arrow IPC 流，其 Content-Type 为 application/vnd.apache.arrow.stream——不是 JSON 信封。有效流的前 4 个字节是 IPC 延续标记 ff ff ff ff。请使用 pyarrow.ipc.open_stream() 解析。
线上格式：
executeQuery返回原始 Apache Arrow IPC 字节流（Content-Type: application/vnd.apache.arrow.stream）——不是 JSON。不要尝试用jq解析——没有 JSON 信封可供提取。使用--output-file保存字节并按 Arrow 格式解析（参见示例 5–7）。
失败时仍返回 HTTP 200：即使底层源查询失败（Kusto SEM0100、T-SQL 语法错误、缺少列等），
executeQuery也会返回200 OK和application/vnd.apache.arrow.stream。错误嵌入在流的PQ Arrow Metadata部分中，格式为{"Error":"..."}——检测代码片段请参见 dataflows-authoring-cli § mashup-preview.md → 检测 Arrow 正文中的失败。简单的 HTTP 状态检查会将失败误判为成功。
意图区分（权威的
executeQuery参考文档见 mashup-preview.md）：同一个executeQuery端点支持两种不同的意图。此技能涵盖使用类意图：
- (a) 执行已保存的查询——仅使用请求正文
{"QueryName":"<saved-shared>"}（不包含customMashupDocument）。- (b) 临时只读
customMashupDocument——预览候选section Section1; ...文档，不打算通过updateDefinition持久化（示例 7）。如果你打算持久化 M，请使用
dataflows-authoring-cli§ 工作流 C（以预览驱动的创作循环）——其中补充了引导绑定规则（为新的带凭据数据流解决连接绑定的先有鸡还是先有蛋问题）、自动包装规则、避免无界预览的硬性要求，以及预览后的持久化步骤。
自动包装注意事项：Fabric REST API 要求
customMashupDocument是一个完整的section Section1; ... shared X = ...;文档。服务端不会自动包装原始的let ... in ...表达式——请发送完整的 section 文档，并确保QueryName请求字段与文档中声明的某个shared成员相匹配。
正文格式：发送一个扁平对象，其中顶层包含
QueryName（字段名在线上传输时不区分大小写；规范形式为 PascalCase）。{"queries":[{...}]}数组格式无论内部字段的大小写如何，都会始终返回400 DataflowExecuteQueryError: Invalid query name。错误的QueryName值会返回QueryNotFound（错误代码不同）。请参见 dataflows-authoring-cli § mashup-preview.md → 请求正文。
耗时较长的已保存查询发生
TimedOut时的恢复方法：如果已保存的shared <Q>成员对服务端executeQuery预算来说开销过大（观测到的上限约为 90 秒；表现为400 DataflowExecuteQueryError→InternalErrorCode: EvaluationError, Message: Evaluation result error code: TimedOut），不要重试相同的请求正文——结果是确定性的。请改为重新发送请求，将customMashupDocument设为完整的已持久化mashup.pq（通过示例 2 解码），并追加一个探测查询shared __PreviewCap = Table.FirstN(<Q>, 10);，同时将QueryName设为"__PreviewCap"。对于大多数连接器（Kusto、SQL、OData、web 表），上限会向下推送，使数据源只返回样本。请参见 dataflows-authoring-cli § mashup-preview.md → 已观测到的InternalErrorCode值。
先决条件
- Dataflow 必须存在（通过上面的示例 1 验证）
- Dataflow 中必须存在查询名称（通过示例 2 中的 mashup.pq 验证）
- 用户必须在该 Dataflow 上拥有 Contributor 角色或更高权限（读写）
- 可选：查询参数和自定义 M 代码修改
示例 5：执行查询并保存结果
提示词：“在此 Dataflow 中执行 SalesData 查询并保存结果。”
Agent 响应：
# Step 1: Identify the query to execute (must be a `shared` member of the dataflow's mashup)
QUERY_NAME="SalesData" # Replace with your query name
# Step 2: Build the request body. To execute the persisted query as-is, send ONLY
# QueryName — omit customMashupDocument. (Do NOT pass a self-referential
# `shared X = let Source = #shared[X] in Source` — that recurses on itself
# inside the candidate document.) For custom M, see Example 7.
jq -n --arg q "$QUERY_NAME" '{QueryName: $q}' > req.json
# Step 3: Execute the query (raw Apache Arrow IPC stream is written directly to disk)
az rest --method post \
--url "https://api.fabric.microsoft.com/v1/workspaces/${WS_ID}/dataflows/${DF_ID}/executeQuery" \
--resource "https://api.fabric.microsoft.com" \
--body "@req.json" \
--output-file "${QUERY_NAME}_results.arrow"
# Step 4: Check for embedded source errors (HTTP 200 does NOT mean success)
if grep -q '"Error":"' "${QUERY_NAME}_results.arrow"; then
echo "Query execution failed (embedded in Arrow body):"
python3 -c "import re,sys; raw=open(sys.argv[1],'rb').read().decode('utf-8','replace'); m=re.search(r'\\{\"Error\":\"[^\"]+\"\\}', raw); print(m.group(0) if m else '(marker present, JSON not parsed)')" "${QUERY_NAME}_results.arrow"
exit 1
fi
echo "Query results saved to ${QUERY_NAME}_results.arrow"
为什么使用
--output-file？ 不使用它时，az rest会将原始字节转储到标准输出，并显示警告Not a json response, outputting to stdout. For binary data suggest use "--output-file" to write to a file。对于executeQuery，始终传入--output-file，以便干净地捕获 Arrow IPC 流。前面文档中的jq -r '.data' | base64 --decode模式不适用于az rest——这里没有可供提取的 JSON 封装。
示例 5b：将查询结果呈现为 Markdown 表格
提示词：“显示这些结果的前 10 行。”
Agent 响应 — 读取已保存的 Arrow 文件，并在聊天中将 head(N) 呈现为 Markdown 表格。每次执行 executeQuery 后都必须这样做（默认 N=10）。可视化预览能捕获嵌入式错误检测器无法发现的静默成功错误：筛选器删除了所有行、引用了错误的列、转换类型错误、筛选条件偏差一位、联接键拼写错误。
N=${N:-10}; ARROW_FILE="${QUERY_NAME}_results.arrow"
python3 - <<EOF
import json, sys, pyarrow as pa, pyarrow.ipc as ipc
with open("$ARROW_FILE", "rb") as f:
try:
reader = ipc.open_stream(f)
except pa.lib.ArrowInvalid:
f.seek(0); reader = ipc.open_file(f)
# Surface any embedded engine error before rendering (mirrors mashup-preview.md § Error handling — A)
for v in (reader.schema.metadata or {}).values():
s = v.decode("utf-8", errors="replace")
if '"Error"' in s:
try:
msg = json.loads(s)["Error"]
except (ValueError, KeyError):
msg = s[:200]
sys.exit(f"Preview failed: {msg}")
table = reader.read_all()
rows, cols = table.num_rows, table.num_columns
print(f"**{rows} rows × {cols} columns** · {', '.join(table.schema.names)}\n")
if rows == 0:
sys.exit(0)
# Slice at the Arrow level before converting — avoids materialising the whole result.
df = table.slice(0, $N).to_pandas().copy()
# Truncate long string cells so the chat table stays readable.
for c in df.select_dtypes(include=["object", "string"]).columns:
df[c] = df[c].astype(str).str.slice(0, 50)
try:
print(df.to_markdown(index=False)) # requires the `tabulate` package
except ImportError:
print(df.to_string(index=False)) # fallback: fixed-width text
EOF
可选依赖项：
pandas.DataFrame.to_markdown()需要tabulate包——通过环境中标准的 Python 工具安装一次，方式与示例 6 中安装pyarrow/pandas相同。如果缺少tabulate，代码片段会回退到定宽的to_string()，因此仍可正常呈现。
何时跳过呈现：默认呈现
head(N)。仅在以下情况下跳过：(a) 用户明确要求原始 Arrow 文件且不需要预览；(b) 结果行数超过约 1000——此时仍应呈现head(N)，并添加一行说明总行数；或 (c) 结果正被直接流式传输到另一个会直接使用 Arrow 文件的工具。在其他所有情况下都应呈现——仅保存文件，用户在聊天中无法看到。
示例 6：将 Arrow 结果转换为 CSV
提示词：“将我的 Arrow 查询结果转换为 CSV，以便我在 Excel 中打开。”
Agent 响应：
先决条件（仅需设置一次的客户端环境）：此示例使用
pyarrow解码 Arrow IPC 流，并使用pandas写入 CSV。通过环境中标准的 Python 工具安装一次。这些是本地客户端依赖项，不属于 Fabric 配方的一部分。
python3 <<EOF
import pyarrow as pa
import pandas as pd
import sys
ARROW_FILE = "${QUERY_NAME}_results.arrow"
CSV_FILE = "${QUERY_NAME}_results.csv"
# Open the Arrow IPC stream directly (raw on the wire, no envelope)
with open(ARROW_FILE, "rb") as f:
reader = pa.ipc.open_stream(f)
# Defensive: surface any error embedded in the stream's PQ Arrow Metadata
md = reader.schema.metadata or {}
for k, v in md.items():
s = v.decode("utf-8", errors="replace")
if '"Error"' in s:
print(f"Preview failed: {s}", file=sys.stderr)
sys.exit(1)
table = reader.read_all()
# Convert to pandas and export as CSV
df = table.to_pandas()
df.to_csv(CSV_FILE, index=False)
print(f"Converted {len(df)} rows to CSV")
print("Columns:", list(df.columns))
EOF
示例 7：使用自定义 M 代码进行查询
提示词：“对这个 Dataflow 运行一次临时的 M 查询，但不保存。”
意图：临时只读执行。
customMashupDocument不会持久化。如果你打算通过updateDefinition保存 M，请改用dataflows-authoring-cli§ 工作流 C——它包含 bootstrap-bind、自动封装和预览后的持久化规则。
Agent 响应：
# Execute a query with custom M code (e.g., filter, aggregate, transform).
# The customMashupDocument must be a complete `section` document; az rest does NOT auto-wrap raw expressions.
CUSTOM_M='section Section1;
shared CustomQuery = let
Source = Table.FromRecords({[id=1, name="Alice"], [id=2, name="Bob"]}),
Filtered = Table.SelectRows(Source, each [id] > 0)
in
Filtered;'
jq -n --arg m "$CUSTOM_M" '{QueryName: "CustomQuery", customMashupDocument: $m}' > req.json
az rest --method post \
--url "https://api.fabric.microsoft.com/v1/workspaces/${WS_ID}/dataflows/${DF_ID}/executeQuery" \
--resource "https://api.fabric.microsoft.com" \
--body "@req.json" \
--output-file custom_results.arrow
# Always check for embedded errors before treating the file as a success
if grep -q '"Error":"' custom_results.arrow; then
echo "Custom query failed; inspect custom_results.arrow for the embedded {\"Error\":...} block."
exit 1
fi
输出要求
此技能完成任务后，Agent 应返回：
| 字段 | 约定 |
|---|---|
| 详细程度 | 状态说明应简洁概述（3–10 行）；列表或检查类响应应使用 Markdown 表格。 |
| 默认格式 | list 类查询使用 Markdown 表格；单资源响应使用围栏 JSON 代码块；原始解码后的 mashup.pq 使用围栏 ```m 代码块。对于 executeQuery：将完整的 Arrow 流保存到文件，并且在聊天中将 head(N)（默认 N=10）呈现为 Markdown 表格——参见 示例 5b。仅当用户明确要求、rows > 1000（呈现 head + 总数说明），或结果正被流式传输到另一个工具时，才跳过呈现。 |
| 副作用说明 | 这是一项只读技能——绝不暗示会执行修改，并拒绝任何创建、更新或删除请求（转至 dataflows-authoring-cli）。 |
| 验证 | 在响应中包含源 URL（例如 az rest --url 的值），以便用户重现调用。 |
| 错误呈现 | 如果 executeQuery 返回的 Arrow 中包含嵌入的 {"Error":"..."}，应原样呈现错误，且不得将部分结果作为成功结果展示。 |