# MCP JSON-RPC 直连模式（Feishu Project MCP）

## 适用场景

需要批量拉取 MCP 数据（>5页）时，直连比通过 Hermes 工具逐页调用快得多。单页 1.3s，复用 session。

## Quick Start

```bash
# Token 位置
cat ~/.hermes/config.yaml | grep -A2 feishu_project
# 复制 mcp.feishu_project.token 值
```

## 字段值提取（按 value_type）

MCP 返回的 `moql_field_list` 中每个字段的 `value_type` 决定如何取值：

| value_type | 值结构 | 提取代码 |
|-----------|--------|---------|
| `string_value` | `{"string_value": "xxx"}` | `v.get("string_value","")` |
| `key_label_value` | `{"key_label_value": {"label":"xxx"}}` | `(v.get("key_label_value") or {}).get("label","")` |
| `key_label_value_list` | `{"key_label_value_list": [{...}]}` | labels逗号连接 |
| `user_value` | `{"user_value": {"name_cn":"xxx","name_en":"xxx"}}` | `uv.get("name_cn","")` **← 注意是name_cn不是name** |
| `double_value` | `{"double_value": 123.45}` | `v.get("double_value","")` |
| `number` 类型 | 同上 | 同上 |

### 通用提取函数

```python
def extract_val(field):
    """通用MCP字段值提取"""
    vt = field.get("value_type", "")
    v = field.get("value") or {}
    if vt == "string_value":
        return v.get("string_value", "")
    if vt == "key_label_value":
        return (v.get("key_label_value") or {}).get("label", "")
    if vt == "key_label_value_list":
        items = v.get("key_label_value_list") or []
        return ",".join(i.get("label", "") for i in items)
    if vt == "user_value":
        uv = v.get("user_value", {})
        return uv.get("name_cn", "") or uv.get("name", "")
    if vt == "double_value":
        return v.get("double_value", "")
    if vt == "long_value":
        return v.get("long_value", 0)
    # fallback: try to get first dict value
    vals = list(v.values())
    if vals:
        if isinstance(vals[0], str): return vals[0]
        if isinstance(vals[0], dict): return vals[0].get("name_cn","") or vals[0].get("label","") or str(vals[0])
    return ""
```

## 完整脚本模板

```python
#!/usr/bin/env python3
import json, urllib.request, uuid, re

URL = "https://project.feishu.cn/mcp_server/v1"
TOKEN = "m-xxxxx"  # 从 ~/.hermes/config.yaml 获取
PROJECT_KEY = "6593cd71471290e3cc6be6e6"

MQL_TMPL = ("SELECT `物流单号`, `字段2`, ... FROM `销售管理`.`销售订单` "
            "WHERE `物流单号` is not null AND `状态` not in ('已结束','已终止') "
            "AND `创建时间` >= '2026-06-24' ORDER BY `创建时间` DESC LIMIT 50 OFFSET {offset}")

session_headers = {
    "Content-Type": "application/json",
    "Accept": "application/json, text/event-stream",
    "X-Mcp-Token": TOKEN,
}

def rpc(payload, sid=None):
    headers = dict(session_headers)
    if sid: headers["Mcp-Session-Id"] = sid
    req = urllib.request.Request(URL, data=json.dumps(payload).encode(),
                                  headers=headers, method="POST")
    resp = urllib.request.urlopen(req, timeout=120)
    sid_out = resp.headers.get("Mcp-Session-Id", sid)
    raw = resp.read().decode("utf-8", "replace")
    texts = []
    if "data:" in raw:
        for line in raw.splitlines():
            if line.startswith("data:"): texts.append(line[5:].strip())
        raw = "\n".join(texts)
    return json.loads(raw), sid_out

# 1. Initialize session
resp, sid = rpc({"jsonrpc":"2.0","id":1,"method":"initialize",
    "params":{"protocolVersion":"2024-11-05","capabilities":{},
              "clientInfo":{"name":"my-script","version":"1.0"}}})

# 2. Send initialized notification
h = dict(session_headers)
if sid: h["Mcp-Session-Id"] = sid
try:
    urllib.request.urlopen(urllib.request.Request(URL,
        data=json.dumps({"jsonrpc":"2.0","method":"notifications/initialized"}).encode(),
        headers=h, method="POST"), timeout=30).read()
except: pass  # 202 is OK

# 3. Batch query
records = {}
for offset in range(0, 500, 50):
    call = {"jsonrpc":"2.0","id":offset//50+1,"method":"tools/call",
            "params":{"name":"search_by_mql",
                      "arguments":{"project_key":PROJECT_KEY,
                                   "mql":MQL_TMPL.format(offset=offset)}}}
    resp, sid = rpc(call, sid)
    if "error" in resp: continue

    content = resp["result"]["content"][0]["text"]
    m = re.search(r"\{.*\}", content, re.S)
    inner = json.loads(m.group(0))
    if "result" in inner:
        inner = json.loads(inner["result"]) if isinstance(inner["result"],str) else inner["result"]
    else:
        inner = inner

    for row in inner["data"]["1"]:
        f = {}
        for x in row["moql_field_list"]:
            f[x["name"]] = extract_val(x)
        # ... dedupe + store
```

## 关键陷阱

1. **Token位置**：`~/.hermes/config.yaml` → `mcp.feishu_project.token`
2. **SSE 解析**：响应可能是 SSE 格式，需提取 `data:` 行
3. **JSON 解析用 raw_decode，不用 regex**：`re.search(r"\\{.*\\}", content)` 在某些响应格式下会返回 None → 用 `JSONDecoder.raw_decode(content.strip())` 更稳健
4. **user_value vs user_value_list**：
   - 普通用户字段(`创建者`)：`value_type=user_value` → `v.get("user_value",{}).get("name_cn","")`
   - 角色字段(`__下单人`, `__当前负责人`)：`value_type=user_value_list` → 取 `user_value_list[0].name_cn`
5. **name_cn 不是 name**：MCP 返回的用户中文字段是 `name_cn`，用 `name` 会返回空
6. **会话过期刷新**：单 session 约 30-35 页后开始返回空数据 → 每 20 页重建 session（`init_session()` 重调 initialize + notifications/initialized）
7. **分页方式**：只能用 `LIMIT offset, count`，`group_pagination_list` 不可靠（返回重复数据）
8. **每页限制**：50条/页
9. **写入前备份**：`shutil.copy2(OUTPUT, OUTPUT + ".bak")` 防止会话过期后空数据覆盖目标文件