企业官网建设流程全解析

1. 为什么 Python Agent 调 Java 老系统，安全链路最容易断在中间

很多团队做 Agent 落地时，Python 侧写得飞快，Java 侧的老业务系统却动不得——它承载着订单、库存、结算这些核心逻辑，接口早就稳定运行了好几年。于是常见做法是：把 Java 系统包装成 MCP Server 暴露工具，Python Agent 作为 MCP Client 去调用。听起来很顺，但真正上线前你会发现，agent（Python）+传统业务系统（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/console，Key 在https://taotoken.net/api-keys生成。

具体配置上，Python 侧用 OpenAI 兼容方式接入，把base_url指向 TaoToken 即可。这样 LangGraph 里的模型节点、MCP Server 里的辅助模型调用，都能复用同一套凭证。下面是一个可复制的环境变量片段，放在 gateway 和 agent service 共用的.env里：

# .env 统一模型通道配置 TAOTOKEN_BASE_URL=https://taotoken.net/api TAOTOKEN_API_KEY=sk-你的统一Key TAOTOKEN_MODEL_ID=claude-sonnet-4-5 # 内部签名密钥，gateway 与 agent service 必须一致 INTERNAL_IDENTITY_SECRET=请换成32位以上随机串 # MCP 内部 JWT 配置 MCP_JWT_ISSUER=agent-service MCP_JWT_AUDIENCE=java-mcp-server MCP_JWT_TTL_SECONDS=120

注意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(title="Agent 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_code=401, content="missing identity") audience = "agent-service" headers = build_internal_identity_headers( identity, audience=audience, secret=get_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(timeout=30.0) as client: upstream = await client.request( request.method, target_url, content=body or None, headers=headers, params=request.query_params, ) return Response( content=upstream.content, status_code=upstream.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(f"missing {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_id=tenant_id, user_id=user_id, channel=channel) expected = build_internal_identity_headers( identity, audience=audience, secret=get_internal_secret(), timestamp=issued_at )[SIGNATURE_HEADER] if not hmac.compare_digest(provided_sig, expected): raise IdentityError("internal identity signature is invalid") return identity

hmac.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, secret=os.environ["INTERNAL_IDENTITY_SECRET"], issuer=os.environ["MCP_JWT_ISSUER"], audience=os.environ["MCP_JWT_AUDIENCE"], ttl_seconds=int(os.environ["MCP_JWT_TTL_SECONDS"]), issued_at=int(time.time()), token_id=str(uuid.uuid4()), execution_proof=execution_proof, ) return {"Authorization": f"Bearer {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(seconds=spec.service_timeout), "sse_read_timeout": timedelta(seconds=spec.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 mismatch

MCP 内部 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这五个就够定位问题。