AI 工具网关：在 Kubernetes 中沙盒化 Agent 访问

AI agent 拥有无限的野心和未定义的访问边界。你已经沙盒化了其他一切。以下是如何也沙盒化你的 agent。

2026-06-299 分钟阅读

你已经沙盒化了容器。你已经沙盒化了 Lambda 函数。你已经通过最小权限 IAM 角色沙盒化了数据库连接。但你没有沙盒化你的 AI agent。

这是大多数平台团队尚未触及的 AI 采用曲线的一部分。讨论一直集中在能力上——agent 能做什么，哪个模型最好，如何将工具链在一起。安全讨论紧随其后，当出现问题时会展开：一个 agent 调用它本无权访问的删除端点，通过提示注入泄露凭据，或者因为无人设置速率限制而以每分钟 10,000 次请求冲击外部 API。

AI 工具网关是 agent 与其可以调用的工具之间的代理层。它在边界上执行允许列表、速率限制、认证和日志记录——就像 API 网关为人工调用者执行这些操作一样。区别在于调用者是一个 LLM，它可能被提示去做人类不会尝试的事情。

注意

本地平台工程系列

1. 使用 kind 运行本地 Kubernetes：设计上短暂，生产环境诚实
2. Gateway API 实践：从 Ingress 迁移到 Envoy 调试
3. 多租户可观测性：平台规模的 LGTM
4. 使用 Cilium 和 Kyverno 的网络控制：真正有效的策略
5. 观察 LLM 推理：真正重要的指标
6. AI 工具网关：在 Kubernetes 中沙盒化 Agent 访问（你在这里）

---

AI Agent 的威胁模型

在构建网关之前，你需要一个威胁模型。AI agent 的攻击面与服务不同，因为：

调用者是非确定性的。人类 API 调用者有一组固定的操作会去尝试。Agent 的操作取决于提示、对话历史、可用工具以及模型对这三者的解释。同一个 agent 在周二的行为可能与周一不同，因为上下文发生了变化。

工具调用参数由 LLM 生成。当一个 agent 调用 search_database(query="...") 时，查询字符串由模型生成，而不是由开发者编写。对 agent 的精心构造输入（提示注入）可能导致它生成带有旨在泄露数据或引起副作用的参数的调用。

Agent 可以自主链式调用。一个多步骤的 agent 可能在没有人工审查的情况下按顺序调用五个工具。如果第三步产生了意外结果，agent 可能会尝试通过调用额外工具来纠正——包括那些它已被授权但预期不会被使用的工具。

默认情况下，配置错误的 agent 的爆炸半径是无界的。一个拥有 GitHub token 和文件系统工具访问权限的 agent，如果被提示，可以读取每个仓库并写入任何文件。你未打算这样做的事实是无关紧要的。

网关通过将能力降低到明确需要的范围、对明确允许的操作进行速率限制，并记录所有内容以供审计来解决这个问题。

---

工具网关的作用

工具网关位于 agent 及其工具之间。当一个 agent 想要调用一个工具时，请求会经过网关，网关执行以下操作：

1. 验证 agent 身份（这是哪个 agent？）
2. 授权调用（这个 agent 是否被允许调用此工具？）
3. 验证参数（参数是否匹配架构？它们是否通过内容检查？）
4. 限制调用速率（这个 agent 是否超出其配额？）
5. 记录调用（agent 调用了什么，使用什么参数，以及返回了什么？）
6. 转发到实际工具
7. 返回结果——或者一个经过净化处理的结果

这与 API 网关没有根本区别。操作上的区别在于，你需要在 agent 可以调用的每个工具之前都部署这个网关，因为每个工具都是一个潜在的爆炸半径。

---

在 Kubernetes 上构建工具网关

最简单的实现方式是使用 Envoy Gateway（来自第二部分）加上一个用于策略强制执行的 Envoy 扩展过滤器。对于生产环境，专用工具如 Portkey、LiteLLM Proxy 或自定义的 FastAPI 网关能提供更多控制。

选项 1：使用流量策略的 Envoy Gateway

对于基于 HTTP 的工具，第二部分中的网关已经通过 EnvoyProxy 策略资源处理了认证和速率限制：

manifests/gateway/agent-traffic-policy.yaml

COPYapiVersion: gateway.envoyproxy.io/v1alpha1
kind: BackendTrafficPolicy
metadata:
  name: agent-rate-limit
  namespace: infra
