公开 API 文档

curl -X POST "https://api.dk-ai.tech/api/tools/xhs_search_notes" \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -H "Content-Type: application/json; charset=utf-8" \
  --data-binary '{"keyword":"\u5496\u5561"}'

{
  "creditsUsed": 20,
  "creditsRemaining": 9980,
  "traceId": "trace_xxx",
  "durationMs": 1280,
  "data": {
    "items": [
      {
        "contentId": "example_content_id",
        "title": "示例标题",
        "contentSummary": "搜索列表返回的正文摘要",
        "contentUrl": "https://www.xiaohongshu.com/explore/example_content_id",
        "desktopUrl": "DK 返回的官方 PC 内容链接"
      }
    ],
    "resultCount": 1,
    "pagination": {
      "hasMore": true,
      "nextPageToken": "NEXT_PAGE_TOKEN",
      "cursorParam": "pageToken"
    }
  }
}

{
  "mcpServers": {
    "dk-social": {
      "url": "https://api.dk-ai.tech/mcp",
      "headers": {
        "Authorization": "Bearer YOUR_API_KEY"
      }
    }
  }
}

// MCP tools/call 返回后：
const execution = result.structuredContent
const data = execution.data
const items = data.items
const nextPageToken = data.pagination?.nextPageToken
const creditsUsed = execution.creditsUsed
const creditsRemaining = execution.creditsRemaining
const traceId = execution.traceId

import requests

api_key = "YOUR_API_KEY"
page_token = None

for page in range(1, 6):
    payload = {"keyword": "咖啡"}
    if page_token:
        payload["pageToken"] = page_token

    response = requests.post(
        "https://api.dk-ai.tech/api/tools/xhs_search_notes",
        headers={"Authorization": f"Bearer {api_key}"},
        json=payload,
        timeout=60,
    )
    envelope = response.json()
    data = envelope["data"]
    items = data["items"]
    pagination = data.get("pagination") or {}

    append_items_to_csv(items)
    save_checkpoint(page, envelope)

    page_token = pagination.get("nextPageToken")
    if not pagination.get("hasMore") or not page_token:
        break

import json
from pathlib import Path

import requests

api_key = "YOUR_API_KEY"
output = Path("results.jsonl")
checkpoint = Path("checkpoint.json")
state = json.loads(checkpoint.read_text("utf-8")) if checkpoint.exists() else {"page": 0, "pageToken": None}

for page in range(state["page"] + 1, 21):
    body = {"keyword": "咖啡"}
    if state["pageToken"]:
        body["pageToken"] = state["pageToken"]

    response = requests.post(
        "https://api.dk-ai.tech/api/tools/dy_search_videos",
        headers={"Authorization": f"Bearer {api_key}"},
        json=body,
        timeout=60,
    )
    response.raise_for_status()
    envelope = response.json()
    data = envelope["data"]

    with output.open("a", encoding="utf-8") as file:
        for item in data["items"]:
            file.write(json.dumps(item, ensure_ascii=False) + "\n")

    pagination = data.get("pagination") or {}
    state = {"page": page, "pageToken": pagination.get("nextPageToken")}
    checkpoint.write_text(json.dumps(state, ensure_ascii=False), encoding="utf-8")

    if page % 5 == 0:
        print(f"checkpoint saved after page {page}")
    if not pagination.get("hasMore") or not state["pageToken"]:
        break