iboard/scripts/fetchers/eastmoney-mcp.fetcher.js

// 东方财富妙想 MCP 数据源
// 协议: StreamableHttp (MCP 2025-03-26 spec), HTTP POST + JSON-RPC 2.0
//   - 认证: header `em_api_key: <EM_API_KEY>`
//   - 妙想 MCP 是"自然语言驱动": 传 query 字符串, 内部 LLM 解析 + 调底层数据接口
//   - 重要: 每次调用 5-20s + LLM token 成本, 必须配合"DB 优先" (refresh 时已存在的指标跳过)
//
// 需要的 .env:
//   EM_API_KEY=xxx                  必填
//   EASTMONEY_MCP_BASE=...          可选, 默认 https://mxapi.eastmoney.com/mxds/mcp
//
// INDICATORS 数组: 配置要拉的指标
//   每项:
//     name      → Indicator.name (主键)
//     groupName → Indicator.groupName
//     title     → data.title
//     unit      → data.unit
//     tool      → MCP tool 名
//     query     → 自然语言查询
//     pickSheet(sheets) → 选一张 sheet (默认取第一张)
//     parseSheet(sheet) → [{period, value}, ...]   默认按宽表转长表 (列=日期, 行=值)

import axios from 'axios'

const DEFAULT_BASE = 'https://mxapi.eastmoney.com/mxds/mcp'
const TIMEOUT = 60000

// ====== INDICATORS 配置 ======
// 跑 npm run data:discover 看可用 tool, 再加新指标
const INDICATORS = [
  {
    name: 'china-cpi-yoy',
    groupName: 'china-macro',
    title: '中国 CPI 同比',
    unit: '%',
    tool: 'mx_macro_data',
    query: '查询中国 2015 年至今 CPI 月度同比数据，每一个月的数值',
    pickSheet: (sheets) =>
      sheets.find((s) => s.items?.[0]?.[0]?.includes('CPI') && s.items[0][0].includes('同比'))
      || sheets.find((s) => s.sheetName?.includes('宏观数据'))
      || sheets[0],
  },
  {
    name: 'china-gdp-yoy',
    groupName: 'china-macro',
    title: '中国 GDP 同比',
    unit: '%',
    tool: 'mx_macro_data',
    query: '查询中国 2015-2025 年 GDP 不变价同比增速，季度数据，每一季度的数值',
    pickSheet: (sheets) =>
      sheets.find((s) => s.items?.[0]?.[0]?.includes('GDP') && s.items[0][0].includes('同比'))
      || sheets[0],
  },
]

export const META = {
  name: 'eastmoney-mcp',
  description: '东方财富妙想 MCP (StreamableHttp, NL 查询)',
  docs: 'https://mxapi.eastmoney.com/mxds/doc/mcp-install.md',
}

// ====== StreamableHttp 客户端 ======

class McpClient {
  constructor(baseUrl, apiKey) {
    this.baseUrl = baseUrl
    this.apiKey = apiKey
    this.id = 0
  }
  async _request(method, params) {
    const payload = { jsonrpc: '2.0', id: ++this.id, method, ...(params ? { params } : {}) }
    const resp = await axios.post(this.baseUrl, payload, {
      timeout: TIMEOUT,
      headers: {
        'Content-Type': 'application/json',
        'Accept': 'application/json, text/event-stream',
        'em_api_key': this.apiKey,
      },
      responseType: 'text',
      transformResponse: [(d) => d],
      validateStatus: () => true,
    })
    if (resp.status >= 400) {
      throw new Error(`HTTP ${resp.status}: ${String(resp.data).slice(0, 300)}`)
    }
    const parsed = parseMcpResponse(String(resp.data || ''))
    // JSON-RPC 信封 {jsonrpc,id,result|error}, 统一解包 result
    if (parsed.error) {
      throw new Error(`MCP error ${parsed.error.code}: ${parsed.error.message}`)
    }
    return parsed.result || parsed
  }
  async listTools() {
    const r = await this._request('tools/list')
    return r.tools || []
  }
  async callTool(name, args = {}) {
    const r = await this._request('tools/call', { name, arguments: args })
    if (r.isError) {
      const msg = (r.content || []).map((c) => c.text || '').join('; ')
      throw new Error(`tool ${name} error: ${msg || JSON.stringify(r).slice(0, 200)}`)
    }
    return r
  }
}

