Skip to main content

Protect your first agent

Getting Started

This path takes one agent that runs in your Kubernetes cluster from its first call through the gateway to Enforce mode. You don't design Access Bindings or Blueprints up front: AIControls learns them from what the agent does. Plan on about 30 minutes, most of it waiting for the agent to make calls.

Self-managed installations only

Workload Trust is available only in private, self-managed AIControls installations, on premises or in a cloud cluster that you run. It needs a dedicated cluster that you run, with Kyverno, Kyverno Runtime, and cluster OIDC and RBAC access in place. Nirmata-hosted (SaaS) workspaces don't include it. See Feature availability.

The Dashboard tracks this path in the Protect your first agent card. Each step links to the screen where you do it, and is checked off when AIControls sees it done. Dismiss the card to hide it.

Before you begin​

  • A Kubernetes cluster and permission to install a Helm chart and deploy a workload.
  • An API key for an LLM provider, such as an Anthropic sk-ant-... key. To try this path at no cost, with a free model and a public MCP server, follow Try AIControls with free models and a public MCP server.
Try it on a local kind cluster

If you have a checkout of the AIControls source, one command creates a kind cluster, installs AIControls from that checkout, adds an Anthropic provider with your key, and deploys the sample agent from step 3:

ANTHROPIC_API_KEY=sk-ant-... DEPLOY_AGENT=1 make kind-quickstart

It also adds the public DeepWiki MCP server as deepwiki, which the sample agent calls. It prints the UI URL and the admin login, and starts a port-forward. The cluster gets its own kubeconfig file, ~/.kube/aicontrols-quickstart, so your default kubeconfig and its current context don't change. Log in, keep Governance for AI agents in Kubernetes selected in the setup wizard, and continue at step 4. To delete the cluster, run make kind-quickstart-down. For the options, see Quickstart on kind.

1. Install​

  1. Install AIControls with the Helm chart. Follow Self-hosted Kubernetes getting started, and turn on the LLM gateway:

    llmProxy:
      enabled: true

    The chart defaults do the rest:

    • Agent identity. At startup, AIControls reads your cluster's ServiceAccount token issuer from the Kubernetes API and verifies each agent's projected token for the aicontrols audience. The pod log shows cluster OIDC issuer discovered from the Kubernetes API with the issuer. To use another issuer or audience, set identity.clusterOIDCIssuer and identity.clusterOIDCAudience. See Issuer discovery.
    • Workload attribution. workloadIdentity.podRead.enabled is on, so the gateway can map each caller's Pod to its Deployment and discover the agent from its first call. It adds a cluster-wide ClusterRole with get on Pods and their owner kinds, so AIControls can read any Pod's spec in any namespace.
    • Persistent data. The data directory is a 5 GiB PersistentVolumeClaim from your default StorageClass, so registrations and learned access survive a pod restart.
  2. Log in. The setup wizard opens on How will you use AIControls?. Keep Governance for AI agents in Kubernetes selected. A new installation blocks and records calls from agents you haven't registered yet, and you can change that later. See Unregistered agents.

2. Set up a provider and test it​

  1. In the setup wizard's Model step, choose your provider, enter its key and a model, and select Save and test. The step is done when a test call through the gateway answers. The wizard grants the model to the default Blueprint, so agents you register with it can call the model. See Connect a model. Or go to Settings → Providers, select Add Upstream, choose any provider, enter its API key, and select Test Connection. See Add a provider.
  2. Optional: add an MCP server on Inventory → MCP Servers, and test it with Test connection on that page. The sample agent calls DeepWiki's read_wiki_structure tool by default; with make kind-quickstart, DeepWiki is already there as deepwiki. See Connect an MCP Server.

3. Deploy an agent that calls the gateway​

