技术教程 · 阅读约 10 分钟

把 Codex 接入飞书?不如先搞懂怎么用 API 自建一个团队共享 AI 助手

把 Codex 接入飞书?不如先搞懂怎么用 API 自建一个团队共享 AI 助手

最近 V2EX 上有个帖子让我来了兴趣:CodexBridge,把 OpenAI Codex 变成飞书里的团队共享助手,支持多智能体、权限控制、额度审计。

听起来很酷。但实际上手之后发现,最难的不是接飞书,而是"团队共享"这件事本身——你得有一个稳定、可控、能统一计费的 API 入口,才能谈后面的权限和审计。

这篇文章就从这个核心问题出发,手把手带你搭一个最小可用的「团队 AI 网关」:统一 base URL、按 key 隔离额度、记录每次调用日志。Python,不到 80 行,直接能跑。


痛点:为什么团队共享 AI 这么难搞?

个人用 ChatGPT 或者 API 很简单,但一到团队就头疼:

  • 每人自己充值,报销一团乱麻
  • key 泄露了不知道是谁干的
  • 某个同事写了个死循环,一晚上烧掉几百块
  • 想限制实习生只能用 GPT-4o mini,做不到

本质上缺一个中间层:统一出口、分发虚拟 key、记日志、控额度。


架构设计(极简版)


团队成员

   │  虚拟 key(自己生成)

   ▼

[你的网关服务]  ←── FastAPI,本地或内网部署

   │  真实 API key

   ▼

[上游 LLM API]  ←── api2everything.xyz/v1(兼容 OpenAI 格式)

网关做三件事:

1. 验证虚拟 key,找到对应成员

2. 检查余额/额度

3. 转发请求 + 写日志


完整代码

依赖先装好:


pip install fastapi uvicorn httpx

新建 gateway.py:


import time

import uuid

import httpx

from fastapi import FastAPI, Request, HTTPException

from fastapi.responses import StreamingResponse



app = FastAPI()



# ── 配置区 ──────────────────────────────────────────

UPSTREAM_BASE_URL = "https://api2everything.xyz/v1"

UPSTREAM_API_KEY  = "sk-你的无量Api密钥"   # 统一用一个真实 key



# 虚拟 key 表:key -> {name, quota_remaining, total_used}

VIRTUAL_KEYS = {

    "team-alice-001": {"name": "Alice",  "quota": 10.0, "used": 0.0},

    "team-bob-002":   {"name": "Bob",    "quota":  5.0, "used": 0.0},

    "team-intern-003":{"name": "Intern", "quota":  1.0, "used": 0.0},

}



# 简易价格表(元/1k tokens),仅做演示

MODEL_PRICE = {

    "gpt-4o":          0.018,

    "gpt-4o-mini":     0.001,

    "claude-sonnet-4": 0.022,

    "deepseek-chat":   0.002,

}

# ────────────────────────────────────────────────────





def estimate_cost(model: str, prompt_tokens: int, completion_tokens: int) -> float:

    price = MODEL_PRICE.get(model, 0.01)

    return (prompt_tokens + completion_tokens) / 1000 * price





def check_key(virtual_key: str) -> dict:

    if virtual_key not in VIRTUAL_KEYS:

        raise HTTPException(status_code=401, detail="Invalid virtual key")

    member = VIRTUAL_KEYS[virtual_key]

    if member["used"] >= member["quota"]:

        raise HTTPException(

            status_code=429,

            detail=f"{member['name']} 额度已用完(上限 ¥{member['quota']})"

        )

    return member





@app.post("/v1/chat/completions")

async def proxy_chat(request: Request):

    # 1. 取虚拟 key(从 Authorization header)

    auth = request.headers.get("Authorization", "")

    virtual_key = auth.removeprefix("Bearer ").strip()

    member = check_key(virtual_key)



    # 2. 读请求体

    body = await request.json()

    model = body.get("model", "gpt-4o-mini")



    # 3. 转发到上游

    headers = {

        "Authorization": f"Bearer {UPSTREAM_API_KEY}",

        "Content-Type": "application/json",

    }



    start = time.time()

    async with httpx.AsyncClient(timeout=120) as client:

        resp = await client.post(

            f"{UPSTREAM_BASE_URL}/chat/completions",

            json=body,

            headers=headers,

        )



    elapsed = round(time.time() - start, 2)



    if resp.status_code != 200:

        raise HTTPException(status_code=resp.status_code, detail=resp.text)



    data = resp.json()



    # 4. 计费 & 日志

    usage = data.get("usage", {})

    cost  = estimate_cost(

        model,

        usage.get("prompt_tokens", 0),

        usage.get("completion_tokens", 0),

    )

    member["used"] = round(member["used"] + cost, 4)



    print(

        f"[{time.strftime('%H:%M:%S')}] "

        f"{member['name']} | {model} | "

        f"¥{cost:.4f} | 累计¥{member['used']:.3f}/{member['quota']} | {elapsed}s"

    )



    return data





@app.get("/admin/usage")

async def show_usage():

    """简单查看各成员用量"""

    return {

        k: {

            "name":  v["name"],

            "used":  v["used"],

            "quota": v["quota"],

            "pct":   f"{v['used']/v['quota']*100:.1f}%",

        }

        for k, v in VIRTUAL_KEYS.items()

    }

启动:


uvicorn gateway:app --host 0.0.0.0 --port 8080


测试一下


from openai import OpenAI



# 团队成员只需要知道网关地址和自己的虚拟 key

client = OpenAI(

    base_url="http://localhost:8080/v1",

    api_key="team-alice-001",          # 虚拟 key

)



resp = client.chat.completions.create(

    model="gpt-4o-mini",

    messages=[{"role": "user", "content": "用一句话解释什么是向量数据库"}],

)

print(resp.choices[0].message.content)

查看用量:


curl http://localhost:8080/admin/usage

输出大概长这样:


{

  "team-alice-001": {"name": "Alice", "used": 0.003, "quota": 10.0, "pct": "0.0%"},

  "team-bob-002":   {"name": "Bob",   "used": 0.0,   "quota": 5.0,  "pct": "0.0%"}

}


扩展方向

这个 80 行版本跑通之后,可以继续加:

  • 持久化:把 VIRTUAL_KEYS 换成 SQLite 或 Redis,重启不丢数据
  • 流式输出:把 httpx 换成流式转发,支持 stream=True
  • 飞书 Webhook:额度告警时推飞书消息(这就是 CodexBridge 在做的事)
  • 模型白名单:Intern 的 key 只允许调 gpt-4o-mini,在 check_key 里加一行判断

关于 API 费用

代码里的 UPSTREAM_BASE_URL 用的是 无量Api,国内直连,不用科学上网,支持 OpenAI/Claude/Gemini/DeepSeek 等 300+ 模型,比官方便宜大约 65%。

对团队网关这种场景来说,API 费用是长期成本,省 60% 不是小数目。注册还送 ¥1 可以先跑通流程,余额永久有效不会过期。

直达:api2everything.xyz

有问题欢迎评论区聊,特别是流式转发那块坑比较多,我可以单独写一篇。觉得有用的话点个赞,下期写「给网关加速率限制和 JWT 鉴权」。