如果你最近在折腾 Codex 或 Claude Code应该已经感觉到模型能力不是瓶颈配额才是。我这边的情况更具体——团队里的自动化任务经常要调 Antigravity 去跑 Agent 工作流而 Antigravity 的平台配额又会和 Codex / Claude Code 的调用挤在一起两边互相抢。第一个版本我用官方 API 直连结果半天就被 429 打崩后来我换了个思路不直连而是在中间加一层可复用的封装把 Antigravity 包装成 Codex 和 Claude Code 都能调用的外部 Skill。这篇文章就是这套方案的完整复盘从架构选择到代码实现、再到配额优化策略全部来自实际部署经验。先把话说在前面标题里的“突破配额限制”不是去破解服务商的限流而是通过架构手段把有限的配额用得更聪明——干掉重复调用、抹平请求毛刺、在合法前提下做资源调度。这是工程问题不是薅羊毛问题。下面进入正题。1. 先把事情说透Antigravity 与编程 Agent 的配额困局1.1 Antigravity 平台到底能做什么Antigravity 是谷歌云生态里的一个 AI Agent 开发平台主打“用自然语言描述任务系统自动编排工作流并执行”。你可以创建 Agent 处理邮件、整理文档、搜集资料、跑数据管道也可以把它嵌入到自己的系统里。对开发者来说最值钱的是它把任务拆解、工具调用、执行结果聚合这些脏活都包了你只需要定义目标和输入参数。我实际用它跑过两类事情一类是周期性的信息汇总比如每天把团队邮件里的待办事项抽出来整理成结构化清单另一类是跨系统的动作类任务比如根据表格数据生成周报并触发后续流程。这两类任务如果让 Codex 或 Claude Code 自己写代码来做每次都要重新组织逻辑而 Antigravity 的优势在于工作流是平台托管的状态和结果都不需要我维护。1.2 配额限制的真实面貌Antigravity 的配额限制通常不是单一维度而是从每分钟请求数RPM、每日请求数DPD、并发数、积分余额等多个角度同时卡你。我实测下来最难受的是 RPM 太低——如果你让 Codex 一口气并行调用多个 Agent 任务很容易在几秒内就把每分钟配额打满然后整条流水线开始连续报 429。更麻烦的是配额限制往往对“创建 Agent / 工作流”和“执行工作流”分别计数两者的消耗比例不一样。创建类请求虽然频率低但单次消耗高执行类请求频率高容易被 RPM 卡住。这种不对称性导致你在设计封装层时必须区别对待不能用一个限流策略覆盖所有操作。1.3 为什么要在 Codex / Claude Code 里外接一个 SkillCodex 和 Claude Code 默认不具备 Antigravity 的能力。你可以让它们写代码去调 Antigravity API但每次都要处理鉴权、重试、错误解析代码会变得很难维护而且每个项目都要复制一遍。封装成外部 Skill 之后Agent 只负责在合适的时机“呼叫”这个 Skill剩下的鉴权、限流、缓存、错误处理全部由封装层承担。编程 Agent 的注意力应该放在代码逻辑上而不是放在 API 调用细节上。这也是我最终选择“中间代理服务 MCP / Skill 注册”而不是“项目内直接调 SDK”的根本原因。2. 整体方案设计为什么是“代理服务 MCP / Skill”双通道2.1 三种封装方式横评我在动手之前把可能的封装方式列了一遍权衡之后才定的方案。这里直接分享对比结论。方案Claude Code 适配性Codex 适配性统一限流与缓存维护成本官方 SDK / 项目内直连需要写胶水代码需要写胶水代码不支持低自建 HTTP 代理服务通过 MCP 接入通过 CLI / Webhook 接入支持中自建 MCP Server原生支持不支持需额外适配支持中最终我选了“自建 HTTP 代理服务”这条路线因为它的适配面最广Claude Code 通过 MCP 协议接入Codex 通过 CLI 包装后的工具声明接入两边共用同一个代理核心限流、缓存、配额调度都在一个地方维护。2.2 架构总览一次调用是怎么走完的整个链路的调用顺序是这样的Codex 或 Claude Code 判断当前任务需要调用 AntigravityAgent 调用本地注册的 Skill / MCP 工具发起 HTTP 请求到代理服务代理服务先查缓存命中就直接返回完全消耗 Antigravity 配额缓存未命中进入限流队列限流器按 RPM 的 80% 放行代理服务携带 API Key 调用 Antigravity 平台结果返回后代理服务把数据写入缓存再把结果交回给 Agent。我在实际部署中发现如果不加缓存第 3 步永远拿不到命中数据配额消耗会直线上升加了缓存之后很多重复性任务的调用直接消失效果非常明显。2.3 关键决策自建代理而不是直连官方 SDK直连官方 SDK 看起来更简单但有两个问题绕不开。第一Agent 进程里必须保存 API Key安全性差尤其多人共用开发机的时候第二每个项目都要写一套限流和重试根本维护不过来。自建代理的核心优势在于“策略集中化”。限流参数、缓存规则、配额调度全在一个服务里调整Agent 侧只需要知道“调用这个 Skill 能拿到结果”。我选择用 FastAPI 来实现代理服务因为它原生支持异步而且 pydantic 可以直接校验请求参数省掉一堆手写解析代码。建议代理服务不要做得太重。能用环境变量解决的不要上数据库能单机跑的不要上服务编排先把链路跑通再考虑扩展。3. 动手实战从零搭建 Antigravity 外部调用 Skill3.1 第一步准备凭证与初始化工程先在 Antigravity 控制台创建项目拿到 API Key并记下你的 API 端点地址。不同的区域和项目版本端点格式不一样以控制台里的实际地址为准别抄网上的固定 URL。工程初始化我用 Python 虚拟环境依赖只要四个fastapi、uvicorn、httpx、pydantic。MCP Server 部分额外装mcp。mkdir antigravity-bridge cd antigravity-bridge python -m venv .venv source .venv/bin/activate pip install fastapi uvicorn httpx pydantic mcp然后在项目根目录创建.env文件ANTIGRAVITY_BASE_URLhttps://your-antigravity-endpoint ANTIGRAVITY_API_KEYyour_api_key_here RPM_LIMIT60 CACHE_TTL300 PORT8010RPM_LIMIT的值我建议设成你实际配额额度的 80%留出缓冲给非预期请求避免代理服务本身变成新的限流点。3.2 第二步搭一个本地代理服务代理服务的核心逻辑是三层缓存检查、限流、转发调用。下面是我实际在用的核心代码做了简化但保留了完整流程。# app.py import asyncio import hashlib import itertools import json import logging import os import time from contextlib import asynccontextmanager import httpx from fastapi import FastAPI, HTTPException from pydantic import BaseModel logging.basicConfig(levellogging.INFO) logger logging.getLogger(antigravity-bridge) ANTIGRAVITY_BASE os.getenv(ANTIGRAVITY_BASE_URL, ) ANTIGRAVITY_API_KEY os.getenv(ANTIGRAVITY_API_KEY, ) CACHE_TTL int(os.getenv(CACHE_TTL, 300)) RPM_LIMIT int(os.getenv(RPM_LIMIT, 60)) PORT int(os.getenv(PORT, 8010)) cache {} key_pool itertools.cycle( [k.strip() for k in os.getenv(ANTIGRAVITY_API_KEYS, ANTIGRAVITY_API_KEY).split(,) if k.strip()] ) app FastAPI(titleantigravity-bridge) class RunRequest(BaseModel): workflow: str params: dict {} class RateLimiter: def __init__(self, rate: float): self.min_interval 1.0 / rate self._last 0.0 self._lock asyncio.Lock() async def acquire(self): async with self._lock: now time.monotonic() wait self.min_interval - (now - self._last) if wait 0: await asyncio.sleep(wait) now time.monotonic() self._last now limiter RateLimiter(RPM_LIMIT / 60) async def call_antigravity(workflow: str, params: dict): api_key next(key_pool) async with httpx.AsyncClient(timeout120) as client: resp await client.post( f{ANTIGRAVITY_BASE}/v1/workflows/{workflow}:run, headers{Authorization: fBearer {api_key}}, jsonparams, ) if resp.status_code 429: logger.warning(rate limited, body%s, resp.text[:200]) resp.raise_for_status() return resp.json() def cache_key(workflow: str, params: dict) - str: raw f{workflow}|{json.dumps(params, sort_keysTrue)} return hashlib.sha256(raw.encode()).hexdigest() app.post(/tools/antigravity/run) async def run_tool(req: RunRequest): key cache_key(req.workflow, req.params) if key in cache and time.time() - cache[key][ts] CACHE_TTL: logger.info(cache hit: %s, key) return {cache: True, workflow: req.workflow, result: cache[key][data]} async with limiter: try: result await call_antigravity(req.workflow, req.params) except Exception as exc: raise HTTPException(status_code502, detailstr(exc)) from exc cache[key] {data: result, ts: time.time()} return {cache: False, workflow: req.workflow, result: result} if __name__ __main__: import uvicorn uvicorn.run(app, host127.0.0.1, portPORT)这段代码有几个细节值得说明。key_pool用itertools.cycle轮流取 key后面配额优化会用到RateLimiter用asyncio.Lock保证并发安全防止代理服务本身被打穿缓存的 key 使用可排序的 JSON 序列化避免参数顺序不同导致缓存失效。3.3 第三步把服务注册为 Claude Code 的 MCPClaude Code 对 MCP 的支持比较成熟我们可以把上面的 HTTP 服务封装成一个 MCP Server让 Claude Code 直接认识它。关键是写一个 FastMCP 入口文件把 HTTP 调用包成工具函数。# antigravity_mcp_server.py import httpx from mcp.server.fastmcp import FastMCP mcp FastMCP(antigravity) BRIDGE_URL http://127.0.0.1:8010 mcp.tool() def run_workflow(workflow: str, params: dict {}) - str: 运行 Antigravity workflow返回执行结果 JSON。 resp httpx.post( f{BRIDGE_URL}/tools/antigravity/run, json{workflow: workflow, params: params}, timeout180, ) resp.raise_for_status() return resp.text if __name__ __main__: mcp.run()这里有个易错点MCP 工具函数的返回类型必须是字符串不能直接返回 dict否则 Claude 可能无法解析。我在开发时踩过一次坑Claude 一直提示“工具返回格式无效”改成resp.text字符串返回后问题立刻消失。注册方式有两种。一种是用命令行直接注册claude mcp add antigravity --command python --args -m antigravity_mcp_server另一种是在项目根目录写.mcp.json这样项目所有成员共用配置{ mcpServers: { antigravity: { command: python, args: [-m, antigravity_mcp_server], env: { BRIDGE_URL: http://127.0.0.1:8010 } } } }注册完成后在 Claude Code 里输入claude mcp list就能看到antigravity已连接。我在 mac 上实测用 stdio 传输最稳HTTP transport 偶尔会因端口占用导致重连失败。3.4 第四步把服务注册为 Codex 的 SkillCodex 的 Skill 机制更偏向“声明式工具定义”。做法很简单在项目里放一个可执行的 CLI 脚本然后在AGENTS.md里声明这个工具的存在和用法。先写一个最简 CLI 封装脚本#!/usr/bin/env bash # antigravity-cli set -euo pipefail TASK${1:-} curl -s -X POST ${ANTIGRAVITY_BRIDGE_URL:-http://127.0.0.1:8010}/tools/antigravity/run \ -H Content-Type: application/json \ -d {\workflow\:\$TASK\,\params\:{}}给它加执行权限chmod x antigravity-cli然后在AGENTS.md里声明## Tools ### antigravity-cli 当任务描述为具体的外部 Agent 执行任务如整理邮件、汇总文档、运行工作流时必须调用 antigravity-cli task_description 注意不要替用户想象任务结果必须真实调用后返回。同一任务不要短时间重复调用。关键点是“必须调用”这几个字。Codex 本质上是按概率选择工具的 Agent如果描述写得模棱两可它很可能选择自己动手写代码而不是调用工具。我在测试时发现明确写上“不要替用户想象任务结果”之后Codex 调用工具的频率明显提升。4. 配额优化实战把每一格额度用在刀刃上4.1 缓存先行大部分重复调用可以直接消除配额优化里投入产出比最高的一步就是缓存。我本地跑了一周数据相同或相似任务在连续交互中的重复率高达三到五成。比如“总结昨天邮件”这类任务一个上午可能被触发多次每次结果都一样缓存命中就能省下这些配额。内存缓存适合单机单进程但如果你和我一样需要多副本部署建议用 Redis 做共享缓存import redis r redis.Redis.from_url(os.getenv(REDIS_URL, redis://localhost:6379/0)) val r.get(key) if val: return json.loads(val) r.setex(key, CACHE_TTL, json.dumps(result))缓存 TTL 也是个讲究点。通用任务我设 300 秒周期任务设 1800 秒跟任务本身的更新频率匹配起来。太短的 TTL 缓存命中率低太长的 TTL 容易给用户返回过期结果。4.2 队列和限流把毛刺请求抹平即使有缓存真实调用仍然是免不了的。Agent 喜欢并发抛任务瞬间请求量可能是平均量的四五倍直接冲击 RPM 配额。我在代理服务里实现了限流器RPM 设为配额的 80%。比如配额是 60 RPM代理服务就按 48 RPM 放行剩下的 20% 留给突发情况。实际体验是调用量大的时候后到的请求会在队列里多等几秒但不会出现连续 429 导致整条链路失败的情况。限流的另一层价值是保护 Agent 侧的体验。Claude Code 在等待工具返回时是有进度反馈的多等两秒完全无感但如果收到一串 429 错误它就可能开始自行“发挥”比如反复重试同一个调用反而把配额烧得更快。4.3 多账号资源池合规前提下的扩容思路单账号配额不够用最直接的方案是申请提高配额或升级合同这是最稳妥的正道。如果你们团队本身就有多个合法项目或测试环境多账号资源池也是一种可行的扩容方案。实现上就是前面代码里的key_poolANTIGRAVITY_API_KEYS环境变量里传入多个 Key用itertools.cycle轮流分配。但这里必须强调合规前提每个账号对应一个真实、合法的项目所有调用都符合服务条款不要用多账号来规避限制或做违规用途。我在团队内部是按项目隔离 Key 的每个项目一个 Key代理服务只做路由不做跨项目混用。这样既分散了配额压力也方便从日志里追踪每个项目的消耗量。4.4 配额预警与指数退避别等到被限流才反应最后一道防线是预警和退避。Antigravity 的响应头里通常会带回剩余配额信息代理服务可以读取并在低于阈值时记录日志、发送告警。一旦真的遇到 429指数退避是保命手段。实际代码里我这样处理def should_retry(status: int, retry_count: int) - float: if status 429 and retry_count 5: return min(2 ** retry_count, 60) random.uniform(0, 0.5) return 0初始退避 1 秒每次翻倍封顶 60 秒再加 0.5 秒以内的随机抖动避免多个请求同时重试造成“惊群”。这里的关键是重试次数必须封顶否则 Agent 会无限等待用户体验会变得很差。5. 常见问题排查实录错误码与现场处置5.1 我在实际部署中踩过的坑部署这套系统不是一帆风顺的有几个问题值得单独拎出来讲。第一个问题是 MCP Server 的工具返回格式。FastMCP 的工具函数如果返回 dict在部分 Claude Code 版本里会导致解析失败。有一次我排查了很久最后发现整个调用链路都是通的就是返回格式不匹配。改成字符串返回后问题彻底消失。第二个问题是 Codex 不调用 Skill。最初我把AGENTS.md里的描述写得太委婉比如“可以尝试使用”结果 Codex 十次里有八次选择自己去写代码实现。后来改成强指令“必须调用不要替用户想象结果”调用频率立刻上来。第三个问题是限流参数设得太紧。一开始我把 RPM 设成配额的 90%结果并发毛刺一来还是触发 429。后来调到 80%情况明显好转。这个经验说明预留缓冲比极限压榨更重要。5.2 问题速查表错误码或现象常见原因处置方式401 UnauthorizedAPI Key 未加载、Key 被轮换检查环境变量从控制台复制新 Key403 Forbidden项目未开通 Antigravity 服务到控制台启用 API等待几分钟生效429 Too Many RequestsRPM 或每日额度耗尽确认限流器生效开启指数退避检查缓存命中率500 Internal Server Error平台侧服务异常指数重试临时用缓存兜底MCP connection failedtransport 协议不匹配统一用 stdio检查python -m是否能执行Codex 不调用 SkillAGENTS.md 描述不够明确在描述中加强制语气明确“必须调用”中文输出乱码终端编码问题设置PYTHONUTF81确保返回体为 UTF-8这里的重点是错误处理一定要在代理服务侧做而不是丢给 Agent 做。Agent 面对一堆英文错误码时容易做出奇怪的行为而代理服务可以把错误信息收敛成简短中文提示Agent 只需要知道“调用失败原因已记录”然后继续做它擅长的代码工作。6. 最后分享几点实操体会最后聊几句非技术层面的心得。做这个封装项目我最大的体会是难度不在写代码而在“接口约定”。你和 Agent 之间的工具描述、返回格式、错误处理约定决定了这个 Skill 好不好用。描述写得清楚Agent 调用就精准描述含混Agent 就会乱猜甚至瞎编结果。另外缓存、限流、退避这三件事缺一不可。只加缓存不加限流瞬时毛刺照样打爆配额只加限流不加退避429 之后的重试风暴会让你死得更难看。我建议不要一开始就追求完美架构先让代理服务跑通再逐步加缓存和队列。我第一次部署时差点想直接上服务编排忍住了后来证明先单机跑、再按需扩展是对的。这个封装方案后续还可以继续扩展比如把 Antigravity 出的结构化结果自动回填到项目文档里或者结合消息队列做异步任务分发这些都是顺理成章的下一步。
如果你最近在折腾 Codex 或 Claude Code应该已经感觉到模型能力不是瓶颈配额才是。我这边的情况更具体——团队里的自动化任务经常要调 Antigravity 去跑 Agent 工作流而 Antigravity 的平台配额又会和 Codex / Claude Code 的调用挤在一起两边互相抢。第一个版本我用官方 API 直连结果半天就被 429 打崩后来我换了个思路不直连而是在中间加一层可复用的封装把 Antigravity 包装成 Codex 和 Claude Code 都能调用的外部 Skill。这篇文章就是这套方案的完整复盘从架构选择到代码实现、再到配额优化策略全部来自实际部署经验。先把话说在前面标题里的“突破配额限制”不是去破解服务商的限流而是通过架构手段把有限的配额用得更聪明——干掉重复调用、抹平请求毛刺、在合法前提下做资源调度。这是工程问题不是薅羊毛问题。下面进入正题。1. 先把事情说透Antigravity 与编程 Agent 的配额困局1.1 Antigravity 平台到底能做什么Antigravity 是谷歌云生态里的一个 AI Agent 开发平台主打“用自然语言描述任务系统自动编排工作流并执行”。你可以创建 Agent 处理邮件、整理文档、搜集资料、跑数据管道也可以把它嵌入到自己的系统里。对开发者来说最值钱的是它把任务拆解、工具调用、执行结果聚合这些脏活都包了你只需要定义目标和输入参数。我实际用它跑过两类事情一类是周期性的信息汇总比如每天把团队邮件里的待办事项抽出来整理成结构化清单另一类是跨系统的动作类任务比如根据表格数据生成周报并触发后续流程。这两类任务如果让 Codex 或 Claude Code 自己写代码来做每次都要重新组织逻辑而 Antigravity 的优势在于工作流是平台托管的状态和结果都不需要我维护。1.2 配额限制的真实面貌Antigravity 的配额限制通常不是单一维度而是从每分钟请求数RPM、每日请求数DPD、并发数、积分余额等多个角度同时卡你。我实测下来最难受的是 RPM 太低——如果你让 Codex 一口气并行调用多个 Agent 任务很容易在几秒内就把每分钟配额打满然后整条流水线开始连续报 429。更麻烦的是配额限制往往对“创建 Agent / 工作流”和“执行工作流”分别计数两者的消耗比例不一样。创建类请求虽然频率低但单次消耗高执行类请求频率高容易被 RPM 卡住。这种不对称性导致你在设计封装层时必须区别对待不能用一个限流策略覆盖所有操作。1.3 为什么要在 Codex / Claude Code 里外接一个 SkillCodex 和 Claude Code 默认不具备 Antigravity 的能力。你可以让它们写代码去调 Antigravity API但每次都要处理鉴权、重试、错误解析代码会变得很难维护而且每个项目都要复制一遍。封装成外部 Skill 之后Agent 只负责在合适的时机“呼叫”这个 Skill剩下的鉴权、限流、缓存、错误处理全部由封装层承担。编程 Agent 的注意力应该放在代码逻辑上而不是放在 API 调用细节上。这也是我最终选择“中间代理服务 MCP / Skill 注册”而不是“项目内直接调 SDK”的根本原因。2. 整体方案设计为什么是“代理服务 MCP / Skill”双通道2.1 三种封装方式横评我在动手之前把可能的封装方式列了一遍权衡之后才定的方案。这里直接分享对比结论。方案Claude Code 适配性Codex 适配性统一限流与缓存维护成本官方 SDK / 项目内直连需要写胶水代码需要写胶水代码不支持低自建 HTTP 代理服务通过 MCP 接入通过 CLI / Webhook 接入支持中自建 MCP Server原生支持不支持需额外适配支持中最终我选了“自建 HTTP 代理服务”这条路线因为它的适配面最广Claude Code 通过 MCP 协议接入Codex 通过 CLI 包装后的工具声明接入两边共用同一个代理核心限流、缓存、配额调度都在一个地方维护。2.2 架构总览一次调用是怎么走完的整个链路的调用顺序是这样的Codex 或 Claude Code 判断当前任务需要调用 AntigravityAgent 调用本地注册的 Skill / MCP 工具发起 HTTP 请求到代理服务代理服务先查缓存命中就直接返回完全消耗 Antigravity 配额缓存未命中进入限流队列限流器按 RPM 的 80% 放行代理服务携带 API Key 调用 Antigravity 平台结果返回后代理服务把数据写入缓存再把结果交回给 Agent。我在实际部署中发现如果不加缓存第 3 步永远拿不到命中数据配额消耗会直线上升加了缓存之后很多重复性任务的调用直接消失效果非常明显。2.3 关键决策自建代理而不是直连官方 SDK直连官方 SDK 看起来更简单但有两个问题绕不开。第一Agent 进程里必须保存 API Key安全性差尤其多人共用开发机的时候第二每个项目都要写一套限流和重试根本维护不过来。自建代理的核心优势在于“策略集中化”。限流参数、缓存规则、配额调度全在一个服务里调整Agent 侧只需要知道“调用这个 Skill 能拿到结果”。我选择用 FastAPI 来实现代理服务因为它原生支持异步而且 pydantic 可以直接校验请求参数省掉一堆手写解析代码。建议代理服务不要做得太重。能用环境变量解决的不要上数据库能单机跑的不要上服务编排先把链路跑通再考虑扩展。3. 动手实战从零搭建 Antigravity 外部调用 Skill3.1 第一步准备凭证与初始化工程先在 Antigravity 控制台创建项目拿到 API Key并记下你的 API 端点地址。不同的区域和项目版本端点格式不一样以控制台里的实际地址为准别抄网上的固定 URL。工程初始化我用 Python 虚拟环境依赖只要四个fastapi、uvicorn、httpx、pydantic。MCP Server 部分额外装mcp。mkdir antigravity-bridge cd antigravity-bridge python -m venv .venv source .venv/bin/activate pip install fastapi uvicorn httpx pydantic mcp然后在项目根目录创建.env文件ANTIGRAVITY_BASE_URLhttps://your-antigravity-endpoint ANTIGRAVITY_API_KEYyour_api_key_here RPM_LIMIT60 CACHE_TTL300 PORT8010RPM_LIMIT的值我建议设成你实际配额额度的 80%留出缓冲给非预期请求避免代理服务本身变成新的限流点。3.2 第二步搭一个本地代理服务代理服务的核心逻辑是三层缓存检查、限流、转发调用。下面是我实际在用的核心代码做了简化但保留了完整流程。# app.py import asyncio import hashlib import itertools import json import logging import os import time from contextlib import asynccontextmanager import httpx from fastapi import FastAPI, HTTPException from pydantic import BaseModel logging.basicConfig(levellogging.INFO) logger logging.getLogger(antigravity-bridge) ANTIGRAVITY_BASE os.getenv(ANTIGRAVITY_BASE_URL, ) ANTIGRAVITY_API_KEY os.getenv(ANTIGRAVITY_API_KEY, ) CACHE_TTL int(os.getenv(CACHE_TTL, 300)) RPM_LIMIT int(os.getenv(RPM_LIMIT, 60)) PORT int(os.getenv(PORT, 8010)) cache {} key_pool itertools.cycle( [k.strip() for k in os.getenv(ANTIGRAVITY_API_KEYS, ANTIGRAVITY_API_KEY).split(,) if k.strip()] ) app FastAPI(titleantigravity-bridge) class RunRequest(BaseModel): workflow: str params: dict {} class RateLimiter: def __init__(self, rate: float): self.min_interval 1.0 / rate self._last 0.0 self._lock asyncio.Lock() async def acquire(self): async with self._lock: now time.monotonic() wait self.min_interval - (now - self._last) if wait 0: await asyncio.sleep(wait) now time.monotonic() self._last now limiter RateLimiter(RPM_LIMIT / 60) async def call_antigravity(workflow: str, params: dict): api_key next(key_pool) async with httpx.AsyncClient(timeout120) as client: resp await client.post( f{ANTIGRAVITY_BASE}/v1/workflows/{workflow}:run, headers{Authorization: fBearer {api_key}}, jsonparams, ) if resp.status_code 429: logger.warning(rate limited, body%s, resp.text[:200]) resp.raise_for_status() return resp.json() def cache_key(workflow: str, params: dict) - str: raw f{workflow}|{json.dumps(params, sort_keysTrue)} return hashlib.sha256(raw.encode()).hexdigest() app.post(/tools/antigravity/run) async def run_tool(req: RunRequest): key cache_key(req.workflow, req.params) if key in cache and time.time() - cache[key][ts] CACHE_TTL: logger.info(cache hit: %s, key) return {cache: True, workflow: req.workflow, result: cache[key][data]} async with limiter: try: result await call_antigravity(req.workflow, req.params) except Exception as exc: raise HTTPException(status_code502, detailstr(exc)) from exc cache[key] {data: result, ts: time.time()} return {cache: False, workflow: req.workflow, result: result} if __name__ __main__: import uvicorn uvicorn.run(app, host127.0.0.1, portPORT)这段代码有几个细节值得说明。key_pool用itertools.cycle轮流取 key后面配额优化会用到RateLimiter用asyncio.Lock保证并发安全防止代理服务本身被打穿缓存的 key 使用可排序的 JSON 序列化避免参数顺序不同导致缓存失效。3.3 第三步把服务注册为 Claude Code 的 MCPClaude Code 对 MCP 的支持比较成熟我们可以把上面的 HTTP 服务封装成一个 MCP Server让 Claude Code 直接认识它。关键是写一个 FastMCP 入口文件把 HTTP 调用包成工具函数。# antigravity_mcp_server.py import httpx from mcp.server.fastmcp import FastMCP mcp FastMCP(antigravity) BRIDGE_URL http://127.0.0.1:8010 mcp.tool() def run_workflow(workflow: str, params: dict {}) - str: 运行 Antigravity workflow返回执行结果 JSON。 resp httpx.post( f{BRIDGE_URL}/tools/antigravity/run, json{workflow: workflow, params: params}, timeout180, ) resp.raise_for_status() return resp.text if __name__ __main__: mcp.run()这里有个易错点MCP 工具函数的返回类型必须是字符串不能直接返回 dict否则 Claude 可能无法解析。我在开发时踩过一次坑Claude 一直提示“工具返回格式无效”改成resp.text字符串返回后问题立刻消失。注册方式有两种。一种是用命令行直接注册claude mcp add antigravity --command python --args -m antigravity_mcp_server另一种是在项目根目录写.mcp.json这样项目所有成员共用配置{ mcpServers: { antigravity: { command: python, args: [-m, antigravity_mcp_server], env: { BRIDGE_URL: http://127.0.0.1:8010 } } } }注册完成后在 Claude Code 里输入claude mcp list就能看到antigravity已连接。我在 mac 上实测用 stdio 传输最稳HTTP transport 偶尔会因端口占用导致重连失败。3.4 第四步把服务注册为 Codex 的 SkillCodex 的 Skill 机制更偏向“声明式工具定义”。做法很简单在项目里放一个可执行的 CLI 脚本然后在AGENTS.md里声明这个工具的存在和用法。先写一个最简 CLI 封装脚本#!/usr/bin/env bash # antigravity-cli set -euo pipefail TASK${1:-} curl -s -X POST ${ANTIGRAVITY_BRIDGE_URL:-http://127.0.0.1:8010}/tools/antigravity/run \ -H Content-Type: application/json \ -d {\workflow\:\$TASK\,\params\:{}}给它加执行权限chmod x antigravity-cli然后在AGENTS.md里声明## Tools ### antigravity-cli 当任务描述为具体的外部 Agent 执行任务如整理邮件、汇总文档、运行工作流时必须调用 antigravity-cli task_description 注意不要替用户想象任务结果必须真实调用后返回。同一任务不要短时间重复调用。关键点是“必须调用”这几个字。Codex 本质上是按概率选择工具的 Agent如果描述写得模棱两可它很可能选择自己动手写代码而不是调用工具。我在测试时发现明确写上“不要替用户想象任务结果”之后Codex 调用工具的频率明显提升。4. 配额优化实战把每一格额度用在刀刃上4.1 缓存先行大部分重复调用可以直接消除配额优化里投入产出比最高的一步就是缓存。我本地跑了一周数据相同或相似任务在连续交互中的重复率高达三到五成。比如“总结昨天邮件”这类任务一个上午可能被触发多次每次结果都一样缓存命中就能省下这些配额。内存缓存适合单机单进程但如果你和我一样需要多副本部署建议用 Redis 做共享缓存import redis r redis.Redis.from_url(os.getenv(REDIS_URL, redis://localhost:6379/0)) val r.get(key) if val: return json.loads(val) r.setex(key, CACHE_TTL, json.dumps(result))缓存 TTL 也是个讲究点。通用任务我设 300 秒周期任务设 1800 秒跟任务本身的更新频率匹配起来。太短的 TTL 缓存命中率低太长的 TTL 容易给用户返回过期结果。4.2 队列和限流把毛刺请求抹平即使有缓存真实调用仍然是免不了的。Agent 喜欢并发抛任务瞬间请求量可能是平均量的四五倍直接冲击 RPM 配额。我在代理服务里实现了限流器RPM 设为配额的 80%。比如配额是 60 RPM代理服务就按 48 RPM 放行剩下的 20% 留给突发情况。实际体验是调用量大的时候后到的请求会在队列里多等几秒但不会出现连续 429 导致整条链路失败的情况。限流的另一层价值是保护 Agent 侧的体验。Claude Code 在等待工具返回时是有进度反馈的多等两秒完全无感但如果收到一串 429 错误它就可能开始自行“发挥”比如反复重试同一个调用反而把配额烧得更快。4.3 多账号资源池合规前提下的扩容思路单账号配额不够用最直接的方案是申请提高配额或升级合同这是最稳妥的正道。如果你们团队本身就有多个合法项目或测试环境多账号资源池也是一种可行的扩容方案。实现上就是前面代码里的key_poolANTIGRAVITY_API_KEYS环境变量里传入多个 Key用itertools.cycle轮流分配。但这里必须强调合规前提每个账号对应一个真实、合法的项目所有调用都符合服务条款不要用多账号来规避限制或做违规用途。我在团队内部是按项目隔离 Key 的每个项目一个 Key代理服务只做路由不做跨项目混用。这样既分散了配额压力也方便从日志里追踪每个项目的消耗量。4.4 配额预警与指数退避别等到被限流才反应最后一道防线是预警和退避。Antigravity 的响应头里通常会带回剩余配额信息代理服务可以读取并在低于阈值时记录日志、发送告警。一旦真的遇到 429指数退避是保命手段。实际代码里我这样处理def should_retry(status: int, retry_count: int) - float: if status 429 and retry_count 5: return min(2 ** retry_count, 60) random.uniform(0, 0.5) return 0初始退避 1 秒每次翻倍封顶 60 秒再加 0.5 秒以内的随机抖动避免多个请求同时重试造成“惊群”。这里的关键是重试次数必须封顶否则 Agent 会无限等待用户体验会变得很差。5. 常见问题排查实录错误码与现场处置5.1 我在实际部署中踩过的坑部署这套系统不是一帆风顺的有几个问题值得单独拎出来讲。第一个问题是 MCP Server 的工具返回格式。FastMCP 的工具函数如果返回 dict在部分 Claude Code 版本里会导致解析失败。有一次我排查了很久最后发现整个调用链路都是通的就是返回格式不匹配。改成字符串返回后问题彻底消失。第二个问题是 Codex 不调用 Skill。最初我把AGENTS.md里的描述写得太委婉比如“可以尝试使用”结果 Codex 十次里有八次选择自己去写代码实现。后来改成强指令“必须调用不要替用户想象结果”调用频率立刻上来。第三个问题是限流参数设得太紧。一开始我把 RPM 设成配额的 90%结果并发毛刺一来还是触发 429。后来调到 80%情况明显好转。这个经验说明预留缓冲比极限压榨更重要。5.2 问题速查表错误码或现象常见原因处置方式401 UnauthorizedAPI Key 未加载、Key 被轮换检查环境变量从控制台复制新 Key403 Forbidden项目未开通 Antigravity 服务到控制台启用 API等待几分钟生效429 Too Many RequestsRPM 或每日额度耗尽确认限流器生效开启指数退避检查缓存命中率500 Internal Server Error平台侧服务异常指数重试临时用缓存兜底MCP connection failedtransport 协议不匹配统一用 stdio检查python -m是否能执行Codex 不调用 SkillAGENTS.md 描述不够明确在描述中加强制语气明确“必须调用”中文输出乱码终端编码问题设置PYTHONUTF81确保返回体为 UTF-8这里的重点是错误处理一定要在代理服务侧做而不是丢给 Agent 做。Agent 面对一堆英文错误码时容易做出奇怪的行为而代理服务可以把错误信息收敛成简短中文提示Agent 只需要知道“调用失败原因已记录”然后继续做它擅长的代码工作。6. 最后分享几点实操体会最后聊几句非技术层面的心得。做这个封装项目我最大的体会是难度不在写代码而在“接口约定”。你和 Agent 之间的工具描述、返回格式、错误处理约定决定了这个 Skill 好不好用。描述写得清楚Agent 调用就精准描述含混Agent 就会乱猜甚至瞎编结果。另外缓存、限流、退避这三件事缺一不可。只加缓存不加限流瞬时毛刺照样打爆配额只加限流不加退避429 之后的重试风暴会让你死得更难看。我建议不要一开始就追求完美架构先让代理服务跑通再逐步加缓存和队列。我第一次部署时差点想直接上服务编排忍住了后来证明先单机跑、再按需扩展是对的。这个封装方案后续还可以继续扩展比如把 Antigravity 出的结构化结果自动回填到项目文档里或者结合消息队列做异步任务分发这些都是顺理成章的下一步。
平方复杂度无法被常规优化消除｜TaoToken 统一 Key 通道实测)