The sample agent below is a small FAQ agent: every five minutes it answers a support question with the model in MODEL, and calls the MCP tool in MCP_TOOL. By default, it calls the free OpenRouter model nvidia/nemotron-3.5-lightning:free and DeepWiki's read_wiki_structure tool. It calls right away when it starts, so to see the effect of a change sooner, restart it with kubectl -n support rollout restart deploy/faq-agent. It's a short Python script, with no dependencies, in the public python:3.13-alpine image. It authenticates with its own projected ServiceAccount token and has no provider key of its own.

  • If your release isn't named aicontrols in the aicontrols namespace, change GATEWAY_URL.
  • If you didn't set up OpenRouter, set MODEL to a model your upstream serves.
  • If you didn't add DeepWiki as deepwiki, set MCP_TOOL to a tool on your own MCP server, as <server>__<tool>, or to "" for LLM calls only. For Anthropic, set MODEL to claude-sonnet-4-6 and API to anthropic, so the agent calls the Anthropic endpoint (/v1/messages) instead of the OpenAI-compatible one (/v1/chat/completions).
apiVersion: v1
kind: Namespace
metadata:
  name: support
---
apiVersion: v1
kind: ServiceAccount
metadata:
  name: faq-agent
  namespace: support
---
apiVersion: v1
kind: ConfigMap
metadata:
  name: faq-agent
  namespace: support
