Datadata REST API

功能概览

本 skill 提供 Datadata REST API 的完整参考文档，通过 urllib.request（仅标准库，零额外依赖）展示所有端点的调用方式。

首要用例是生成独立 Python 脚本（爬虫、ETL、批处理），但 API 文档本身是通用的 — 任何需要直接调用 Datadata API 的场景均可参考。

交互式操作（聊天中执行查询、探索数据等）请使用 datadata-manual skill。 MCP 已完整覆盖搜索、查询、Data Spaces 等日常交互功能。

核心能力

• REST API 参考 — 所有端点的完整说明，含请求/响应示例
• urllib.request 调用模板 — 即拿即用的 Python 代码片段
• 数据源操作 — 搜索、元信息、表结构、Schema 扫描
• SQL 查询 — execute-adhoc（只读，异步）、结果下载（NDJSON/CSV）
• Dataspace SQL 执行 — 通过 execute 端点运行任意 DuckDB SQL（建表/写入/改表/删表），同步返回
• 设备授权 — 脚本中自动获取/刷新 API Key

使用场景

以下场景请使用 datadata-manual skill：

• 聊天中交互式查询数据（"帮我查一下销售数据"）
• 探索数据源结构（"看看这个 datasource 有哪些表"）
• 设置表/列注释、触发扫描等即时操作

概念

• Datasource — 查询目标的数据源。不同类型的 datasource（dataspace、MySQL、ClickHouse、CSV 等）有不同的表命名约定。
• Data space — 录入数据的目标，dataspace 类型 datasource（旧类型名 ducklake 已废弃）。通过 POST /dataspaces/{datasourceId}/execute 执行任意 DuckDB SQL 来管理表结构与数据。
• 读 / 写分离 — 查询/读取任何数据走 execute-adhoc（查询引擎，把 dataspace 作为只读数据源挂载）；修改某个 dataspace 内的表结构或数据走 POST /dataspaces/{id}/execute（单一 dataspace，同步执行）。
• Query (execute-adhoc) — 只读抽象，包含 SQL 脚本、datasource 绑定和查询引擎类型。
• Execution — execute-adhoc 查询的后台执行实例。通过 /executions/{id}/result 异步获取结果。（Dataspace SQL 执行是同步的，无 execution。）

API Key

手动设置（推荐用于脚本）

export DATADATA_API_KEY="ak_..."
export DATADATA_BASE_URL="https://www.datadata.com"  # 可选，本地开发时覆盖

API Key 在 Datadata 网页端创建：登录 → 头像 → Settings → API Keys → 创建新 Key。

所需权限：

• datasources:read — 查询元信息
• queries:execute-adhoc — 执行 SQL（读取查询）
• executions:get — 获取查询结果
• datasources:scan — 设置注释、触发扫描

Dataspace SQL 执行（POST /dataspaces/{id}/execute）当前无需特殊权限 — 有效 API Key（登录）即可，无需 data-spaces:write。

设备授权（适用于无人值守脚本）

脚本中可通过设备授权自动获取 API Key（有效期 90 天）：

# Step 1: 发起设备授权
resp = _request(f"{BASE_URL}/api/v1/api-keys/device-flow/code", method="POST")
print(f"请打开: {resp['verificationUriComplete']}")

# Step 2: 等待用户完成登录后换取 token
resp2 = _request(f"{BASE_URL}/api/v1/api-keys/device-flow/token", method="POST",
                 payload={"deviceCode": resp["deviceCode"]})
api_key = resp2["apiKey"]["key"]

Python 请求模板

所有脚本的基础模板（仅标准库，零依赖）：

import json, urllib.request, os

API_KEY = os.environ.get("DATADATA_API_KEY", "ak_...")
BASE_URL = os.environ.get("DATADATA_BASE_URL", "https://www.datadata.com")

def _request(url, method="GET", payload=None):
    headers = {"X-Datadata-Api-key": API_KEY, "Accept": "application/json"}
    data = json.dumps(payload).encode() if payload else None
    if payload:
        headers["Content-Type"] = "application/json"
    req = urllib.request.Request(url, data=data, headers=headers, method=method)
    with urllib.request.urlopen(req) as resp:
        raw = resp.read().decode()
        return json.loads(raw) if raw else None

规则

🔴 搜索数据源：必须让用户确认，禁止自动选用

搜索结果绝不能由 Agent 自动选取。将结果以序号列表呈现，等待用户明确选择。

🔴 最小操作原则：完成一步即停

Agent 只执行用户明确要求的操作。生成脚本后立即停止，不要自动运行或推断后续步骤。

🔴 生成的脚本必须零依赖

只使用 Python 标准库（urllib.request、json、os 等），不依赖 requests、pandas 等第三方库。

查询只读

execute-adhoc 仅支持 SELECT（只读查询引擎，dataspace 以只读方式挂载）。INSERT/UPDATE/DELETE/DDL 请使用 Dataspace SQL 执行端点 POST /dataspaces/{datasourceId}/execute。

结果处理

生成的脚本应将查询结果保存到文件，不要硬编码打印大数据集。

References

相关 skill

• datadata-dql — DQL（Starlark）数据处理脚本编写