End-User and Platform Approval Gates for MCP Agents

git clone https://github.com/tjorourke/solo-labs.git
cd solo/agentic-hitl-kind

export ANTHROPIC_API_KEY=sk-ant-...

./scripts/quick.sh up
./scripts/port-forward.sh   # leave running

# Two independent FastMCP servers. Each has its own tool registry. The
# only thing they share is the Python process and the Starlette router.
public     = FastMCP("ops-tools-public",     stateless_http=True)
privileged = FastMCP("ops-tools-privileged", stateless_http=True)

# Public tier — anything the agent can do without platform sign-off.
# `truncate_table` is still dangerous; agent HITL gates it in the chat.
@public.tool()
def cluster_db_query(sql: str) -> dict: ...   # read-only
@public.tool()
def truncate_table(table: str) -> dict: ...   # mutating — agent HITL

# Privileged tier — platform-team-only operations.
# The gateway has extAuth on /privileged, so every call here parks at
# hitl-extauth until a platform reviewer approves it in the hitl-ui.
@privileged.tool()
def run_migration(version: str) -> dict: ...  # gateway HITL

# Mount each MCP server under its own path. The gateway distinguishes
# them by URL prefix — no JSON-RPC body inspection, no CEL expression.
app = Starlette(routes=[
    Mount("/public",     app=public.streamable_http_app()),
    Mount("/privileged", app=privileged.streamable_http_app()),
])

apiVersion: gateway.networking.k8s.io/v1
kind: HTTPRoute
metadata: { name: mcp-privileged, namespace: ops-tools }
spec:
  parentRefs: [{ name: hitl-gateway, namespace: agentgateway-system }]
  rules:
    - matches: [{ path: { type: PathPrefix, value: /privileged } }]
      backendRefs: [{ name: ops-tools, port: 8080 }]
---
apiVersion: agentgateway.dev/v1alpha1
kind: AgentgatewayPolicy
metadata: { name: privileged-extauth, namespace: ops-tools }
spec:
  targetRefs:
    - { group: gateway.networking.k8s.io, kind: HTTPRoute, name: mcp-privileged }
  traffic:
    # phase defaults to PostRouting. PreRouting can only target a Gateway
    # or Listener; we target an HTTPRoute to gate only the /privileged path,
    # so the default is what we want.
    extAuth:
      backendRef: { name: hitl-extauth, namespace: hitl, port: 9001 }
      forwardBody: { maxSize: 65536 }   # ext-auth uses this to skip
      grpc: {}                          # non-tools/call (initialize, etc.)

apiVersion: kagent.dev/v1alpha2
kind: Agent
metadata: { name: dba-assistant, namespace: kagent }
spec:
  type: Declarative
  declarative:
    modelConfig: default-model-config
    systemMessage: |
      You are a DBA assistant for a small orders database. ...
    tools:
      - type: McpServer
        mcpServer:
          kind: RemoteMCPServer
          name: ops-tools-public
          toolNames: [cluster_db_query, truncate_table]
          requireApproval:                    #  ◄── AGENT HITL
            - truncate_table
      - type: McpServer
        mcpServer:
          kind: RemoteMCPServer
          name: ops-tools-privileged
          toolNames: [run_migration]
          # no requireApproval — the gateway extAuth policy is the gate

TOOLS_REQUIRING_APPROVAL = {"truncate_table"}

async def run_tools(state):
    last = state["messages"][-1]
    results = []
    for tc in last.tool_calls:
        # AGENT HITL: pause the graph and surface a card in the kagent chat.
        # The kagent executor needs this exact shape to render an approval UI.
        if tc["name"] in TOOLS_REQUIRING_APPROVAL:
            decision = interrupt({
                "action_requests": [
                    {"name": tc["name"], "args": tc["args"], "id": tc["id"]}
                ]
            })
            if decision.get("decision_type") != "approve":
                results.append(ToolMessage(
                    content="Tool call was rejected by user.",
                    tool_call_id=tc["id"], name=tc["name"]))
                continue
        # Either an approved sensitive tool, or a normal one. Gateway HITL
        # (if applicable) fires transparently inside ainvoke — the agent
        # has no knowledge of it.
        result = await TOOL_MAP[tc["name"]].ainvoke(tc["args"])
        results.append(ToolMessage(
            content=str(result), tool_call_id=tc["id"], name=tc["name"]))
    return {"messages": results}

# Mock DB state + audit log
kubectl --context kind-hitl -n ops-tools port-forward svc/ops-tools 8081:8080 &
curl -s localhost:8081/state | jq

# Pending requests directly from ext-auth (bypass the UI)
kubectl --context kind-hitl -n hitl exec deploy/hitl-extauth -- \
  wget -qO- http://localhost:8081/pending | jq

# Watch ext-auth log lines as Check() is called and decisions arrive
kubectl --context kind-hitl -n hitl logs -f deploy/hitl-extauth