agent（Python）+传统业务系统（Java）下如何保证安全性（代码全链路讲解）｜TaoToken 统一 Key 通道实践

1. 为什么 Python Agent 调 Java 老系统安全链路最容易断在中间很多团队做 Agent 落地时Python 侧写得飞快Java 侧的老业务系统却动不得——它承载着订单、库存、结算这些核心逻辑接口早就稳定运行了好几年。于是常见做法是把 Java 系统包装成 MCP Server 暴露工具Python Agent 作为 MCP Client 去调用。听起来很顺但真正上线前你会发现agentPython传统业务系统Java的安全性问题几乎全压在“中间这段”上。我见过最典型的翻车场景是这样的Agent 的 gateway 进程做了 JWT 鉴权看起来挺安全但 gateway 转发到 agent service 时只带了明文请求头谁都能伪造一个X-User-Id: 1直接打 agent service 的端口agent service 再去调 Java MCP Server 时又只传了个Authorization: Bearer但没校验来源。结果就是——外部用户 A 能通过构造请求让系统以用户 B 的身份去查订单。这不是理论风险是真实会被扫出来的漏洞。所以这篇要讲的是代码全链路从外部请求进 gateway到 gateway 转发 agent service再到 agent service 通过 MCP 协议调 Java 业务系统每一跳都要有身份凭证且凭证必须能防篡改。核心手段就三个JWT 做外部身份、HMAC 签名做内部上下文完整性、内部 JWT 做 MCP 连接鉴权。下面按可复制的顺序拆开讲每一步都给配置和代码。适合谁看正在把 Java 老系统接入 Agent 的后端同学、做 MCP Server 封装的工程师、以及需要给 Agent 加审计和限流的架构同学。你不需要是安全专家但需要能读懂 Python 中间件和 Java 的 Filter/Interceptor。先明确一个前提Agent 拆成 gateway 和 agent service 两个独立进程不是过度设计。gateway 面向公网负责鉴权、限流、审计、防提示注入agent service 面向内网负责跑 LangGraph、管理 MCP 连接。两者之间必须有信任边界否则 gateway 的所有校验都白做。这个边界就是后面 HMAC 签名要解决的问题。2. TaoToken 统一 Key 通道把模型调用凭证收口到一处在讲业务链路安全之前得先解决一个容易被忽略的隐患Agent 里到处散落的模型 API Key。Python 代码里写一个、配置文件里写一个、MCP Server 里再写一个一旦泄露就是全线失守。更麻烦的是Java 老系统如果也要调模型做意图识别Key 管理会更乱。我的做法是用 TaoToken 做统一 Key 通道所有模型调用走同一个入口业务侧只认一个 Key。这样安全边界清晰Key 只在 gateway 或 agent service 的环境变量里出现不落到 Java 业务代码里。TaoToken 的接入地址是https://taotoken.net/api控制台在https://taotoken.net/consoleKey 在https://taotoken.net/api-keys生成。具体配置上Python 侧用 OpenAI 兼容方式接入把base_url指向 TaoToken 即可。这样 LangGraph 里的模型节点、MCP Server 里的辅助模型调用都能复用同一套凭证。下面是一个可复制的环境变量片段放在 gateway 和 agent service 共用的.env里# .env 统一模型通道配置 TAOTOKEN_BASE_URLhttps://taotoken.net/api TAOTOKEN_API_KEYsk-你的统一Key TAOTOKEN_MODEL_IDclaude-sonnet-4-5 # 内部签名密钥gateway 与 agent service 必须一致 INTERNAL_IDENTITY_SECRET请换成32位以上随机串 # MCP 内部 JWT 配置 MCP_JWT_ISSUERagent-service MCP_JWT_AUDIENCEjava-mcp-server MCP_JWT_TTL_SECONDS120注意INTERNAL_IDENTITY_SECRET这个值它是后面 HMAC 签名的根密钥绝对不能提交到 Git也不要用弱口令。生产环境建议从密钥管理服务注入。MCP_JWT_TTL_SECONDS设短一点120 秒足够一次工具调用过期即失效降低重放风险。如果你还在用 Coding Plan 做长期编码或 Agent 开发可以把开发期的模型调用也切到统一通道避免本地调试时用个人 Key 混进生产配置。Coding Plan 入口在https://taotoken.net/coding-plan模型对话调试在https://taotoken.net/chat。这样从开发到上线Key 的来源是一致的审计时也好追溯。这里要强调一点TaoToken 是模型调用的统一通道不是业务系统的鉴权中心。业务身份用户是谁、租户是谁仍然由你自己的 JWT 体系负责两者不要混。模型 Key 管的是“能不能调模型”业务 JWT 管的是“能不能访问这条数据”职责分开出问题时排查方向才清晰。配置好之后Python 侧读取方式统一封装成一个函数避免各处硬编码# config.py import os from functools import lru_cache lru_cache def get_model_config() - dict: return { base_url: os.environ[TAOTOKEN_BASE_URL], api_key: os.environ[TAOTOKEN_API_KEY], model: os.environ.get(TAOTOKEN_MODEL_ID, claude-sonnet-4-5), } lru_cache def get_internal_secret() - str: secret os.environ[INTERNAL_IDENTITY_SECRET] if len(secret) 32: raise RuntimeError(INTERNAL_IDENTITY_SECRET too short) return secret这样 gateway 和 agent service 都从同一份配置读密钥签名校验才能对得上。下一步进入真正的链路外部 JWT 怎么在 gateway 被解析成上下文。3. 可复制配置gateway 中间件与 HMAC 签名转发gateway 是整个链路的入口它的职责是解析外部 JWT、生成身份上下文、做限流和审计、然后把请求带着签名转发给 agent service。这里的关键是签名——agent service 不信任任何没有签名的请求头哪怕它来自内网。先看 gateway 的main.py中间件顺序很重要从外到内依次是限流、审计、租户解析、提示注入防护# gateway/main.py from fastapi import FastAPI from gateway.routers.agent_proxy import router as agent_proxy_router from gateway.middlewares.rate_limit import RateLimitMiddleware from gateway.middlewares.audit_log import AuditLogMiddleware from gateway.middlewares.tenant_resolve import TenantResolveMiddleware from gateway.middlewares.prompt_injection import PromptInjectionMiddleware app FastAPI(titleAgent Gateway) app.include_router(agent_proxy_router) app.add_middleware(PromptInjectionMiddleware) app.add_middleware(RateLimitMiddleware) app.add_middleware(AuditLogMiddleware) app.add_middleware(TenantResolveMiddleware)TenantResolveMiddleware负责解析 JWT 并把身份写进request.state后续中间件和路由都能拿到# gateway/middlewares/tenant_resolve.py from starlette.middleware.base import BaseHTTPMiddleware from gateway.security.jwt import decode_external_jwt class TenantResolveMiddleware(BaseHTTPMiddleware): async def dispatch(self, request, call_next): auth request.headers.get(Authorization, ) if auth.startswith(Bearer ): token auth.removeprefix(Bearer ).strip() identity decode_external_jwt(token) # 解析失败会抛 401 request.state.identity identity return await call_next(request)decode_external_jwt用公钥验签解析出tenant_id、user_id、channel等字段。解析失败直接返回 401不进入后续流程。这一步挡住了绝大多数伪造身份的请求。接下来是转发路由它把身份上下文和签名一起传给 agent service# gateway/routers/agent_proxy.py import hashlib import hmac import time import httpx from fastapi import APIRouter, Request, Response from gateway.config import get_internal_secret router APIRouter() TENANT_HEADER X-Internal-Tenant USER_HEADER X-Internal-User CHANNEL_HEADER X-Internal-Channel AUDIENCE_HEADER X-Internal-Audience TIMESTAMP_HEADER X-Internal-Timestamp SIGNATURE_HEADER X-Internal-Signature def _canonical_payload(identity, audience: str, issued_at: int) - str: return |.join([ str(identity.tenant_id), str(identity.user_id), identity.channel, audience, str(issued_at), ]) def build_internal_identity_headers(identity, audience: str, secret: str, timestamp: int | None None) - dict: issued_at timestamp or int(time.time()) signature hmac.new( secret.encode(utf-8), _canonical_payload(identity, audience, issued_at).encode(utf-8), hashlib.sha256, ).hexdigest() return { TENANT_HEADER: str(identity.tenant_id), USER_HEADER: str(identity.user_id), CHANNEL_HEADER: identity.channel, AUDIENCE_HEADER: audience, TIMESTAMP_HEADER: str(issued_at), SIGNATURE_HEADER: signature, } router.api_route(/agent/{path:path}, methods[GET, POST]) async def proxy_agent_request(path: str, request: Request) - Response: identity getattr(request.state, identity, None) if identity is None: return Response(status_code401, contentmissing identity) audience agent-service headers build_internal_identity_headers( identity, audienceaudience, secretget_internal_secret() ) headers[Content-Type] request.headers.get(Content-Type, application/json) body await request.body() target_url f{AGENT_SERVICE_BASE}/{path} async with httpx.AsyncClient(timeout30.0) as client: upstream await client.request( request.method, target_url, contentbody or None, headersheaders, paramsrequest.query_params, ) return Response( contentupstream.content, status_codeupstream.status_code, headers{Content-Type: upstream.headers.get(Content-Type, application/json)}, )这段代码的核心是build_internal_identity_headers把身份字段按固定顺序拼成规范字符串用 HMAC-SHA256 签名。agent service 收到后会用同样的方式重算签名不一致就拒绝。这样即使有人能访问 agent service 的内网端口没有密钥也伪造不出合法签名。注意_canonical_payload的字段顺序必须两端完全一致任何一端改了顺序都会导致验签失败。建议把这个函数抽到共享模块或者用同一份代码生成避免手写不一致。4. 验证请求agent service 验签、MCP 内部 JWT 与 Java 侧解析agent service 收到转发请求后第一件事是验签。验签通过才把身份放进 LangGraph 的 Runnable config后续调用工具时再生成内部 JWT 去连 Java MCP Server。先看验签逻辑# agent_service/security/internal_identity.py import hmac from agent_service.security.errors import IdentityError from agent_service.config import get_internal_secret from gateway.routers.agent_proxy import build_internal_identity_headers, SIGNATURE_HEADER def verify_internal_identity(headers: dict, audience: str agent-service): required lambda k: headers.get(k) or (_ for _ in ()).throw(IdentityError(fmissing {k})) tenant_id required(X-Internal-Tenant) user_id required(X-Internal-User) channel required(X-Internal-Channel) issued_at int(required(X-Internal-Timestamp)) provided_sig required(SIGNATURE_HEADER) identity SimpleIdentity(tenant_idtenant_id, user_iduser_id, channelchannel) expected build_internal_identity_headers( identity, audienceaudience, secretget_internal_secret(), timestampissued_at )[SIGNATURE_HEADER] if not hmac.compare_digest(provided_sig, expected): raise IdentityError(internal identity signature is invalid) return identityhmac.compare_digest是防时序攻击的写法不要用比较签名。验签通过后身份对象会作为 LangGraph 的 config 传入工具节点调用时再生成 MCP 内部 JWT# agent_service/mcp/client.py import os import time import uuid from datetime import timedelta from agent_service.security.mcp_jwt import build_mcp_jwt def build_mcp_headers(identity, execution_proof: str) - dict: token build_mcp_jwt( identity, secretos.environ[INTERNAL_IDENTITY_SECRET], issueros.environ[MCP_JWT_ISSUER], audienceos.environ[MCP_JWT_AUDIENCE], ttl_secondsint(os.environ[MCP_JWT_TTL_SECONDS]), issued_atint(time.time()), token_idstr(uuid.uuid4()), execution_proofexecution_proof, ) return {Authorization: fBearer {token}} def build_connection(spec) - dict: headers build_mcp_headers(spec.identity, spec.execution_proof) connection { url: os.getenv(spec.url_env, spec.default_url), transport: streamable_http, headers: headers, timeout: timedelta(secondsspec.service_timeout), sse_read_timeout: timedelta(secondsspec.service_read_timeout), } return {spec.server_name: connection}execution_proof是一次工具调用的唯一标识可以绑定到具体的执行上下文防止 token 被挪用到别的调用上。token_id配合短 TTL服务端可以做一次性校验或重放检测。Java 侧作为 MCP Server需要在入口处解析这个内部 JWT。用 Spring 的话写一个OncePerRequestFilter// java-mcp-server/src/main/java/com/example/mcp/InternalJwtFilter.java Component public class InternalJwtFilter extends OncePerRequestFilter { private final JwtVerifier verifier; public InternalJwtFilter(JwtVerifier verifier) { this.verifier verifier; } Override protected void doFilterInternal(HttpServletRequest req, HttpServletResponse resp, FilterChain chain) throws ServletException, IOException { String auth req.getHeader(Authorization); if (auth null || !auth.startsWith(Bearer )) { resp.setStatus(401); resp.getWriter().write({\error\:\missing internal jwt\}); return; } try { Identity identity verifier.verify(auth.substring(7)); req.setAttribute(identity, identity); chain.doFilter(req, resp); } catch (JwtException e) { resp.setStatus(401); resp.getWriter().write({\error\:\invalid internal jwt\}); } } }JwtVerifier用共享密钥或公钥验签校验iss、aud、exp并检查token_id是否已被使用。验签通过后把身份写入 request attribute业务 Controller 从这里取用户身份而不是从请求参数里取——这一点很关键参数里的用户 ID 永远不可信。到这里一次完整调用链就闭环了外部 JWT → gateway 解析 → HMAC 签名转发 → agent service 验签 → 生成 MCP 内部 JWT → Java 侧验签 → 业务执行。每一跳都有独立凭证任何一跳被篡改都会在下一跳被拦下。5. 本篇常见错排查401、签名不匹配与 OAuth 报错实际部署时最容易卡在几个固定报错上。下面按真实日志对照排查。报错一401 missing identity或internal identity signature is invalid先确认 gateway 和 agent service 的INTERNAL_IDENTITY_SECRET是否完全一致。常见坑是 gateway 从.env读、agent service 从容器环境变量读两边值不同。用下面命令比对# 在 gateway 和 agent service 容器内分别执行 python -c import os; print(os.environ[INTERNAL_IDENTITY_SECRET][:8])如果前 8 位不一致就是配置没同步。另外检查_canonical_payload的字段顺序任何一端改了顺序都会导致签名不匹配。建议把签名函数抽成共享包两端引用同一份代码。报错二local proxy failed或连接 agent service 超时这通常是 gateway 转发地址配错。检查AGENT_SERVICE_BASE是否指向 agent service 的实际监听地址和端口。如果 agent service 在容器里不要用localhost要用服务名或容器网络 IP。另外确认 agent service 的端口没有被防火墙挡住内网也要放行。报错三reading choices或模型调用返回空这个报错一般出现在模型调用环节说明 TaoToken 通道的响应解析出了问题。先确认TAOTOKEN_BASE_URL是https://taotoken.net/api不要多加/v1或漏掉路径。然后检查TAOTOKEN_MODEL_ID是否是通道支持的模型名。可以用 curl 快速验证curl -s https://taotoken.net/api/v1/chat/completions \ -H Authorization: Bearer $TAOTOKEN_API_KEY \ -H Content-Type: application/json \ -d {model:claude-sonnet-4-5,messages:[{role:user,content:ping}]}如果返回正常说明 Key 和通道没问题问题在 Python 侧的解析逻辑。如果返回 401去https://taotoken.net/api-keys确认 Key 是否有效、是否被禁用。报错四Java 侧OAuth或JWT audience mismatchMCP 内部 JWT 的aud必须和 Java 侧配置的 audience 完全一致。检查MCP_JWT_AUDIENCE和 Java 的JwtVerifier里配置的 audience 是否相同。另外iss也要匹配exp过期时间不要设太长120 秒足够。如果 Java 侧用的是 OAuth 资源服务器配置注意不要和内部 JWT 的校验逻辑混在一起两者是独立的鉴权层。报错五越权请求没有被拦截如果你按上面的链路做了但构造一个X-Internal-User: 999的请求仍然能通过说明 agent service 没有强制验签或者验签逻辑被绕过了。检查 agent service 的路由是否都经过verify_internal_identity有没有哪个接口直接读了请求头里的用户 ID。用下面这个测试请求验证拦截效果# 直接打 agent service不带合法签名应该返回 401 curl -i -X POST http://agent-service:8001/agent/query \ -H X-Internal-Tenant: 1 \ -H X-Internal-User: 999 \ -H X-Internal-Channel: web \ -H X-Internal-Timestamp: 1700000000 \ -H X-Internal-Signature: deadbeef预期结果是401 internal identity signature is invalid。如果返回了业务数据说明验签没生效回去检查中间件注册顺序和路由依赖。排查时记住一个原则每一跳的凭证只对下一跳有效。外部 JWT 不能直接拿去调 Java 系统内部 HMAC 签名不能替代 MCP JWT。分层校验虽然多写点代码但出问题时定位非常快——看是哪一跳的 401就知道是哪层凭证没对上。6. 把安全链路固化成可复用的接入方式整套链路跑通后建议把关键配置和调用方式固化下来避免每次加新工具都重新踩坑。模型调用统一走 TaoToken 通道Key 在https://taotoken.net/api-keys管理接入文档在https://taotoken.net/doc需要调试模型行为时用https://taotoken.net/chat快速验证。长期做 Agent 开发的话Coding Plan 入口在https://taotoken.net/coding-plan可以把开发期的调用也纳入统一审计。真正要固化的其实是三件事签名函数的共享、内部 JWT 的短 TTL、以及 Java 侧从 request attribute 取身份而不是从参数取。这三条做到了即使后面加十个新工具安全边界也不会松。我试过在网关层加一个统一的签名校验装饰器所有转发路由自动带上新增接口时不用重复写签名逻辑出错概率会低很多。最后留一个实用技巧在 gateway 的审计日志里记录token_id和execution_proof这样一次越权尝试能直接关联到具体的调用链排查时不用翻一堆日志。审计字段不用多tenant_id、user_id、path、timestamp、signature_valid这五个就够定位问题。