ThinkingData MCP（Python）

把数数（ThinkingData）Open API 封装成面向 AI 的只读 MCP 服务。项目使用官方 Python MCP SDK 的稳定 1.x 版本、FastMCP、httpx 和 Pydantic。

已实现能力

• list_events：查询项目事件，默认隐藏已隐藏事件。

• list_properties：查询事件属性或用户属性。

• list_reports：查询已保存报表，可按名称、类型和创建人过滤。

• event_analysis：使用简化参数执行事件分析。

• raw_event_analysis：安全执行从数数报表复制的原始事件分析 JSON。

• retention_analysis：执行初始事件到回访事件的留存分析。

• query_sql：执行单条 Trino 只读 SQL，AST 校验后自动添加或收紧 LIMIT；兼容数数逐行 JSON 响应。

• start_paged_sql / submit_async_sql：启动分页或异步只读 SQL 任务。

• sql_task_status / get_sql_result_page / cancel_sql_task：管理 SQL 任务生命周期。

• 元数据内存 TTL 缓存。

• 数数 Token 仅由服务端读取，不出现在 MCP 工具参数或错误信息中。

Related MCP server: Mixpanel MCP Server

环境要求

• Python 3.11+

• 推荐使用 uv 管理环境

• 数数私有化集群地址、项目 ID 和项目级查询密钥

本地启动

cp .env.example .env
# 编辑 .env，填写 THINKINGDATA_BASE_URL、THINKINGDATA_TOKEN、THINKINGDATA_PROJECT_ID

uv sync --extra dev
uv run thinkingdata-mcp

默认 Streamable HTTP 地址：

http://127.0.0.1:8000/mcp

使用 MCP Inspector 调试：

npx -y @modelcontextprotocol/inspector

如需以 stdio 方式启动：

MCP_TRANSPORT=stdio uv run thinkingdata-mcp

配置

THINKINGDATA_PROJECT_ID 对应 Open API 请求中的 projectId。它通常是一个正整数；不要填写类似 2202573b57974fb09e92361bb7519651 的项目 APPID。可以从数数页面 URL、项目管理页面或让数数管理员查询数字项目 ID。

工具示例

事件分析工具接受一个结构化请求：

{
  "request": {
    "event": "login",
    "metric": "TRIG_USER_NUM",
    "start_time": "2026-07-01 00:00:00",
    "end_time": "2026-07-31 23:59:59",
    "granularity": "day",
    "group_by": [{"property": "channel", "table_type": "event"}],
    "filters": [
      {
        "property": "country",
        "operator": "equal",
        "values": ["CN"],
        "table_type": "event"
      }
    ]
  }
}

支持的聚合包括 TOTAL_TIMES、TRIG_USER_NUM、PER_CAPITA_TIMES、SUM、AVG、MAX、MIN。属性聚合必须提供 metric_property。

原始报表查询

将数数报表中“复制 API 模型查询条件”得到的 JSON 作为 payload 传给 raw_event_analysis。服务会拒绝包含 Token、密码或授权字段的请求，并强制覆盖 projectId、limit 和 timeoutSeconds。

留存分析

{
  "request": {
    "initial_event": "register",
    "return_event": "app_start",
    "start_time": "2026-07-01 00:00:00",
    "end_time": "2026-07-31 23:59:59",
    "unit_num": 7,
    "granularity": "day",
    "subject_property": "#user_id"
  }
}

事件名必须使用数数项目中的真实事件名，可先通过 list_events 确认。

分页和异步 SQL

1. 使用 start_paged_sql 等待查询完成并获取 taskId，或使用 submit_async_sql 立即提交。

2. 通过 sql_task_status 查询任务，直到状态为 FINISHED。

3. 通过 get_sql_result_page 从 page_id=0 开始逐页下载。

4. 不再需要的运行中任务可通过 cancel_sql_task 取消。

分页和异步 SQL 同样只允许单条 SELECT/WITH，并会强制总行数 LIMIT。数数要求 page_size 至少为 1000。

安全边界

• 优先配置项目级 Token，不使用 root Token。

• query_sql 只接受一条 SELECT/WITH 查询，并拒绝 DDL、DML 和动态 LIMIT。

• 分页和异步 SQL 使用独立的总行数上限，但仍执行同样的只读 AST 校验。

• raw_event_analysis 不允许客户端指定项目或传入密钥。

• 服务会强制结果行数上限，但生产环境仍应在网关增加访问认证、请求限流和审计。

• 数数要求 Token 出现在 URL Query；反向代理和访问日志必须配置查询参数脱敏。

• 不要把 .env 提交到 Git。

• SQL 白名单目前依赖项目级 Token 隔离；如果一个 Token 能跨项目访问，应在投入生产前增加表级白名单。

测试与检查

uv run pytest
uv run ruff check .
uv run mypy

Docker

docker build -t thinkingdata-mcp .
docker run --rm -p 8000:8000 --env-file .env \
  -e MCP_HOST=0.0.0.0 thinkingdata-mcp