Macrocosmos SN13 API — 社交媒体数据技能

通过 Macrocosmos SN13 API（运行于 Bittensor 网络）按关键词、用户名、日期范围及多种筛选条件，实时获取 X（Twitter）和 Reddit 的社交媒体数据，并附带互动指标（engagement metrics）。

元数据

• name: macrocosmos-social-data
• version: 1.0.1
• homepage: https://github.com/macrocosm-os/macrocosmos-mcp
• source: https://github.com/macrocosm-os/macrocosmos-mcp
• pypi: https://pypi.org/project/macrocosmos-mcp
• subnet: Bittensor SN13（Data Universe）
• author: Macrocosmos AI
• license: MIT

必需的环境变量

配置说明：必须将 MC_API 设置为环境变量。该密钥在 REST 调用中作为 Bearer Token 放入 Authorization 请求头；或直接传入 Python SDK 客户端。

API 终端地址（Endpoint）

POST https://constellation.api.cloud.macrocosmos.ai/sn13.v1.Sn13Service/OnDemandData

请求头（Headers）

Content-Type: application/json
Authorization: Bearer 

请求格式

{
  "source": "X",
  "usernames": ["@elonmusk"],
  "keywords": ["AI", "bittensor"],
  "start_date": "2026-01-01",
  "end_date": "2026-02-10",
  "limit": 10,
  "keyword_mode": "any"
}

参数说明

响应格式

{
  "data": [
    {
      "datetime": "2026-02-10T17:30:58Z",
      "source": "x",
      "text": "Tweet content here",
      "uri": "https://x.com/username/status/123456",
      "user": {
        "username": "example_user",
        "display_name": "Example User",
        "followers_count": 1500,
        "following_count": 300,
        "user_description": "Bio text",
        "user_blue_verified": true,
        "profile_image_url": "https://pbs.twimg.com/..."
      },
      "tweet": {
        "id": "123456",
        "like_count": 42,
        "retweet_count": 10,
        "reply_count": 5,
        "quote_count": 2,
        "view_count": 5000,
        "bookmark_count": 3,
        "hashtags": ["#AI", "#bittensor"],
        "language": "en",
        "is_reply": false,
        "is_quote": false,
        "conversation_id": "123456"
      }
    }
  ]
}

curl 示例

1. 在 X 上按关键词搜索

curl -s -X POST https://constellation.api.cloud.macrocosmos.ai/sn13.v1.Sn13Service/OnDemandData 
  -H "Content-Type: application/json" 
  -H "Authorization: Bearer YOUR_API_KEY" 
  -d '{
    "source": "X",
    "keywords": ["bittensor"],
    "start_date": "2026-01-01",
    "limit": 10
  }'

2. 获取特定用户的推文

curl -s -X POST https://constellation.api.cloud.macrocosmos.ai/sn13.v1.Sn13Service/OnDemandData 
  -H "Content-Type: application/json" 
  -H "Authorization: Bearer YOUR_API_KEY" 
  -d '{
    "source": "X",
    "usernames": ["@MacrocosmosAI"],
    "start_date": "2026-01-01",
    "limit": 10
  }'

3. 多关键词 AND 搜索（逻辑与）

curl -s -X POST https://constellation.api.cloud.macrocosmos.ai/sn13.v1.Sn13Service/OnDemandData 
  -H "Content-Type: application/json" 
  -H "Authorization: Bearer YOUR_API_KEY" 
  -d '{
    "source": "X",
    "keywords": ["chutes", "bittensor"],
    "keyword_mode": "all",
    "start_date": "2026-01-01",
    "limit": 20
  }'

4. Reddit 搜索

curl -s -X POST https://constellation.api.cloud.macrocosmos.ai/sn13.v1.Sn13Service/OnDemandData 
  -H "Content-Type: application/json" 
  -H "Authorization: Bearer YOUR_API_KEY" 
  -d '{
    "source": "REDDIT",
    "keywords": ["r/MachineLearning", "transformers"],
    "start_date": "2026-02-01",
    "limit": 50
  }'

5. 用户 + 关键词联合过滤

curl -s -X POST https://constellation.api.cloud.macrocosmos.ai/sn13.v1.Sn13Service/OnDemandData 
  -H "Content-Type: application/json" 
  -H "Authorization: Bearer YOUR_API_KEY" 
  -d '{
    "source": "X",
    "usernames": ["@opentensor"],
    "keywords": ["subnet"],
    "start_date": "2026-01-01",
    "limit": 20
  }'