spec:
  targetRef:
    group: gateway.networking.k8s.io
    kind: HTTPRoute
    name: tool-github
    namespace: agents
  rateLimit:
    type: Global
    global:
      rules:
        - clientSelectors:
            - headers:
                - type: Distinct
                  name: x-agent-id
          limit:
            requests: 100
            unit: Minute

manifests/gateway/agent-traffic-policy.yaml

COPYapiVersion: gateway.envoyproxy.io/v1alpha1
kind: BackendTrafficPolicy
metadata:
  name: agent-rate-limit
  namespace: infra
spec:
  targetRef:
    group: gateway.networking.k8s.io
    kind: HTTPRoute
    name: tool-github
    namespace: agents
  rateLimit:
    type: Global
    global:
      rules:
        - clientSelectors:
            - headers:
                - type: Distinct
                  name: x-agent-id
          limit:
            requests: 100
            unit: Minute

基于 x-agent-id 头的速率限制意味着每个 agent 都有自己的配额。一个 agent 达到其限制不会影响其他 agent。

对于认证，使用指向 OIDC 签发服务的 Envoy 外部认证过滤器。Agent 使用限定在其允许工具集范围内的短期令牌进行认证。网关在转发请求之前验证令牌。

选项 2：专用网关服务

为了获得更多控制——参数验证、内容检查、结构化的审计日志——一个轻量级的 FastAPI 网关是实用的：

gateway/main.py

COPYfrom fastapi import FastAPI, Request, HTTPException, Depends
from fastapi.responses import JSONResponse
import httpx
import logging
import time
from typing import Any
 
app = FastAPI()
logger = logging.getLogger("tool-gateway")
 
TOOL_ALLOW_LIST: dict[str, dict] = {
    "search": {
        "upstream": "http://search-service.tools:8080",
        "allowed_agents": ["research-agent", "support-agent"],
        "rate_limit_per_minute": 60,
        "max_query_length": 500,
    },
    "code_executor": {
        "upstream": "http://executor.tools:8080",
        "allowed_agents": ["dev-agent"],
        "rate_limit_per_minute": 10,
        "blocked_patterns": ["rm -rf", "DROP TABLE", "os.system"],
    },
}
 
request_counts: dict[str, list[float]] = {}
 
def check_rate_limit(agent_id: str, tool_name: str, limit: int) -> None:
    key = f"{agent_id}:{tool_name}"
    now = time.time()
    window = 60.0
    calls = [t for t in request_counts.get(key, []) if now - t < window]
    if len(calls) >= limit:
        raise HTTPException(status_code=429, detail="Rate limit exceeded")
    calls.append(now)
    request_counts[key] = calls
 
@app.post("/tools/{tool_name}")
async def call_tool(tool_name: str, request: Request) -> JSONResponse:
    agent_id = request.headers.get("x-agent-id")
    if not agent_id:
        raise HTTPException(status_code=401, detail="Missing x-agent-id header")
 
    tool = TOOL_ALLOW_LIST.get(tool_name)
    if not tool:
        raise HTTPException(status_code=404, detail=f"Tool '{tool_name}' not found")
 
    if agent_id not in tool["allowed_agents"]:
        logger.warning(f"Agent {agent_id} attempted unauthorized access to {tool_name}")
        raise HTTPException(status_code=403, detail="Agent not authorised for this tool")
 
    check_rate_limit(agent_id, tool_name, tool["rate_limit_per_minute"])
 
    body = await request.json()
    args = body.get("arguments", {})
 
    # Content inspection
    for pattern in tool.get("blocked_patterns", []):
        for v in args.values():
            if isinstance(v, str) and pattern in v:
                logger.error(f"Blocked pattern '{pattern}' in {tool_name} call from {agent_id}")
                raise HTTPException(status_code=400, detail="Argument contains blocked content")
 
    # Audit log
    logger.info({
        "event": "tool_call",
        "agent_id": agent_id,
        "tool": tool_name,
        "arguments": args,
        "timestamp": time.time(),
    })
 
    async with httpx.AsyncClient() as client:
        response = await client.post(
            tool["upstream"] + request.url.path,
            json=body,
            headers={"x-forwarded-agent": agent_id},
            timeout=30.0,
        )
 
    return JSONResponse(content=response.json(), status_code=response.status_code)

gateway/main.py

COPYfrom fastapi import FastAPI, Request, HTTPException, Depends
from fastapi.responses import JSONResponse
import httpx
import logging
import time
from typing import Any
 
app = FastAPI()
logger = logging.getLogger("tool-gateway")
 
