tfrs-auth (Python)

TFRS AccessToken 客户端工具包：把「PAT / client_credentials → 短 RS256 JWT 换发 （RFC 8693）+ 缓存/刷新」与「Claims/scope 契约 + 纯验签」抽成可复用包，避免每个 MCP 工具 / SDK 重复实现 token 获取与生命周期管理。

下游集成只涉及两个角色：主叫方（换发并携带 token）与被叫方（验签收到的 token）。

SoT = TFRSManager Contract Registry。本包 contract.py 为镜像，变更先改 Manager。 强制执行层（验签中间件/Socket.IO/ESO/默认拒绝 CI）不在本包，留 TFRobotServer。

安装

已发布至 PyPI：https://pypi.org/project/tfrs-auth/（MIT 许可证）。

# 从 PyPI（推荐）
uv add tfrs-auth                  # 核心：契约 + 纯验签（仅依赖 pyjwt[crypto]）
uv add "tfrs-auth[httpx]"         # 含客户端 token source / JWKS live 拉取
# 或：pip install "tfrs-auth[httpx]"

# 或从 git + tag（未发布版本）
uv add "tfrs-auth[httpx] @ git+https://cnb.cool/turingfocus/foundation/tfrs-foundation-py@<tag>#subdirectory=packages/tfrs-auth"

• 核心（契约 + verify.from_jwks 纯验签）：只需 pyjwt[crypto]，无 httpx 也可 import。
• 可选 extra httpx：换发 token（token source）或 JwtVerifier.from_url live 拉取 JWKS 时才需要。

用法

主叫方 —— 换发并携带 token（client_credentials，S5 A2A 主叫端）

from tfrs_auth import AsyncCachingTokenSource, ClientCredentials

cred = ClientCredentials.for_robot(
    client_id=machine_client_id,          # 机器人自身 Account.ID
    client_secret=machine_client_secret,  # tfp_ 明文（machineClientSecret）
    callee_robot_id=callee_id,            # → aud=robot:<calleeId>
    scope=["a2a:invoke"],                 # 省略则不带 scope
)
async with AsyncCachingTokenSource(
    cred, token_url="https://<user-host>/api/v1/oauth/token"
) as source:
    token = await source.token()          # 缓存 + 临期自动刷新 + single-flight + 429/503 退避
    headers = {"Authorization": f"{token.token_type} {token.access_token}"}

• source 长期持有、复用（它本身就是缓存）；不要每个请求新建一个。.token() 幂等： 缓存未临期直接返回，临期了自动换发，并发只放一个换发在途、其余等待后命中缓存。
• 同步场景用 CachingTokenSource（线程锁做 single-flight，API 与 async 版对称）。
• 想复用既有 httpx client：传 http_client=...；token source 只关闭自己创建的 client。

被叫方 —— 验签收到的 token（资源服务器侧）

from tfrs_auth import JwtVerifier

# 线上：从 JWKS URL 构造（拉取 + 按 kid 缓存带 TTL + 未知-kid 限速刷新）
verifier = JwtVerifier.from_url(
    "https://<user-host>/.well-known/jwks.json",
    issuer="https://<user-host>",         # 强烈建议传
    audience=f"robot:{my_robot_id}",      # 强烈建议传：受众隔离
)
claims = verifier.verify(access_token)    # 成功返回 Claims；失败抛 TokenVerificationError
assert claims.has_scope("a2a:invoke")

离线/测试零网络：JwtVerifier.from_jwks(jwks_dict, issuer=..., audience=...)。verifier 同样长期复用（内含 JWKS 缓存）；单次想覆盖受众用 verifier.verify(tok, audience="robot:123")。

⚠️ 资源服务器务必传 issuer 与 audience：二者仅在提供时才校验。不传 audience 会放行签给任意机器人的 token，丧失 aud=robot:<id> 受众隔离（issuer 之于颁发者 隔离同理）。from_url 的未知-kid 刷新已内置限速（min_refresh_interval，默认 10s）， 避免被构造的随机-kid token 放大成对 JWKS 端点的 DoS。

错误处理（带类型，绝不字符串匹配）

from tfrs_auth import (
    TokenExchangeError,      # 换发失败基类；.retryable 标识 429/503
    TokenVerificationError,  # 验签失败（过期 / 签名错 / aud·iss 不符）
    TransportError,          # JWKS / 换发的网络层失败
    TemporarilyUnavailableError, InvalidClientError, InvalidScopeError,  # 细分子类
)

# 主叫方：退避已内置，兜住自动重试后仍失败的情况
try:
    token = await source.token()
except TokenExchangeError as e:
    ...  # e.retryable 区分 429/503 等瞬时失败

# 被叫方：网络失败与验签失败应同时兜住
try:
    claims = verifier.verify(access_token)
except (TokenVerificationError, TransportError):
    ...  # → 401

速查表

模块

边界与状态

• 已实现：contract / errors / client(sync+async caching) / credentials.ClientCredentials / credentials.PatCredential / credentials.UserJwtCredential / verify / transport.BearerAuth（+ AsyncBearerAuth）。
• 无骨架：三层 + transport 全部已实现。transport 为 opt-in（from tfrs_auth.transport import，需 tfrs-auth[httpx]；不进包根 __all__，保 httpx-optional）。