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

1. 选凭证（三种主体，按调用方身份选）

from tfrs_auth import (
    CachingTokenSource, AsyncCachingTokenSource,
    ClientCredentials, PatCredential, UserJwtCredential,
)

TOKEN_URL = "https://<user-host>/api/v1/oauth/token"
AUDIENCE = "robot:<callee_public_id>"   # 被调机器人 public_id（受众隔离）

# ① client_credentials —— 机器身份（机器人以自身机器凭证换发，A2A 主叫端）
cred = ClientCredentials.for_robot(
    client_id="turingfocus:000101",           # 机器人自身 public_id
    client_secret=machine_client_secret,      # tfp_ 明文（machineClientSecret）
    callee_org_slug="turingfocus",            # → audience=robot:turingfocus:000042
    callee_employee_no="000042",
    scope=["a2a:invoke"],                     # 省略则不带 scope
)
#   或直构：ClientCredentials(client_id=..., client_secret=..., audience=AUDIENCE)

# ② PAT —— 用户 PAT 主体（用户自管理 PAT，tfp_ 不透明令牌）
cred = PatCredential(pat="tfp_...", audience=AUDIENCE, scope="config:read")

# ③ User JWT —— 用户登录态 JWT 主体（session JWT，来自登录）
cred = UserJwtCredential(user_jwt=user_jwt_str, audience=AUDIENCE)

三种凭证喂给同一个 token source（凭证无关）。source 长期持有、复用（它本身就是缓存）； .token() 幂等：未临期直返、临期自动换发、并发只放一个换发在途（single-flight）+ 429/503 退避。 同步用 CachingTokenSource、异步用 AsyncCachingTokenSource（API 对称）。

2. 发请求 —— 自动携带合法 Token

方式 A：httpx + BearerAuth（推荐） —— 自动注入 Authorization: Bearer，遇 401 自动刷新重试一次。

import httpx
from tfrs_auth.transport import BearerAuth   # 需 tfrs-auth[httpx]

with CachingTokenSource(cred, token_url=TOKEN_URL) as source:
    with httpx.Client(auth=BearerAuth(source)) as client:
        resp = client.get("https://<api-host>/data")   # 自动带 Bearer <jwt>

# 异步（用法对称）：
# async with AsyncCachingTokenSource(cred, token_url=TOKEN_URL) as source:
#     async with httpx.AsyncClient(auth=AsyncBearerAuth(source)) as client:
#         resp = await client.get("https://<api-host>/data")

方式 B：自带 httpx / requests / 任意库（手动取 token 注入 header）

tf-core 尚未开发；若你用自己的 HTTP 库，手动从 source 取 token 字符串塞进请求头：

import requests

with CachingTokenSource(cred, token_url=TOKEN_URL) as source:
    token = source.token()   # Token(.access_token, .token_type="Bearer", .expires_at, ...)
    resp = requests.get(
        "https://<api-host>/data",
        headers={"Authorization": f"{token.token_type} {token.access_token}"},
    )
    # 遇 401 时手动刷新重试一次（等价 BearerAuth 的内部行为）：
    # source.invalidate()           # 丢缓存
    # token = source.token()        # 强制重取新 token 后重试

手动方式不自动处理 401：source 仍管缓存 / 临期刷新，但 401 后的「丢缓存重取」需你显式调 source.invalidate() 再 .token()。要复用既有 httpx client 传 http_client=...；source 只关闭 自己创建的 client。BearerAuth / AsyncBearerAuth 是 opt-in 模块（from tfrs_auth.transport import， 需 tfrs-auth[httpx]；不进包根 __all__，故 import tfrs_auth 仍可在无 httpx 时使用）。

被叫方 —— 验签收到的 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_public_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:turingfocus:000123")。

⚠️ 资源服务器务必传 issuer 与 audience：二者仅在提供时才校验。不传 audience 会放行签给任意机器人的 token，丧失 aud=robot:{public_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）。