// 解析响应: JSON 或 SSE
function parseMcpResponse(text) {
  const t = text.trim()
  if (t.startsWith('{')) return JSON.parse(t)
  if (t.startsWith('data:')) {
    const lines = t.split('\n').filter((l) => l.startsWith('data:'))
    let last = null
    for (const line of lines) {
      const json = line.slice(5).trim()
      if (!json || json === '[DONE]') continue
      try { last = JSON.parse(json) } catch { /* skip */ }
    }
    if (!last) throw new Error('SSE 无有效 data 块: ' + text.slice(0, 200))
    return last
  }
  throw new Error('未知响应格式: ' + t.slice(0, 200))
}

// 默认 sheet 解析: 宽表 (列=日期, 行=值) → 长表
// 例: { columns:["指标","来源","2024-12","2024-11"], items:[["CPI","统计局","0.1","0.2"]] }
//   → [{period:"2024-12", value:0.1}, {period:"2024-11", value:0.2}]
// 返回时反转成从老到新
function defaultParseSheet(sheet) {
  if (!sheet || !sheet.columns || !sheet.items || !sheet.items[0]) return []
  const cols = sheet.columns
  const row = sheet.items[0]
  const dates = cols.slice(2)         // 跳过 [指标名, 来源]
  const values = row.slice(2)
  const out = []
  for (let i = 0; i < dates.length; i++) {
    const v = parseFloat(values[i])
    if (!isNaN(v)) out.push({ period: dates[i], value: v })
  }
  return out.reverse()  // 从老到新
}

// ====== fetcher 入口 ======

export async function listAvailableTools() {
  const key = process.env.EM_API_KEY
  if (!key) throw new Error('缺少 EM_API_KEY')
  const base = process.env.EASTMONEY_MCP_BASE || DEFAULT_BASE
  const client = new McpClient(base, key)
  return client.listTools()
}

// opts.skipNames: Set<string> — DB 已存在的指标名, 跳过不调 MCP (省 token)
// opts.onlyNames: Set<string> — 白名单, 只跑这些 (用于补跑单个)
export async function fetchAll(opts = {}) {
  if (INDICATORS.length === 0) {
    throw new Error('eastmoney-mcp: INDICATORS 数组为空, 加新指标见文件顶部注释')
  }
  const skip = opts.skipNames || new Set()
  const only = opts.onlyNames || null

  const key = process.env.EM_API_KEY
  if (!key) throw new Error('缺少 EM_API_KEY')
  const base = process.env.EASTMONEY_MCP_BASE || DEFAULT_BASE
  const client = new McpClient(base, key)

  const results = []
  for (const cfg of INDICATORS) {
    if (skip.has(cfg.name)) {
      console.log(`[MCP] ${cfg.name} 跳过 (DB 已存在)`)
      continue
    }
    if (only && !only.has(cfg.name)) continue
    try {
      console.log(`[MCP] ${cfg.name} 调用 ${cfg.tool}: ${cfg.query}`)
      const r = await client.callTool(cfg.tool, { query: cfg.query })
      const text = (r.content || []).map((c) => c.text || '').join('')
      let payload
      try { payload = JSON.parse(text) } catch { payload = { raw: text } }
      const sheets = payload.data || (Array.isArray(payload) ? payload : [payload])
      const sheet = cfg.pickSheet ? cfg.pickSheet(sheets) : sheets[0]
      const parsed = cfg.parseSheet ? cfg.parseSheet(sheet) : defaultParseSheet(sheet)
      if (parsed.length === 0) {
        console.log(`[MCP] ${cfg.name} 返回空 rows, 跳过`)
        continue
      }
      results.push({
        name: cfg.name,
        groupName: cfg.groupName,
        data: {
          unit: cfg.unit,
          title: cfg.title,
          source: 'eastmoney-mcp',
          rows: parsed,
        },
      })
      console.log(`[MCP] ${cfg.name} OK, ${parsed.length} rows (${parsed[0]?.period}..${parsed.at(-1)?.period})`)
    } catch (e) {
      console.error(`[MCP] ${cfg.name} FAIL: ${e.message}`)
    }
  }
  // 全部被 DB 优先跳过时, 返回显式标记而不是空数组 (避免被误判为失败)
  if (results.length === 0 && INDICATORS.every((cfg) => skip.has(cfg.name))) {
    return { skipped: true, reason: 'DB 已存在全部指标' }
  }
  return results
}