给你的A2A Agent加把锁认证鉴权实战指南摘要A2A Agent暴露到网络上等于裸奔。手把手实现API Key、JWT Bearer、OAuth 2.0三种认证方式从开发到生产的完整安全方案。一、上篇的Agent有个大问题上篇你搭了翻译Agent和摘要Agent两个Agent在本地跑得好好的。但有个问题你可能没注意到任何知道你端口号的人都能调用你的Agent。没有认证、没有鉴权、没有审计。你的Agent在10002和10003端口上裸奔谁都能发个JSON-RPC请求过来让你的Agent干活。在本地demo里这不是问题。但一旦你把Agent部署到服务器上——特别是公网环境——这就是一个安全漏洞。别人可以恶意调用你的Agent消耗算力、窃取数据、或者篡改任务结果。A2A协议原生支持认证鉴权。它的做法很聪明把认证信息放在Agent Card里声明客户端看到声明后就知道该怎么认证。今天带你实现三种认证方式从简到繁。二、A2A认证的底层逻辑在写代码之前先搞清楚A2A认证是怎么工作的。两个关键原则原则1认证在HTTP层不在JSON-RPC里。A2A的JSON-RPC payload里不包含任何认证信息。Token、API Key这些东西通过HTTP Header传递比如Authorization: Bearer xxx或X-API-Key: xxx。原则2Agent Card声明认证需求。Agent在自己的Card里通过securitySchemes字段告诉客户端“我需要什么类型的认证、凭证放在哪个Header里”。客户端读取Card后就知道该怎么准备凭证。三级认证方案|级别|方式|适用场景|安全等级|| — | — | — | — ||Level 1|API Key|内部Agent、原型开发|低||Level 2|JWT Bearer|多团队、有身份提供商|中||Level 3|OAuth 2.0|跨组织、合规要求|高|三级递进不是互斥的。你可以从Level 1开始需要时升级到Level 2或3不需要重写Agent。三、Level 1API Key认证5分钟搞定3.1 Agent Card声明在Agent Card里加上securitySchemes字段agent_cardAgentCard(nameProtected Agent,descriptionAn agent with API Key authentication,urlfhttp://{host}:{port}/,version0.1.0,defaultInputModes[text],defaultOutputModes[text],capabilitiesAgentCapabilities(streamingTrue),skills[skill],securitySchemes{apiKey:{type:apiKey,in:header,name:X-API-Key,}},security[{apiKey:[]}],)这段声明告诉客户端我要你在请求头里带一个X-API-Key字段。3.2 写一个认证中间件用Starlette中间件拦截每个请求校验API Keyimportosimporthashlib from starlette.requestsimportRequest from starlette.responsesimportJSONResponse from starlette.middleware.baseimportBaseHTTPMiddleware classAPIKeyMiddleware(BaseHTTPMiddleware):API Key认证中间件。# 存储API Key的SHA-256哈希别存明文VALID_KEYS{hashlib.sha256(os.environ. get(AGENT_API_KEY,dev-key-123).encode()).hexdigest():trusted-client,}asyncdefdispatch(self, request: Request, call_next):# Agent Card始终公开不需要认证ifrequest.url.path/.well-known/ agent.json:returnawaitcall_next(request)# 检查X-API-Key头api_keyrequest.headers. get(X-API-Key)ifnot api_key: returnJSONResponse(status_code401,content{jsonrpc:2.0,error:{code:-32001,message:Missing X-API-Key header,},},)# 校验Keykey_hashhashlib.sha256(api_key.encode()).hexdigest()client_idself.VALID_KEYS.get(key_hash)ifnot client_id: returnJSONResponse(status_code401,content{jsonrpc:2.0,error:{code:-32001,message:Invalid API key,},},)# 把client_id存到request.state里后面可以用 request.state.client_idclient_id returnawaitcall_next(request)关键细节Agent Card的端点/.well-known/ [agent.json](http://agent.json)必须跳过认证。否则客户端连你的Agent Card都拿不到怎么知道需要什么认证3.3 把中间件挂到Server上from starlette.applicationsimportStarlette from starlette.routingimportRoute# 创建Starlette应用挂载中间件appStarlette(routes[Route(/.well-known/ agent.json, agent_card_handler,), Route(/, a2a_handler,methods[POST]),],)app.add_middleware(APIKeyMiddleware)# 用uvicorn跑Starlette应用importuvicorn uvicorn.run(app,hosthost,portport)3.4 客户端带Key调用asyncwith httpx.AsyncClient()as client: respawait client.post(http://localhost:10002/,jsonpayload,headers{X-API-Key:dev-key-123},)一个Header就搞定。3.5 API Key适合什么场景内部服务、原型开发、自己控制的Agent之间调用。它的优点是简单缺点也很明显所有客户端共享同一个Key无法区分谁调的Key泄漏了没法追溯没有权限粒度——有Key就能干所有事 如果你有这些需求升级到Level 2。四、Level 2JWT Bearer认证多团队协作4.1 Agent Card声明securitySchemes{bearer:{type:http,scheme:bearer,bearerFormat:JWT,}},security[{bearer:[]}],4.2 JWT验证中间件JWT的好处Token本身携带身份信息和权限claimsServer不需要查数据库就能验证。importjwt from jwtimportPyJWKClient from starlette.requestsimportRequest from starlette.responsesimportJSONResponse from starlette.middleware.baseimportBaseHTTPMiddleware classJWTMiddleware(BaseHTTPMiddleware): def__init__(self, app, issuer: str, jwks_url: str, audience: str): super().__init__(app)self.issuerissuer self.audienceaudience# JWKS客户端用于获取签名公钥self.jwks_clientPyJWKClient(jwks_url,cache_keysTrue)asyncdefdispatch(self, request: Request, call_next):# Agent Card公开ifrequest.url.path/.well-known/ agent.json:returnawaitcall_next(request)# 提取Bearer Tokenauth_headerrequest.headers. get(Authorization,)ifnot auth_header.startswith(Bearer ): returnJSONResponse(status_code401,content{jsonrpc:2.0,error:{code:-32001,message:Missing Bearer token,},},)tokenauth_header[7:]# 去掉Bearer try:# 获取签名公钥signing_keyself.jwks_client.get_signing_key_from_jwt(token)# 验证Tokenclaimsjwt.decode(token, signing_key.key,algorithms[RS256,ES256],audienceself.audience,issuerself.issuer,options{require:[exp,iss,aud,sub]},)except jwt.ExpiredSignatureError: returnJSONResponse(status_code401,content{jsonrpc:2.0,error:{code:-32001,message:Token expired,},},)except jwt.InvalidTokenError as e: returnJSONResponse(status_code401,content{jsonrpc:2.0,error:{code:-32001,message:fInvalid token: {e},},},)# 把claims存到request里request.state.claimsclaims request.state.client_idclaims.get(sub,unknown)returnawaitcall_next(request)4.3 权限检查ScopeJWT可以携带权限范围scope。比如你的Agent有多个技能你可以规定只有带agent:executescope的Token才能执行任务defrequire_scope(required_scope: str):检查JWT中是否包含指定scope。 defchecker(request: Request): claimsgetattr(request.state,claims,{})scopesclaims.get(scope,).split()ifrequired_scope notin scopes: returnJSONResponse(status_code403,content{jsonrpc:2.0,error:{code:-32003,message:fMissing scope: {required_scope},},},)returnNonereturnchecker# 在请求处理中使用asyncdefhandle_task(request: Request): errorrequire_scope(agent:execute)(request)iferror: returnerror# 正常处理任务...4.4 客户端带Token调用importhttpx asyncdefcall_agent(agent_url: str, text: str, token: str): payload{jsonrpc:2.0,id:1,method:message/send,params:{message:{role:user,parts:[{type:text,text:text}],}},}asyncwith httpx.AsyncClient()as client: respawait client.post(agent_url,jsonpayload,headers{Authorization:fBearer {token}},timeout60,)resp.raise_for_status()returnresp.json()五、Level 3OAuth 2.0 Client Credentials生产级方案当你有多个Agent跨组织协作时需要的是完整的OAuth 2.0流程。5.1 Agent Card声明securitySchemes{oauth2:{type:oauth2,flows:{clientCredentials:{tokenUrl:https:// auth.example.com/oauth2/token,scopes:{agent:read:读取任务状态和元数据,agent:execute:提交和管理任务,agent:admin:配置Agent设置,}}}}},security[{oauth2:[agent:execute]}],注意security字段里的[agent:execute]——这意味着默认需要agent:executescope才能调用这个Agent。5.2 能力级别细粒度控制如果你某个技能需要更高的权限可以在Skill级别覆盖安全声明skills[AgentSkill(idpublic-skill,nameHello,descriptionSays hello,tags[greeting],), AgentSkill(idsensitive-skill,nameData Analysis,descriptionAnalyzes sensitive data,tags[analysis,pii],security[{oauth2:[agent:execute,data:pii]}],),]data:piiskill需要额外的权限才能调用。5.3 客户端自动获取Token生产环境里Token有有效期一般15分钟客户端需要自动刷新importtimeimporthttpx class OAuth2A2AClient:自动管理OAuth2.0Token的A2A客户端。 def__init__(self, token_url: str, client_id: str, client_secret: str, default_scopes: list[str]|NoneNone,): self.token_urltoken_url self.client_idclient_id self.client_secretclient_secret self.default_scopesdefault_scopes or[agent:execute]self._token: str|NoneNone self._token_expiry: float0asyncdef_fetch_token(self, scopes: list[str])-str:向授权服务器请求Token。 asyncwith httpx.AsyncClient()as http: respawait http.post(self.token_url,data{grant_type:client_credentials,client_id:self.client_id,client_secret:self.client_secret,scope: .join(scopes),},)resp.raise_for_status()dataresp.json()self._tokendata[access_token]# 提前60秒刷新self._token_expiry(time.time() data.get(expires_in,3600)-60)returnself._token asyncdefget_token(self, scopes: list[str]|NoneNone)-str:获取有效Token过期自动刷新。 scopesscopes orself.default_scopes ifself._token and time.time()self._token_expiry: returnself._token returnawaitself._fetch_token(scopes)asyncdefdiscover_and_call(self, base_url: str, text: str)-dict:完整流程发现Agent→获取Token→调用。 asyncwith httpx.AsyncClient()as http:# 1. 获取Agent Cardcard_respawait http.get(f{base_url}/.well-known/ agent.json)cardcard_resp.json()# 2. 从Card里提取认证需求oauthcard.get(securitySchemes,{}).get(oauth2,{})flowsoauth.get(flows,{})ccflows.get(clientCredentials,{})self.token_urlcc.get(tokenUrl, self.token_url)# 3. 获取所需scopesecuritycard.get(security,[{}])requiredsecurity[0].get(oauth2, self.default_scopes[:1])# 4. 获取Tokentokenawaitself.get_token(required)# 5. 调用Agentpayload{jsonrpc:2.0,id:1,method:message/send,params:{message:{role:user,parts:[{type:text,text:text}],}},}respawait http.post(card[url],jsonpayload,headers{Authorization:fBearer {token}},timeout60,)resp.raise_for_status()returnresp.json()discover_and_call方法展示了一个完整的自动化流程读Agent Card→提取认证需求→获取Token→调用Agent。这就是一个合格的Agent编排器该做的事。5.4 验证OAuth TokenOAuth 2.0 Client Credentials颁发的Token其实就是JWT。所以Level 2的JWT中间件直接复用不需要额外代码。区别只在于Token的来源Level 2是客户端自己签发的Level 3是授权服务器颁发的。六、三种方案怎么选一个简单的决策树你的Agent只在内部用→ API Key。5分钟搞定够用了。多个团队共享Agent→ JWT Bearer。你有现成的身份提供商比如Auth0、Keycloak直接用。Agent跨组织协作合规要求高→ OAuth 2.0 Client Credentials。有授权服务器、有审计日志、有权限粒度。我的建议先用API Key跑通demo验证业务逻辑没问题了再切换到JWT或OAuth 2.0。别一开始就搞复杂的认证业务逻辑改来改去的时候还要同时调认证代码两头跑。七、安全清单部署前检查部署Agent到生产环境之前过一遍这个清单[ ] 所有通信走HTTPSTLS 1.3[ ] Agent Card不包含明文密钥[ ] API Key存哈希不存明文[ ] Token有效期≤15分钟机器间通信[ ] 日志里记录client_id和scope不记录Token[ ] 定义了最小权限的scopeagent:read / agent:execute / agent:admin[ ] 凭证放在环境变量或密钥管理器里不写进代码[ ] Agent Card端点跳过认证否则客户端发现不了你八、我踩过的3个坑坑1Agent Card端点加了认证。客户端访问/.well-known/ [agent.json](http://agent.json)拿到401直接报错说无法发现Agent。排查了20分钟才发现中间件拦截了这个路径。解决方法中间件里加个条件判断跳过Agent Card路径。坑2API Key存了明文。写demo时图省事把API Key直接写在代码里。后来代码传到Git上被安全扫描工具告警了。解决方法用环境变量并且只存哈希值。坑3OAuth Token过期没处理。第一次写的时候Token过期后客户端直接报错没有自动刷新逻辑。生产环境里Token一般15分钟过期如果不做自动刷新Agent之间的通信会频繁中断。解决方法用上面那个OAuth2A2AClient类在Token过期前自动刷新。下篇预告《MCP管工具A2A管协作双协议联合实战》—— 让Agent既能通过MCP调外部工具又能通过A2A跟其他Agent对话搭一个搜索总结的多Agent工作流。