鲸准 MCP Server 使用说明

{
  "mcpServers": {
    "jingdata": {
      "type": "http",
      "url": "https://mcp.jingdata.com/sse",
      "headers": {
        "Authorization": "Bearer <YOUR_TOKEN>"
      }
    }
  }
}

{
  "jsonrpc": "2.0",
  "id": 2,
  "method": "tools/call",
  "params": {
    "name": "search",
    "arguments": {
      "arg0": "jingdata_project",
      "arg1": "{\"query\":{\"match_all\":{}},\"size\":10,\"track_total_hits\":true}"
    }
  }
}

import http.client, json, time

HOST        = "mcp.jingdata.com"
TOKEN       = "Bearer <YOUR_TOKEN>"        # 向鲸准申请获取
MCP_VERSION = "2024-11-05"                 # MCP 协议版本参数，需与客户端声明一致
session = None

def post(msg):
    global session
    c = http.client.HTTPSConnection(HOST, timeout=60)
    h = {
        "Authorization": TOKEN,
        "Content-Type": "application/json",
        "Accept": "application/json, text/event-stream",
    }
    if session:
        h["Mcp-Session-Id"] = session
    c.request("POST", "/sse", body=json.dumps(msg), headers=h)
    r = c.getresponse()
    sid = r.getheader("Mcp-Session-Id")
    if sid:
        session = sid                                     # 会话复用
    raw = r.read().decode("utf-8", "replace")
    c.close()
    for line in raw.splitlines():                         # 兼容 SSE 分片格式
        if line.startswith("data:"):
            return json.loads(line[5:].strip())
    return json.loads(raw)

def initialize():
    post({"jsonrpc": "2.0", "id": 1, "method": "initialize",
          "params": {"protocolVersion": MCP_VERSION, "capabilities": {},
                     "clientInfo": {"name": "my-app", "version": "1.0"}}})
    post({"jsonrpc": "2.0", "method": "notifications/initialized", "params": {}})

def call_tool(name, arguments):
    return post({"jsonrpc": "2.0", "id": 2, "method": "tools/call",
                 "params": {"name": name, "arguments": arguments}})

# ---- 用法：查最近一个完整交易日的全部投融资交易事件 ----
from datetime import date, timedelta

initialize()
day = date.today() - timedelta(days=1)        # 最近一天，可按需替换
dsl = json.dumps({
    "query": {"bool": {"filter": [
        {"range": {"finance_date": {
            "gte": day.isoformat(),
            "lt": (day + timedelta(days=1)).isoformat()}}}]}},
    "sort": [{"finance_date": {"order": "desc"}}, {"_id": {"order": "asc"}}],
    "size": 10, "from": 0, "track_total_hits": True,
}, ensure_ascii=False)

resp = call_tool("search", {"arg0": "jingdata_large_transaction", "arg1": dsl})
# 大响应会被切成多个 content 分片，必须直接拼接（不要用 "\n".join）
text = "".join(c.get("text", "") for c in resp["result"]["content"])
data = json.loads(text)
print(data["hits"]["total"], "条交易事件")

import time

def fetch_all(index, query, page_size=10, max_pages=None):
    """分页拉取一个索引的全部命中记录。

    已按本服务的约定处理：单页上限 10、稳定排序、按 _id 去重、
    错误正文判定、响应分片拼接、失败退避重试。
    依赖上一个示例中的 call_tool()。
    """
    rows, seen, page = [], set(), 0

    while max_pages is None or page < max_pages:
        sort = list(query.get("sort") or [])
        if not any(k in ("id", "_id") for s in sort for k in s):
            sort.append({"id": {"order": "asc"}})        # 稳定分页的次级排序

        dsl = dict(query)
        dsl["sort"] = sort
        dsl["size"] = page_size
        dsl["from"] = page * page_size
        dsl["track_total_hits"] = True

        data = None
        for attempt in range(4):                          # 最多重试 3 次
            try:
                r = call_tool("search", {
                    "arg0": index,
                    "arg1": json.dumps(dsl, ensure_ascii=False),
                })
                result = r.get("result", {})
                # 大响应会被切成多个 content 分片，必须直接拼接
                body = "".join(c.get("text", "") for c in result.get("content", []))
                if result.get("isError") or not body.strip():
                    raise RuntimeError("isError 或响应正文为空")
                data = json.loads(body)
                if "error" in data:
                    raise RuntimeError(data["error"])     # 如 size 超限、聚合被拒
                break
            except Exception:
                if attempt == 3:
                    raise
                time.sleep(2 ** attempt)                  # 退避重试

        hits = data["hits"]["hits"]
        if not hits:
            break
        for h in hits:                                    # 按 _id 去重
            if h["_id"] not in seen:
                seen.add(h["_id"])
                rows.append(h["_source"])
        if len(hits) < page_size:                         # 已到最后一页
            break
        page += 1

    return rows


