Python 手写 SSE 客户端：对接华为云 CodeArts 智能问答接口的三个坑

最近要把华为云码道（CodeArts）网页端的智能问答接到自己的脚本里，官方 SDK 并没有覆盖这个能力，只能对着网页端的真实请求一点点拼。整趟走下来踩的坑集中在三类：请求头、Cookie 作用域、SSE 读取。这里把结论和可复现代码整理出来，供同样要用 Python 调流式接口的同学参考。

一、少一个请求头就换一种报错

码道问答的接口对普通浏览器指纹并不买账，必须显式带上几个业务头，缺哪个报哪个：

POST https://devcloud.cn-north-4.huaweicloud.cn/chat/codebaseservice/v1/cloudagent/sessions/{sid}/messages
Content-Type: application/json
cftk: <Cookie devclouddevuibjtcftk 的同值>
maas-type: benefit
agent-type: CodeBase
x-codearts-doer-scenario-type: web
x-requested-with: XMLHttpRequest

实测表现很有迷惑性：不带 cftk 时返回 400 且提示 csrf validate fail，看起来像登录态问题，其实是缺 CSRF 令牌头；带上 cftk 但不带 maas-type: benefit，则返回「未找到模型或当前未登录」。后者尤其坑——模型列表接口是能正常返回的，只有发消息这一步报错，很容易误判成账号权限问题。

正确做法是用浏览器同源请求做一次基线：把网页端真实发出的请求头逐条打出来对比，而不是猜。cftk 这类令牌要从对应 host 的 Cookie 里取，而不是从别处复制。

二、同名 Cookie 在不同域各有一份

华为云体系的 Cookie 是按域下发的，JSESSIONID、csrf-eco 这类名字在 .huaweicloud.cn、.huaweicloud.cn 的子域、以及具体 host 上可能各有一份，且值不同。如果图省事把它们拍平写进同一个 Cookie 文件，请求时最窄域的那份会被后写入的覆盖，服务端就会报 CSRF unequal 或者直接判定未登录。

稳妥的写法是按目标 host 过滤，同名取域名最宽的一份：

def jar_for(rows, host):
    best = {}
    for c in rows:                      # rows 来自 CDP Storage.getCookies
        d = (c.get("domain") or "").lstrip(".")
        if not (host == d or host.endswith("." + d)):
            continue
        depth = d.count(".")
        cur = best.get(c["name"])
        if cur is None or depth < cur[0]:
            best[c["name"]] = (depth, c["value"])   # 域名越宽，点越少
    return {k: v[1] for k, v in best.items()}

另一个细节：csrf 请求头的正确取值不一定等于 csrf-eco Cookie，有些站点是在 GET /api/get-ainfo 的响应头里现给的。以响应头为准，Cookie 值只当兜底，能避开一大半「明明登录了却报 CSRF」的怪问题。

三、用 requests 读 SSE 的两个关键点

流式接口返回的是标准 SSE 帧（event: 与 data: 成对），必须 stream=True 配 iter_lines，绝不能用 resp.text，那会等整个响应结束才返回，流式就变成了「卡两分钟然后超时」。

with s.post(msg_url % sid, data=body, headers={"Accept": "text/event-stream"},
            stream=True, timeout=120) as resp:
    deadline, event, parts = time.time() + 90, "", []
    for line in resp.iter_lines(decode_unicode=True):
        if not line:
            continue
        line = line.strip()
        if line.startswith("event:"):
            event = line.split(":", 1)[1].strip()
        elif line.startswith("data:"):
            d = json.loads(line.split(":", 1)[1].strip())
            if event == "error":
                return False, d.get("error")          # 出错帧要立刻退出
            if event == "message":
                parts.append(d.get("content") or "")
        if time.time() > deadline:
            break
    return True, "".join(parts)

第二个关键点是事件类型要分开处理。码道一次问答会依次推 status、session_title、若干 thought、再是 message，最后可能给 file_diff。只累积 message 才是答案；把 thought 也拼进去，会得到一大段推理过程，token 统计也会虚高——一次几十上百字的提问消耗掉三万多 token，多数就是虚高统计造成的错觉，把 benefit/tokens/balance 的 daily_tokens_used 拿出来对一遍就清楚了。

小结

整条链路本身没有难点：建会话 → 发消息 → 按行解析事件 → 读额度接口校验。真正花时间的全是「报错信息与真实原因不对应」的环节：CSRF 提示指向登录、未登录提示指向模型。经验是两步：先用同源浏览器请求建立正确的头部基线，再严格按 host 作用域管理 Cookie。另外流式客户端建议始终带一个显式 deadline 与「出错帧立即退出」，否则异常会被静默的长连接拖成超时，日志里什么线索都留不下。