data:
  GATEWAY_URL: "http://aicontrols.aicontrols.svc.cluster.local:8080"
  # A free OpenRouter model. With an Anthropic upstream: claude-sonnet-4-6 and API "anthropic".
  MODEL: "nvidia/nemotron-3.5-lightning:free"
  API: "openai"
  # An MCP tool as <server>__<tool> and its JSON arguments: DeepWiki's
  # read_wiki_structure for a public repository. Empty for LLM calls only.
  MCP_TOOL: "deepwiki__read_wiki_structure"
  MCP_ARGS: '{"repoName":"facebook/react"}'
  INTERVAL: "300"
  agent.py: |
    # faq-agent: answers a support FAQ question with an LLM every INTERVAL
    # seconds and, when MCP_TOOL is set, calls one MCP tool. Every call goes
    # through the AIControls gateway with the Pod's projected ServiceAccount
    # token, so the agent has no provider key. Each call logs one line:
    #   ALLOWED  the gateway let the call through
    #   BLOCKED  AIControls denied it, with the reason
    #   ERROR    the gateway or the provider failed, with the status
    # Standard library only, so it runs in a stock python image.
    import itertools, json, os, signal, sys, time, urllib.error, urllib.request

    GATEWAY = os.environ.get("GATEWAY_URL", "http://aicontrols.aicontrols.svc.cluster.local:8080").rstrip("/")
    MODEL = os.environ.get("MODEL", "nvidia/nemotron-3.5-lightning:free")
    API = os.environ.get("API", "openai").lower()  # openai: /v1/chat/completions, anthropic: /v1/messages
    MCP_TOOL = os.environ.get("MCP_TOOL", "").strip()  # <server>__<tool>, for example github__list_repos
    MCP_ARGS = os.environ.get("MCP_ARGS", "") or "{}"
    INTERVAL = float(os.environ.get("INTERVAL", "300"))
    TOKEN_FILE = os.environ.get("TOKEN_FILE", "/var/run/secrets/aicontrols/token")
    SYSTEM = "You are the FAQ assistant for Acme Cloud support. Answer in at most two short sentences."
    # Room for a reasoning model's hidden reasoning; free models cost nothing per token.
    MAX_TOKENS = int(os.environ.get("MAX_TOKENS", "800"))
    QUESTIONS = [
        "How do I reset my password?",
        "Can I change the billing email on my account?",
        "How do I export my data?",
        "How do I add a teammate to my workspace?",
    ]
    # Reason codes on the gateway's own denials.
    GATEWAY_CODES = ("workload_not_registered", "POLICY_VIOLATION", "MODEL_DENIED", "CONTENT_BLOCKED",
                     "IDENTITY_REQUIRED", "UNVERIFIED_IDENTITY", "BUDGET_EXHAUSTED", "AGENT_QUARANTINED")


    def log(msg):
        print(time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), msg, flush=True)


    def short(text, n=200):
        text = " ".join(str(text).split())
        return text if len(text) <= n else text[: n - 1] + "…"


    def post(path, body, headers=None):
        """POSTs JSON to the gateway; returns (status, headers, text) for any HTTP status."""
        with open(TOKEN_FILE) as f:  # read every call: the kubelet rotates the token
            token = f.read().strip()
        h = {"Authorization": "Bearer " + token, "Content-Type": "application/json", **(headers or {})}
        req = urllib.request.Request(GATEWAY + path, json.dumps(body).encode(), h, method="POST")
        try:
            with urllib.request.urlopen(req, timeout=60) as r:
                return r.status, r.headers, r.read().decode("utf-8", "replace")
        except urllib.error.HTTPError as e:
            return e.code, e.headers, e.read().decode("utf-8", "replace")


    def failure(status, text):
        """'BLOCKED (403) Blocked by AIControls: …' or 'ERROR (429) …' from an error body."""
        try:
            err = json.loads(text).get("error", text)
        except (ValueError, AttributeError):
            err = text
        msg = err.get("message", json.dumps(err)) if isinstance(err, dict) else str(err)
        if "AIControls" in msg or any(c in msg for c in GATEWAY_CODES):
            if "Blocked by AIControls" not in msg:
                msg = "Blocked by AIControls: " + msg
            return f"BLOCKED ({status})  {short(msg, 300)}"
        code = err.get("code") if isinstance(err, dict) else None
        if code and str(code) not in msg:
            msg = f"{code}: {msg}"
        return f"ERROR ({status})  {short(msg, 300)}"


    def ask(question):
        if API == "anthropic":
            status, _, text = post("/v1/messages", {
                "model": MODEL, "max_tokens": MAX_TOKENS, "system": SYSTEM,
                "messages": [{"role": "user", "content": question}],
            }, {"anthropic-version": "2023-06-01"})
        else:
            status, _, text = post("/v1/chat/completions", {
                "model": MODEL, "max_tokens": MAX_TOKENS,
                "messages": [{"role": "system", "content": SYSTEM}, {"role": "user", "content": question}],
            })
        if status >= 300:
            return log(f"LLM  {MODEL}  {failure(status, text)}")
        try:
            doc = json.loads(text)
            answer = ("".join(b.get("text", "") for b in doc["content"]) if API == "anthropic"
                      else doc["choices"][0]["message"].get("content"))
        except (ValueError, KeyError, IndexError, TypeError):
            answer = text
        log(f"LLM  {MODEL}  ALLOWED  {short(answer) or '(empty answer)'}")


    def rpc(text, content_type):
        """The JSON-RPC message in a JSON or text/event-stream response."""
        lines = [l[5:] for l in text.splitlines() if l.startswith("data:")] if "event-stream" in (content_type or "") else [text]
        for line in lines:
            try:
                return json.loads(line)
            except ValueError:
                continue
        return {}


    def call_tool():
        accept = {"Accept": "application/json, text/event-stream"}
        status, headers, text = post("/mcp", {"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {
            "protocolVersion": "2025-03-26", "capabilities": {},
            "clientInfo": {"name": "faq-agent", "version": "0.2.0"}}}, accept)
        if status >= 300:
            return log(f"MCP  {MCP_TOOL}  {failure(status, text)}")
        if headers.get("Mcp-Session-Id"):
            accept["Mcp-Session-Id"] = headers["Mcp-Session-Id"]
        post("/mcp", {"jsonrpc": "2.0", "method": "notifications/initialized"}, accept)
        status, headers, text = post("/mcp", {"jsonrpc": "2.0", "id": 2, "method": "tools/call",
                                              "params": {"name": MCP_TOOL, "arguments": json.loads(MCP_ARGS)}}, accept)
        msg = rpc(text, headers.get("Content-Type"))
        if status >= 300 or "error" in msg:
            return log(f"MCP  {MCP_TOOL}  {failure(status, json.dumps(msg) if msg else text)}")
        result = msg.get("result", {})
        out = short(" ".join(c.get("text", "") for c in result.get("content", []) if isinstance(c, dict)) or result)
        log(f"MCP  {MCP_TOOL}  {failure(status, out) if result.get('isError') else 'ALLOWED  ' + out}")


    # As PID 1 the default SIGTERM action is ignored; exit promptly on rollout.
    signal.signal(signal.SIGTERM, lambda *_: sys.exit(0))
    log(f"faq-agent: gateway={GATEWAY} api={API} model={MODEL} mcp_tool={MCP_TOOL or '(none)'} every {INTERVAL:g}s")
    for question in itertools.cycle(QUESTIONS):
        log(f"Q: {question}")
        for step in [lambda: ask(question)] + ([call_tool] if MCP_TOOL else []):
            try:
                step()
            except (OSError, ValueError) as e:  # URLError is an OSError
                log(f"ERROR  {e}")
        time.sleep(INTERVAL)
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: faq-agent
  namespace: support
  labels:
    app: faq-agent
    # Declares the Deployment an AI workload, so discovery lists it as
    # confirmed, not suspected, and its primary action is Register.
    nirmata.io/ai: "true"