# ---- 用法：拉取近 30 天的交易事件（最多 5 页） ----
from datetime import date, timedelta

since = (date.today() - timedelta(days=30)).isoformat()

rows = fetch_all("jingdata_large_transaction", {
    "query": {"bool": {"filter": [
        {"range": {"finance_date": {"gte": since}}}]}},
}, max_pages=5)

print(len(rows), "条")

{}

{
  "arg0": "jingdata_large_transaction"
}

{
  "arg0": "focus_phase"
}

{
  "query": {
    "term": {
      "name.keyword": "开放数据平台"
    }
  },
  "size": 10,
  "track_total_hits": true
}

{
  "query": {
    "term": {
      "name.keyword": "开放数据平台"
    }
  },
  "size": 10,
  "track_total_hits": true
}

{
  "arg0": "80",
  "arg3": 10
}

{
  "arg0": "legal-4qjgm3c6t2",
  "arg3": 10
}

[
  {
    "name": "人工智能产业链",
    "id": "chain-19b20d6a969kbxd",
    "child": {
      "name": "基础层",
      "id": "63",
      "child": {
        "name": "数据资源",
        "id": "67",
        "child": {
          "name": "开放数据平台",
          "id": "80"
        }
      }
    }
  }
]

{
  "query": {
    "bool": {
      "filter": [
        {
          "term": {
            "chain_id": "80"
          }
        },
        {
          "term": {
            "level": "1"
          }
        }
      ]
    }
  },
  "size": 10,
  "track_total_hits": true
}

{
  "query": {
    "match": {
      "name": "人工智能"
    }
  },
  "size": 10,
  "track_total_hits": true
}

{
  "query": {
    "match_all": {}
  },
  "size": 1,
  "track_total_hits": true
}

{
  "query": {
    "bool": {
      "filter": [
        {
          "range": {
            "finance_date": {
              "gte": "2026-09-28",
              "lt": "2026-09-29"
            }
          }
        }
      ]
    }
  },
  "sort": [
    {
      "finance_date": {
        "order": "desc"
      }
    },
    {
      "_id": {
        "order": "asc"
      }
    }
  ],
  "size": 10,
  "from": 0,
  "track_total_hits": true
}

{
  "query": {
    "term": {
      "name.keyword": "人工智能"
    }
  },
  "size": 3,
  "track_total_hits": true
}

{
  "query": {
    "nested": {
      "path": "tags_v2",
      "query": {
        "term": {
          "tags_v2.id": "tag-dccb0a7af2"
        }
      }
    }
  },
  "size": 10,
  "track_total_hits": true
}

{
  "query": {
    "term": {
      "address3": "440300"
    }
  },
  "size": 10,
  "track_total_hits": true
}

{
  "query": {
    "bool": {
      "filter": [
        {
          "range": {
            "finance_date": {
              "gte": "2026-01-01"
            }
          }
        },
        {
          "term": {
            "finance_phase": "30"
          }
        }
      ],
      "must": [
        {
          "nested": {
            "path": "investor",
            "query": {
              "match": {
                "investor.name": "红杉"
              }
            }
          }
        }
      ]
    }
  },
  "sort": [
    {
      "finance_date": {
        "order": "desc"
      }
    },
    {
      "_id": {
        "order": "asc"
      }
    }
  ],
  "size": 10,
  "track_total_hits": true
}

{
  "query": {
    "match_all": {}
  },
  "size": 0,
  "aggs": {
    "by_industry": {
      "terms": {
        "field": "industry_v2.name",
        "size": 10
      }
    }
  }
}