TOOL_ALLOW_LIST: dict[str, dict] = {
    "search": {
        "upstream": "http://search-service.tools:8080",
        "allowed_agents": ["research-agent", "support-agent"],
        "rate_limit_per_minute": 60,
        "max_query_length": 500,
    },
    "code_executor": {
        "upstream": "http://executor.tools:8080",
        "allowed_agents": ["dev-agent"],
        "rate_limit_per_minute": 10,
        "blocked_patterns": ["rm -rf", "DROP TABLE", "os.system"],
    },
}
 
request_counts: dict[str, list[float]] = {}
 
def check_rate_limit(agent_id: str, tool_name: str, limit: int) -> None:
    key = f"{agent_id}:{tool_name}"
    now = time.time()
    window = 60.0
    calls = [t for t in request_counts.get(key, []) if now - t < window]
    if len(calls) >= limit:
        raise HTTPException(status_code=429, detail="Rate limit exceeded")
    calls.append(now)
    request_counts[key] = calls
 
@app.post("/tools/{tool_name}")
async def call_tool(tool_name: str, request: Request) -> JSONResponse:
    agent_id = request.headers.get("x-agent-id")
    if not agent_id:
        raise HTTPException(status_code=401, detail="Missing x-agent-id header")
 
    tool = TOOL_ALLOW_LIST.get(tool_name)
    if not tool:
        raise HTTPException(status_code=404, detail=f"Tool '{tool_name}' not found")
 
    if agent_id not in tool["allowed_agents"]:
        logger.warning(f"Agent {agent_id} attempted unauthorized access to {tool_name}")
        raise HTTPException(status_code=403, detail="Agent not authorised for this tool")
 
    check_rate_limit(agent_id, tool_name, tool["rate_limit_per_minute"])
 
    body = await request.json()
    args = body.get("arguments", {})
 
    # Content inspection
    for pattern in tool.get("blocked_patterns", []):
        for v in args.values():
            if isinstance(v, str) and pattern in v:
                logger.error(f"Blocked pattern '{pattern}' in {tool_name} call from {agent_id}")
                raise HTTPException(status_code=400, detail="Argument contains blocked content")
 
    # Audit log
    logger.info({
        "event": "tool_call",
        "agent_id": agent_id,
        "tool": tool_name,
        "arguments": args,
        "timestamp": time.time(),
    })
 
    async with httpx.AsyncClient() as client:
        response = await client.post(
            tool["upstream"] + request.url.path,
            json=body,
            headers={"x-forwarded-agent": agent_id},
            timeout=30.0,
        )
 
    return JSONResponse(content=response.json(), status_code=response.status_code)

这不是生产级的——它使用内存中的速率限制，而不是 Redis——但结构是正确的。关键部分：首先是允许列表（没有声明的工具不存在），agent 授权检查，速率限制，内容检查，结构化审计日志，然后转发。

将其部署为专用 agents 命名空间中的 Kubernetes Deployment，位于来自网关的 HTTPRoute 之后：

manifests/gateway/httproutes.yaml (addition)

COPYapiVersion: gateway.networking.k8s.io/v1
kind: HTTPRoute
metadata:
  name: tool-gateway
  namespace: agents
spec:
  parentRefs:
    - name: platform-gateway
      namespace: infra
  hostnames:
    - "tools.local"
  rules:
    - backendRefs:
        - name: tool-gateway
          port: 8080

manifests/gateway/httproutes.yaml (addition)

COPYapiVersion: gateway.networking.k8s.io/v1
kind: HTTPRoute
metadata:
  name: tool-gateway
  namespace: agents
spec:
  parentRefs:
    - name: platform-gateway
      namespace: infra
  hostnames:
    - "tools.local"
  rules:
    - backendRefs:
        - name: tool-gateway
          port: 8080

Agent 在 http://tools.local:8080/tools/{tool_name} 调用工具。网关是唯一的入口。

---

Agent 工作负载的网络隔离

网关是软件——它可能失败、被错误配置，或者如果 agent 对工具有直接网络访问权限，则可能被绕过。Cilium 网络策略（来自第四部分）在网络层强制执行这一点：

manifests/netpol/agent-isolation.yaml

COPY# Agents can only reach the tool gateway, not tools directly
apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: agents-egress
  namespace: agents
spec:
  podSelector:
    matchLabels:
      role: ai-agent
  policyTypes:
    - Egress
  egress:
    - to:
        - namespaceSelector:
            matchLabels:
              kubernetes.io/metadata.name: agents
          podSelector:
            matchLabels:
              app: tool-gateway
      ports:
        - port: 8080
          protocol: TCP
    - to:  # DNS
        - namespaceSelector: {}
      ports:
        - port: 53
          protocol: UDP