spec:
  replicas: 1
  selector:
    matchLabels:
      app: faq-agent
  template:
    metadata:
      labels:
        app: faq-agent
    spec:
      serviceAccountName: faq-agent
      terminationGracePeriodSeconds: 5
      automountServiceAccountToken: false
      containers:
        - name: agent
          image: python:3.13-alpine
          command: ["python3", "-u", "/app/agent.py"]
          envFrom:
            - configMapRef:
                name: faq-agent
          env:
            - name: PYTHONDONTWRITEBYTECODE
              value: "1"
          volumeMounts:
            - name: app
              mountPath: /app
            - name: aicontrols-token
              mountPath: /var/run/secrets/aicontrols
              readOnly: true
          resources:
            requests: { cpu: 10m, memory: 32Mi }
            limits: { memory: 128Mi }
          securityContext:
            runAsNonRoot: true
            runAsUser: 65534
            allowPrivilegeEscalation: false
            readOnlyRootFilesystem: true
            capabilities: { drop: ["ALL"] }
      volumes:
        - name: app
          configMap:
            name: faq-agent
            items:
              - key: agent.py
                path: agent.py
        - name: aicontrols-token
          projected:
            sources:
              - serviceAccountToken:
                  path: token
                  expirationSeconds: 3600
                  audience: aicontrols   # must match identity.clusterOIDCAudience

Save it as faq-agent.yaml, apply it, and watch its calls. Each call logs one line: ALLOWED with the answer, BLOCKED with the reason AIControls gave, or ERROR with the status, such as a free model's daily limit (429).

kubectl apply -f faq-agent.yaml
kubectl -n support logs deploy/faq-agent -f

The MCP tool call​

The agent calls the MCP tool in MCP_TOOL, as <server>__<tool>, with MCP_ARGS as the tool's arguments in JSON. By default it asks DeepWiki for the wiki structure of a public repository:

kubectl -n support patch configmap faq-agent --type merge -p '{"data":{"MCP_TOOL":"deepwiki__read_wiki_structure","MCP_ARGS":"{\"repoName\":\"facebook/react\"}"}}'
kubectl -n support rollout restart deploy/faq-agent

The agent logs MCP deepwiki__read_wiki_structure ALLOWED with the start of the answer once the call goes through. Keep the tool set before step 6, so the agent learns it before you switch to Enforce.

Give the agent its own ServiceAccount, as the sample does. Registration binds that ServiceAccount, and every Pod that runs as it shares the agent's access.

4. See the agent discovered​

If you're still in the setup wizard, its Protect your first agent step shows the agent as soon as it's discovered, with a Register it button that opens step 5's registration wizard. See Protect your first agent in the wizard.

The agent's first call puts it on Inventory → Workloads as Discovered. Open its row: Why this workload is listed shows the evidence, for example Called openrouter/nvidia/nemotron-3.5-lightning:free through the gateway at a timestamp.

What the agent sees depends on the unregistered workload mode:

  • Enforce, the default for a new installation: the call is blocked, and the Dashboard shows Blocked an unregistered agent. The agent logs:

    LLM  nvidia/nemotron-3.5-lightning:free  BLOCKED (403)  Blocked by AIControls: workload_not_registered: workload support/Deployment/faq-agent is not registered with AIControls; …

  • Audit: the call goes through, the agent logs ALLOWED and the answer, and the Dashboard shows An unregistered agent is calling the gateway.