Python 示例

使用 macrocosmos SDK

import asyncio
import macrocosmos as mc

async def search_tweets():
    client = mc.AsyncSn13Client(api_key="YOUR_API_KEY")

    response = await client.sn13.OnDemandData(
        source="X",
        keywords=["bittensor"],
        usernames=[],
        start_date="2026-01-01",
        end_date=None,
        limit=10,
        keyword_mode="any",
    )

    if hasattr(response, "model_dump"):
        data = response.model_dump()

    for tweet in data["data"]:
        print(f"@{tweet['user']['username']}: {tweet['text'][:100]}")
        print(f"  Likes: {tweet['tweet']['like_count']} | Views: {tweet['tweet']['view_count']}")

asyncio.run(search_tweets())

使用 requests（REST 方式）

import requests

url = "https://constellation.api.cloud.macrocosmos.ai/sn13.v1.Sn13Service/OnDemandData"
headers = {
    "Content-Type": "application/json",
    "Authorization": "Bearer YOUR_API_KEY"
}
payload = {
    "source": "X",
    "keywords": ["bittensor"],
    "start_date": "2026-01-01",
    "limit": 10
}

response = requests.post(url, json=payload, headers=headers)
data = response.json()

for tweet in data["data"]:
    print(f"@{tweet['user']['username']}: {tweet['text'][:100]}")

提示与已知行为

稳定可靠的功能

• 高热度关键词搜索：如 “bittensor”、“AI”、“iran”、“lfg” 等热门词，响应迅速
• 较宽泛的日期范围：将 start_date 设得更早（例如数周或数月前），可显著提升结果召回率
• keyword_mode: "all"：适合查找两个主题的交集（例如 “chutes” AND “bittensor”）

可能不稳定的功能

• 仅按用户名查询：可能出现超时（DEADLINE_EXCEEDED）。建议显式设置较早的 start_date 缓解该问题
• 小众/低频关键词：过于具体的术语可能因矿工节点未索引对应数据而超时
• 未指定 start_date：默认仅检索最近 24 小时数据，易遗漏信息；建议始终显式设置

面向 LLM Agent 的最佳实践

1. 务必设置 start_date —— 不要依赖默认的 24 小时窗口。用户查询建议至少回溯 7 天
2. 优先使用关键词而非用户名 —— 关键词搜索稳定性更高
3. 若需按用户名查询，必须配合设置 start_date（建议回溯数周至数月）
4. 组合主题与子主题时，使用 keyword_mode: "all"（例如 “bittensor” + “chutes”）
5. 优雅处理超时 —— 若查询失败，可尝试扩大日期范围，或切换为关键词搜索
6. 解析互动指标 —— 利用 view_count、like_count、retweet_count 等字段对结果进行相关性排序
7. 检查 is_reply 和 is_quote 字段 —— 根据实际需求过滤原始推文（original tweet）、回复（reply）或引用推文（quote）

Gravity API（大规模数据采集）

当所需数据量超过 1000 条时，请使用 Gravity 系列接口：

创建采集任务

POST /gravity.v1.GravityService/CreateGravityTask

{
  "gravity_tasks": [
    {"platform": "x", "topic": "#bittensor", "keyword": "dTAO"}
  ],
  "name": "Bittensor dTAO Collection"
}

注意： X 平台的话题（topic）必须以 # 或 $ 开头；Reddit 平台则使用子版块格式（如 r/xxx）。

查询任务状态

POST /gravity.v1.GravityService/GetGravityTasks

{
  "gravity_task_id": "multicrawler-xxxx-xxxx",
  "include_crawlers": true
}

构建数据集

POST /gravity.v1.GravityService/BuildDataset

{
  "crawler_id": "crawler-0-multicrawler-xxxx",
  "max_rows": 10000
}

警告：调用此接口将永久停止对应爬虫任务。

获取数据集下载链接

POST /gravity.v1.GravityService/GetDataset

{
  "dataset_id": "dataset-xxxx-xxxx"
}

任务完成后，返回 Parquet 文件的下载 URL。

工作流概览

快速查询（< 1000 条结果）：
  OnDemandData → 即时返回结果

大规模采集（7 日周期爬取）：
  CreateGravityTask → GetGravityTasks（轮询监控）→ BuildDataset → GetDataset（下载）

错误码参考