---
# Tool services only accept calls from the gateway, not directly from agents
apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: tools-ingress
  namespace: tools
spec:
  podSelector: {}
  policyTypes:
    - Ingress
  ingress:
    - from:
        - namespaceSelector:
            matchLabels:
              kubernetes.io/metadata.name: agents
          podSelector:
            matchLabels:
              app: tool-gateway

manifests/netpol/agent-isolation.yaml

COPY# Agents can only reach the tool gateway, not tools directly
apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: agents-egress
  namespace: agents
spec:
  podSelector:
    matchLabels:
      role: ai-agent
  policyTypes:
    - Egress
  egress:
    - to:
        - namespaceSelector:
            matchLabels:
              kubernetes.io/metadata.name: agents
          podSelector:
            matchLabels:
              app: tool-gateway
      ports:
        - port: 8080
          protocol: TCP
    - to:  # DNS
        - namespaceSelector: {}
      ports:
        - port: 53
          protocol: UDP
---
# Tool services only accept calls from the gateway, not directly from agents
apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: tools-ingress
  namespace: tools
spec:
  podSelector: {}
  policyTypes:
    - Ingress
  ingress:
    - from:
        - namespaceSelector:
            matchLabels:
              kubernetes.io/metadata.name: agents
          podSelector:
            matchLabels:
              app: tool-gateway

现在，架构在两个层面得到强制执行：网关软件层和 Cilium 网络层。即使有 Bug 或配置错误的允许列表，agent 也无法绕过网关。

---

审计日志和可观测性

网关发出结构化的 JSON 日志。Promtail（来自第三部分）会自动采集它们。在 Loki 中，查询：

COPY{namespace="agents", app="tool-gateway"} | json | event="tool_call"
| line_format "{{.agent_id}} → {{.tool}} ({{.arguments}})"

COPY{namespace="agents", app="tool-gateway"} | json | event="tool_call"
| line_format "{{.agent_id}} → {{.tool}} ({{.arguments}})"

这能给你一个每个 agent 所做的每个工具调用的按时间顺序的记录，可以按 agent ID、工具名称或时间范围进行过滤。对 agent 事件进行合规审查变成了一个 Loki 查询，而不是一次日志考古挖掘。

对于指标，向网关添加 Prometheus 仪表化：

COPYfrom prometheus_client import Counter, Histogram
 
tool_calls_total = Counter(
    "tool_gateway_calls_total",
    "Total tool calls",
    ["agent_id", "tool", "status"]
)
tool_call_duration = Histogram(
    "tool_gateway_duration_seconds",
    "Tool call duration",
    ["agent_id", "tool"]
)

COPYfrom prometheus_client import Counter, Histogram
 
tool_calls_total = Counter(
    "tool_gateway_calls_total",
    "Total tool calls",
    ["agent_id", "tool", "status"]
)
tool_call_duration = Histogram(
    "tool_gateway_duration_seconds",
    "Tool call duration",
    ["agent_id", "tool"]
)

在 tool_gateway_calls_total{status="403"} 激增时发出告警——这意味着 agent 反复尝试访问其未授权的工具，要么是 Bug，要么是正在进行的提示注入尝试。

---

完整架构

回顾这六个部分：

第一部分中的 kind 集群是基础。Envoy Gateway 将流量路由到所有地方。Cilium 强制执行网络边界。Kyverno 强制执行资源形态。LGTM 栈观察整体。工具网关是 agent 工作负载的访问控制边界。

每一层都在本地运行。每一层在设计上都与生产环境一致。当这迁移到真实集群时，变化的是基础设施（真实的负载均衡器、真实的持久化存储、真实的认证提供者）——架构不会改变。

---

值得借鉴的模式

AI agent 并不是一种需要新安全模型的新型计算。它们是一种新型的调用者，需要有意识地应用相同的安全模型：最小权限访问、速率限制、边界认证、审计日志和网络隔离。

早期做对这件事的平台团队将能够自信地在生产环境中部署 agent 工作负载。那些没有做对的团队将在六个月后阅读一份事件回顾，内容涉及一次提示注入，它调用了一个没人知道 agent 有权限访问的删除端点。

在你需要网关之前就构建好它。当你用一个拥有真实凭据的 agent 时，你就会立刻明白为什么。

← 上一篇观察 LLM 推理：真正重要的指标