fish-edge-gateway-sdk (Python)

接入 apex-edge-gateway 的服务侧 SDK：manifest 声明与暴露、internal auth、trace 透传。协议契约见仓库 docs/contracts/。

安装（不发布 PyPI，直接 git 依赖）

# requirements.txt / pyproject dependency（锁 tag 或 commit，不要追 main）
fish-edge-gateway-sdk @ git+ssh://git@github.com/fishaudio/apex-edge-gateway.git@<tag-or-sha>#subdirectory=sdk-python

# 本地开发
pip install "fish-edge-gateway-sdk @ git+ssh://git@github.com/fishaudio/apex-edge-gateway.git@main#subdirectory=sdk-python"

# uv
uv add "fish-edge-gateway-sdk @ git+ssh://git@github.com/fishaudio/apex-edge-gateway.git@main#subdirectory=sdk-python"

# CI / Docker（https + token，避免挂 ssh key）
pip install "fish-edge-gateway-sdk @ git+https://${GITHUB_TOKEN}@github.com/fishaudio/apex-edge-gateway.git@<sha>#subdirectory=sdk-python"

版本锁定约定：SDK 变更打 sdk-python/vX.Y.Z 形式的 git tag，服务锁 tag 升级；紧急验证可锁 commit sha。

集成

from fastapi import Depends, FastAPI
from fish_edge_gateway_sdk import edge_route, mount_edge, require_gateway

app = FastAPI()

@app.post("/internal/asr", dependencies=[Depends(require_gateway)])
@edge_route(auth="internal")
async def asr(payload: dict) -> dict: ...

mount_edge(app, service="omni-asr")   # 挂载 GET /.edge/manifest

功能门禁（manifest 契约 §9B，仅 auth="api_key"）：调用方 team 不具备要求的 feature 时，gateway 直接回 403 feature_required，请求不会下落到其它路由，handler 无需再判断。feature 名不超过 64 个字符，每条路由合计不超过 4096 个。带门禁的 manifest 自动声明 schema_version: 2：先升级 gateway 再发布（旧 gateway 会拒绝 v2 manifest，而不是不设门禁地服务它）。

@app.post("/v1/tts-pro")
@edge_route(auth="api_key", protocol="bytes",
            requires_features=["tts-pro"],                     # 全部需要
            requires_any_of=[["voice-clone", "enterprise"]])   # 每组至少一个
async def tts_pro(...): ...

上报用量（Usage 与计费维度）

用量随响应走，gateway 读取后剥离，客户端不可见。Usage 是一个计费对象：tokens 加计费维度（如 model_id）。model_id 永远是一个字符串；归属多个模型 = 多个对象。所有 helper 都接受 int | Usage | Sequence[Usage]。

from fish_edge_gateway_sdk import Usage, usage_header, usage_json_field, usage_event, usage_reserve_event

usage_header(120)                                  # {"x-fish-usage-tokens": "120"}（无维度：与 0.2 字节一致）
usage_header([Usage(70).model_id("voice-alice"), Usage(50).model_id("voice-bob")])
# {"x-fish-usage": '[{"model_id":"voice-alice","tokens":70},{"model_id":"voice-bob","tokens":50}]'}

usage_json_field(Usage(120).model_id("voice-alice"))  # {"x-fish-usage": {"tokens": 120, "model_id": "voice-alice"}}

usage_event(usage=Usage(70).model_id("voice-alice"), seq=1, trace_id=tid)
# {"event": "x-fish-usage", "tokens": 70, "model_id": "voice-alice", "seq": 1, "idempotency_key": ...}
usage_event(usage=objs, seq=2, trace_id=tid)          # 多个对象：{"event": ..., "usage": [...], "seq": 2, ...}
usage_reserve_event(objs)                             # WS 预留，整组替换上一个预留
usage_sse_event(objs)                                 # SSE 终结事件：同上平铺/数组规则，无 seq；空序列 = tokens 0

• 一个对象：字段与 event/seq 平铺；多个：放在 usage 数组里。usage_final_event 的 total_tokens 是所有事件所有对象的 Σ tokens，count 是 usage 事件数。
• 维度 key 匹配 ^[a-z][a-z0-9_]{0,31}$，且不能是保留字段（event seq total_tokens count usage trace_id tokens idempotency_key quantity developer_id）；值只能是整数 / 字符串 / 布尔（不支持浮点）。违反时 Usage(...) / with_dim / model_id 抛 ValueError。Usage 不可变且可哈希。
• 空序列上报 tokens 0：这是显式的免费上报，不是缺失。usage_header / usage_json_field 也接受 tokens=n 关键字（与 usage 二选一）。
• x-fish-usage header 与数组形态需要支持它们的 gateway；不带维度的调用（usage_header(n)、usage_event(tokens=n, ...)）线上字节不变。

k8s Service 侧标注（manifest 契约 §2.1）：

metadata:
  annotations:
    fish.audio/edge-discovery: enable
    # 可选：manifest 端点与业务端口分离时
    # fish.audio/edge-discovery-port: "9090"
    # fish.audio/edge-discovery-url: /.edge/manifest

gateway 侧无需额外授权：发现 namespace 的成员资格就是信任边界（manifest 契约 §11），被发现的 manifest 其 route 直接编译。

形态按 Accept 协商的端点（protocol="negotiated"）

一个端点这次回 SSE、下次回裸音频或 JSON 时声明 protocol="negotiated"，gateway 按每个响应的 Content-Type 计费（manifest 契约 §9）。两条约束：

• WebSocket 形态是独立的 ws 路由：FastAPI WebSocket 端点上只接受 protocol="ws"（negotiated 会在 mount_edge 时报错），且要放在不同的 path 上——同 path + GET 的两条声明只有先声明的生效。完整例子见 examples/negotiated_demo.py。
• 免费形态显式报 0：不收费的响应发 usage_header(0)；什么都不发 = usage 缺失，gateway 计入 gateway_usage_missing_total 并告警（usage 契约 §8）。

上线顺序：所有 DC 的 gateway 升级到支持 negotiated 的版本之后，服务才能发布带 negotiated 的 manifest——旧 gateway 会拒掉整份 manifest（manifest 契约 §9 Rollout）。

环境变量

开发

pip install -e "sdk-python[dev]"
pytest sdk-python/tests -q