5. Register it in one step​

  1. On the Dashboard callout, select Register it. Or open the workload's row menu on Inventory → Workloads and select Register….
  2. On Blueprint, keep the Default Workload Blueprint. Its summary card shows its Purpose and the profile the agent gets: it calls an LLM through the AIControls gateway and nothing else. If the agent does more, the step says what it observed. Select Next.
  3. On Identity, check the ServiceAccount: it is the one discovery observed, k8s://support/faq-agent. Select Next.
  4. On Name, name the agent, then select Next. The agent takes its Purpose from the Blueprint.
  5. On Mode, keep Learn first (Audit), then select Next.
  6. On Review, select Register. AIControls registers the workload, binds its identity, and attaches the default Blueprint in one transaction, and opens the workload's AI access tab.

The agent's next calls are attributed to the workload in Audit mode. To narrow or widen a tier for this agent only, select Customize for this workload on the Blueprint or Review step. For every field on the way, see Register a Workload.

6. Learn its access, then promote it​

  1. Let the agent run. On the AI access tab, each call that no Access Binding grants becomes a recommendation within a few minutes.
  2. Review the recommendations, then select Promote all and confirm. AIControls adds every open recommendation to an Access Binding that only this workload uses. See Promote a recommendation.

7. Switch to Enforce​

  1. On the Mode card, select Enforce, then Switch to Enforce.
  2. AIControls previews the unmatched calls from the last 7 days that Enforce would now block. Review them and confirm. If the card says Enforce is not available yet, the workload has no effective grant: promote a recommendation first. See Switch a workload to Enforce.
  3. In Enforce, the calls you promoted are allowed, and any other call is denied. On a new installation, the workload unmatched action is Deny, so you don't need to change the default access policy, which still allows unmatched calls from developers and other callers that aren't registered workloads.

  4. Make a call the workload never made, such as another free model, and see it blocked with accessdefault:llm:workload:

    kubectl -n support patch configmap faq-agent --type merge -p '{"data":{"MODEL":"google/gemma-4-31b-it:free"}}'
    kubectl -n support rollout restart deploy/faq-agent
    kubectl -n support logs deploy/faq-agent -f

    The agent logs LLM google/gemma-4-31b-it:free BLOCKED with the reason, and the blocked call shows on Security → Activity. With Anthropic, use another Claude model, such as claude-haiku-4-5. To go back, set MODEL to the model the workload learned and restart the Deployment again.

    With DeepWiki, call a tool the workload never called, such as ask_question:

    kubectl -n support patch configmap faq-agent --type merge -p '{"data":{"MCP_TOOL":"deepwiki__ask_question","MCP_ARGS":"{\"repoName\":\"facebook/react\",\"question\":\"What is a hook?\"}"}}'
    kubectl -n support rollout restart deploy/faq-agent

    The agent logs MCP deepwiki__ask_question BLOCKED with accessdefault:mcp:workload, while read_wiki_structure, which it learned, stays allowed. Set MCP_TOOL back to deepwiki__read_wiki_structure to go back.

    If the call goes through, check Settings → General → Default access policy → Workload callers: unmatched calls: an installation upgraded from an earlier release keeps Use the MCP and LLM actions above until you select Deny.

With runtime controls on a node with the open/exec hooks, also enable enforcement for the workload's registered behavior. Then kubectl exec into the agent, or a file read outside its declared files tier, is blocked by the kernel. See kubectl exec is blocked by default.

8. Reuse its access for similar workloads​

  1. On the AI access tab, select Create Blueprint from this workload. AIControls fills a new Workload Blueprint from the workload's profile, and you select or create its Access Binding and Runtime Binding.
  2. Register similar workloads, in the same namespace or with the same image, against the new Blueprint in Audit, and switch them to Enforce when they look right.

See Turn a workload's learned access into a Blueprint.

Last updated on