AI gateways: governing model and agent traffic

#The 60-second version

Once several teams build AI assistants and agents, calls to models and tools start leaving from many places. Each team picks its own model, stores its own keys, logs in its own way and has its own idea of a limit.

An AI gateway is one checkpoint that all of that traffic passes through. For every call, it answers six questions:

• Who is calling? A known agent, for a known person.
• May they do this? This model, this tool, for this user.
• How much? Requests per minute, tokens per day, money per month.
• Is the content acceptable? Personal data masked, oversized or harmful input refused.
• Where does it go? Which model or system answers.
• What happened? One audit record, whatever the answer.

It is the same idea as an API gateway, applied to models, tools and agents. It does not replace SAP's own authorization checks. It adds a layer in front of them that every team shares.

#Why it matters to the business

Example. At Calder Pumpworks, a fictional pump maker used throughout this course, accounts payable runs an invoice triage agent. Planning runs a supply watch agent that checks MRP exceptions and purchase order dates. Each team set up its own model account. Then three things happen in one month:

• The planning agent gets stuck in a loop over a weekend and asks the same question thousands of times. The bill arrives a month later.
• An auditor asks which agent read supplier bank data in October, for whom. Two of the three teams can't answer.
• Security finds a model key pasted in a project wiki. Nobody knows which agents use it, so nobody can rotate it safely.

A gateway addresses each one: a rate limit and a daily budget stop the loop within a minute; one audit log answers the auditor; keys to models live only in the gateway, so rotating one touches one place.

The value is in four areas:

• Cost control. OWASP lists unbounded consumption as a top risk for AI applications: runaway use that becomes a denial of service or a "denial of wallet". Budgets and limits are the fix.
• Security. One place to enforce who may call which model and tool, and to mask data before it leaves.
• Audit. One record format for every call, kept for as long as your audit rules require.
• Speed. Teams reuse the gateway's controls instead of rebuilding them. Swapping a model becomes a gateway change, not a project.

The cost is a new piece of shared infrastructure. It needs an owner, high availability and fast security patching, because every AI call now depends on it.

#How SAP does it

SAP announced several gateways in 2026. They cover different traffic, and they are at different stages. As of 3 November 2026:

• Agent Gateway (inbound agent traffic). Outside agents call SAP's Joule agents through it, using the open Agent2Agent (A2A) protocol. SAP's Architecture Center announced general availability of inbound A2A on 27 October 2026. Each outside client is registered and gets an access list of the SAP agents it may call. The SAP agent then acts within the signed-in user's permissions. Calls from SAP agents out to other agents or MCP servers are described as future direction.
• MCP Gateway in SAP Integration Suite (agents calling tools). It turns APIs into tools agents can use, with rate limiting, sign-in and authorization, payload protection and logs, per SAP's July 2026 Integration Suite update.
• SAP Enterprise AI Gateway (all AI traffic). Announced at SAP TechEd on 27 October 2026. SAP's release says it will consolidate authentication, policy enforcement, routing, monitoring, usage tracking and auditing of AI traffic. A follow-up article says it builds on SAP Integration Suite and covers MCP servers, APIs, agent-to-agent connections and large language models. Neither page gives an availability date or price. Treat it as announced, not available.
• Orchestration service in generative AI hub (an app's own model calls). Part of SAP AI Core. It gives one API across models, with content filtering, data masking and grounding for each call, as described on SAP Learning. It is a per-call pipeline for one application, not an enterprise-wide budget and policy point.
• Joule Studio admin controls. SAP announced observability and administrative controls for agent behavior, model usage and spend in Joule Studio on 27 October 2026. They cover agents built in Joule Studio.

What this means today: SAP covers inbound agent traffic and tool access, and has announced the enterprise-wide piece. For your own agents' model traffic and budgets across teams, you still decide how to govern it.

#Which gateway covers which traffic

The column on the right is the design decision. Write down which traffic you route where, and revisit it when SAP publishes availability.

#Questions to ask

1. How many teams call models or agents today, and how many separate model keys exist? Who owns each?
2. Which of our AI calls go through a shared checkpoint, and which go straight to a provider?
3. Can we answer, from one log: which agent called which model or tool, for which person, and what it cost?
4. What stops a looping agent from spending a month's budget in a weekend?
5. Which SAP gateway covers each kind of traffic we have, and which of those are generally available in our region and edition?
6. If we build or buy a gateway now, how do we move to SAP Enterprise AI Gateway later without rewriting every agent?
7. Who patches the gateway, and how fast? It sees every key and every prompt.
8. What happens when the gateway is down: do calls fail, or bypass it?

#Common misconceptions

• "The gateway replaces SAP authorizations." It adds a layer in front. SAP still checks the user's roles on every API call it receives.
• "Our model provider's dashboard is enough." It shows spend per account, not per agent, team or person, and it can't stop a call to a different provider.
• "A gateway makes agents safe." It limits what they can reach and how much. Prompt injection, wrong answers and approval steps still need their own design.
• "If the gateway is down, let calls through." That is called failing open. It turns an outage into a gap in security and audit. Gateways should fail closed.
• "SAP Enterprise AI Gateway is available now." As of 3 November 2026, SAP's pages describe it as announced, with no date. Check before you plan around it.
• "One gateway means one product." Agent, tool and model traffic may pass through different gateways. What matters is that each path has the same controls and one shared log.

#Key terms

All course terms, A to Z, in the glossary

• AI gateway: a checkpoint in front of models, tools and agents that authenticates, authorizes, limits, meters, filters, routes and logs every call.
• Agent Gateway: SAP's governed entry point for agents, used today for outside agents calling Joule agents through A2A.
• MCP Gateway: the part of SAP Integration Suite that exposes APIs as MCP tools for agents, with central policies.
• Token budget: the number of model tokens a team, agent or user may use in a period, enforced before a call goes out.
• Rate limit: the most requests a caller may send in a time window.
• Model routing: choosing which model answers a call, so apps ask for "fast" or "strong" rather than a provider's model name.
• Fail closed: refusing calls when a control can't run, instead of letting them through.
• Unbounded consumption: OWASP's name for uncontrolled model use that causes outages or runaway cost.

#Check yourself

#Mental model: one door, six checks

Think of a building with many offices and many visitors. You can put a guard in every office, each with their own rules and notebook. Or you can put one front desk at the door: it checks the badge, checks which floors the visitor may enter, counts visits, scans bags and writes one visitor log. Offices still lock their own doors.

An AI gateway is the front desk. Agents and apps are visitors; models, tools and other agents are offices. SAP's authorization checks are the office locks: the gateway never replaces them, and they never see the whole building.

Every call runs the same checks in the same order, and the first "no" ends it:

flowchart LR
  A[Agent or app] --> G1{Who?<br/>authenticate}
  G1 --> G2{Allowed?<br/>model, tool, user}
  G2 --> G3{How much?<br/>rate, budget}
  G3 --> G4[Filter<br/>mask, size]
  G4 --> R[Route to model,<br/>tool or agent]
  R --> M[Meter tokens<br/>and cost]
  G1 -. no .-> L[(Audit log)]
  G2 -. no .-> L
  G3 -. no .-> L
  M --> L

Two rules make it work. Every path goes through it: an agent that can reach a model directly makes the gateway optional. And it fails closed: if a check can't run, the call is refused.

#How it works

#What a gateway sits in front of

Agent traffic has three kinds of call, and a gateway can govern all three:

The checks are the same in each column. That is why one log format and one policy model pay off.

#Authentication: who is calling

The gateway first decides which agent sent the call. Lab setups use API keys; production uses tokens from an identity provider, such as OAuth client credentials for the agent and a delegated token naming the user too. Agent permissions and SAP authorizations covers delegated tokens and why the agent's and the user's permissions must intersect.

Two rules from the security guidance for the Model Context Protocol apply to any gateway. A server must not accept tokens that weren't issued for it, and must not pass a caller's token on unchanged. Token passthrough lets clients bypass rate limits and monitoring and leaves logs that name the wrong caller. The gateway checks the token's audience, then uses its own credentials or a new token toward the backend.

#Authorization: what the caller may do

Two allow-lists decide, and both must say yes:

• The agent's list: which models and tools it is registered for. The invoice triage agent has no business changing purchase order dates, even if a poisoned tool description tells it to.
• The user's list: what this person may let an agent do for them. A buyer may not release an invoice for payment through the triage agent, even though the agent itself may draft one for an AP accountant.

This is the same intersection SAP's Architecture Center describes for its Agent Gateway. The gateway's lists are coarse: tool names and models. SAP's authorization objects stay the fine-grained check on data, behind the gateway.

#Limits: how much

• Rate limits cap requests per caller per window, here per agent per rolling minute. They stop bursts and loops.
• Token budgets cap total model use per team, agent or user per day or month. The gateway estimates the tokens a call can use before forwarding it: the input plus the maximum output the caller allows. If that reservation would exceed the budget, the call is refused. After the call, it meters what was really used.
• Size limits refuse oversized input before anything else is spent on it.

OWASP's guidance for unbounded consumption lists the same set: input size limits, rate limits and quotas, timeouts and throttling, and monitoring for unusual use.

#Filtering and routing

Filtering at the gateway is a coarse safety net: mask obvious identifiers such as bank account numbers, refuse oversized input, block disallowed content. Fine-grained filtering and masking for one application belongs nearer the call, for example in SAP's orchestration service or the checks in Guardrails and content filtering.

Routing maps an alias to a model. Apps ask for fast or strong; the gateway decides which deployment answers and where to fall back. Changing models becomes a configuration change with a test run, not a change in every app.

#The audit log

One record per decision, allowed or denied, with: time, request ID, agent, user, team, model or tool, decision, reason, status code, tokens, cost and a fingerprint of the input. Not the prompt text by default: prompts carry business and personal data, and the log would become the most sensitive store you own. If you must keep prompts, keep them in a separate store with its own access rules and retention.

Make the log tamper-evident. A simple way is a hash chain: each record includes the hash of the one before, so a changed, deleted or reordered record breaks the chain. Production systems use write-once storage or a managed audit service as well.

#The gateway is now a target

A gateway sees every key and every prompt, so attackers aim for it. MCP security in 2026 walks through real 2026 vulnerabilities in an open-source AI gateway, including one where a failed sign-in path let requests through. Patch the gateway as fast as you patch SAP, keep its admin interface off the internet, and test that every check fails closed.

#Build it yourself: a small AI gateway

You will build unit11/ai_gateway.py, a local gateway that every call from two agents passes through. It checks the agent's key, the size of the input, the model or tool against the agent's and the user's allow-lists, a rate limit and a team token budget. It masks bank account numbers, forwards the call to a fake model or a fake SAP tool, meters tokens and cost, and writes a tamper-evident audit log. A test matrix proves the policy holds after every change.

Before you start: complete Set up your computer for this course and Set up for Unit 11. This lab uses only modules that come with Python.

flowchart LR
  S[Sample calls<br/>from two agents] --> G[ai_gateway.py<br/>six checks]
  G -->|allowed| B[Fake model or<br/>fake SAP tool]
  G --> L[(gateway_audit.jsonl)]
  L --> R[report: spend<br/>and denials]
  L --> V[verify: chain<br/>unbroken?]
  G --> T[test: policy<br/>matrix]

#What you need

• Your course folder orchestrate-course with its .venv and a unit11 folder.
• About 35 minutes.
• Cost: free. No account, no key, no model call, no SAP system.

#Step 1: Open your course folder and turn on the virtual environment

1. Open VS Code, choose File > Open Folder, and open orchestrate-course.

2. Open a terminal: Terminal > New Terminal.

3. If the prompt doesn't start with (.venv), turn it on:

   • Windows (PowerShell):

     Copy

     .venv\Scripts\Activate.ps1

   • macOS / Linux:

     Copy

     source .venv/bin/activate

4. Check Python:

   Copy

   python --version

   You should see Python 3.10 or newer. On macOS or Linux without .venv, use python3 in every command.

Run every command in this topic from the course folder, not from inside unit11.

#Step 2: Keep the audit log out of Git

1. Open .gitignore in your course folder.

2. Add this line at the end and save:

   Copy

   unit11/gateway_audit.jsonl

   The log names people and agents. In a real system it is audit data with its own retention rules, not code.

#Step 3: Create the gateway

1. In VS Code's file list, right-click unit11, choose New File, and name it ai_gateway.py.
2. Paste the code below and save.

"""Unit 11: a small AI gateway for model and agent traffic.

Every call from an agent, to a model or to a tool, goes through one function:
gateway(). It checks, in this order, and stops at the first "no":

  1. authenticate   the API key belongs to a registered agent
  2. size           the input is not too large
  3. route          the model alias is allowed for the agent's team
     or authorize   the tool is allowed for BOTH the user and the agent
  4. rate limit     the agent has not sent too many requests this minute
  5. budget         the team's daily token budget can cover this call
  6. filter         IBAN-like strings are masked before the call leaves
  then forwards the call to a fake model or a fake tool, meters the
  tokens and cost, and writes one audit record per decision.

Everything is made up: Calder Pumpworks is a fictional company, the
models and tools are local fakes, and no network call is made.

Usage:
  python unit11/ai_gateway.py --sample     run the sample calls, write the audit log
  python unit11/ai_gateway.py report       spend and denials by team and agent
  python unit11/ai_gateway.py verify       check that the audit log is unbroken
  python unit11/ai_gateway.py test         run the policy test matrix (exit 1 on failure)
"""

import argparse
import hashlib
import json
import math
import re
import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path

AUDIT_LOG = Path("unit11/gateway_audit.jsonl")

# --- 1. Policy: who may do what. In production this lives in a config store, not code ---

# Agents authenticate with an API key. The gateway stores only the key's SHA-256 hash.
def key_hash(key):
    return hashlib.sha256(key.encode()).hexdigest()

AGENTS = {
    key_hash("calder-lab-key-invoice-triage"): {
        "agent": "AGT_INV_TRIAGE", "team": "accounts-payable",
        "tools": {"supplier_invoice_get", "po_get", "invoice_release_draft"},
    },
    key_hash("calder-lab-key-supply-watch"): {
        "agent": "AGT_SUPPLY_WATCH", "team": "planning",
        "tools": {"mrp_exceptions_get", "po_get", "po_date_change_draft"},
    },
}

# What each person may let an agent do for them (the user's side of the intersection).
USERS = {
    "00100241": {"role": "AP accountant", "tools": {"supplier_invoice_get", "po_get", "invoice_release_draft"}},
    "00100211": {"role": "buyer", "tools": {"po_get", "po_date_change_draft", "supplier_invoice_get"}},
    "00100153": {"role": "planner", "tools": {"mrp_exceptions_get", "po_get", "po_date_change_draft"}},
}

# Model aliases: apps ask for "fast" or "strong", the gateway decides the real model.
# Prices are course values in USD per 1,000 tokens, not any provider's price list.
MODELS = {
    "fast":   {"backend": "mock-small", "usd_per_1k_in": 0.0005, "usd_per_1k_out": 0.0015},
    "strong": {"backend": "mock-large", "usd_per_1k_in": 0.0050, "usd_per_1k_out": 0.0150},
}

TEAMS = {
    "accounts-payable": {"models": {"fast"}, "daily_tokens": 20000},
    "planning":         {"models": {"fast", "strong"}, "daily_tokens": 4000},
}

RATE_LIMIT = 5             # requests per agent per rolling 60 seconds
MAX_INPUT_CHARS = 8000     # larger inputs are refused before anything else runs
DEFAULT_MAX_TOKENS = 300   # output cap when the caller doesn't set one
IBAN_LIKE = re.compile(r"\b[A-Z]{2}\d{2}(?:\s?[A-Z0-9]{4}){3,7}(?:\s?[A-Z0-9]{1,4})?\b")

# --- 2. Fake backends: a model and some SAP-shaped tools ---

def estimate_tokens(text):
    """Rough rule of thumb: about 4 characters per token. Real gateways use the model's tokenizer."""
    return max(1, math.ceil(len(text) / 4))

def fake_model(backend, prompt, max_tokens):
    out = min(max_tokens, 40 + len(prompt) // 20)
    return {"text": f"[{backend}] answer drafted ({out} tokens)", "tokens_out": out}

FAKE_SAP = {
    "supplier_invoice_get": lambda a: {"invoice": a.get("invoice"), "status": "price block", "amount": "965.00 USD per unit"},
    "po_get": lambda a: {"po": a.get("po"), "supplier": "20000001", "price": "905.00 USD"},
    "mrp_exceptions_get": lambda a: {"plant": a.get("plant"), "exceptions": 4},
    "invoice_release_draft": lambda a: {"draft": "D-" + str(a.get("invoice")), "note": "waits for a person to approve"},
    "po_date_change_draft": lambda a: {"draft": "D-" + str(a.get("po")), "note": "waits for a person to approve"},
}

# --- 3. Audit log: one JSON line per decision, chained by hashes so edits show ---

class Audit:
    def __init__(self, path):
        self.path = path
        self.records = []
        if path and path.exists():
            self.records = [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line.strip()]

    def last_hash(self):
        return self.records[-1]["hash"] if self.records else "0" * 64

    def write(self, record):
        record["prev"] = self.last_hash()
        body = json.dumps(record, sort_keys=True)
        record["hash"] = hashlib.sha256(body.encode()).hexdigest()
        self.records.append(record)
        if self.path:
            with self.path.open("a", encoding="utf-8") as f:
                f.write(json.dumps(record, sort_keys=True) + "\n")

    def tokens_used(self, team, day):
        return sum(r["tokens_in"] + r["tokens_out"] for r in self.records
                   if r["team"] == team and r["decision"] == "allow" and r["ts"][:10] == day)

    def recent_requests(self, agent, now):
        start = now - timedelta(seconds=60)
        return sum(1 for r in self.records
                   if r["agent"] == agent and r["reason"] != "rate limit"
                   and start < datetime.fromisoformat(r["ts"]) <= now)

# --- 4. The gateway ---

def gateway(audit, call):
    """Check one call against the policy, forward it if allowed, audit it either way."""
    now = datetime.fromisoformat(call["ts"])
    text = call.get("prompt", "") + json.dumps(call.get("args", {}))
    record = {
        "ts": call["ts"], "request_id": call["id"], "kind": call["kind"],
        "target": call.get("model") or call.get("tool"), "user": call.get("user"),
        "agent": None, "team": None, "decision": "deny", "status": 0, "reason": "",
        "tokens_in": 0, "tokens_out": 0, "cost_usd": 0.0, "masked": 0,
        # The audit log keeps a fingerprint of the input, never the text itself.
        "input_sha256": hashlib.sha256(text.encode()).hexdigest()[:16],
    }

    def deny(status, reason):
        record.update(status=status, reason=reason)
        audit.write(record)
        return record, None

    # 1. Authenticate: an unknown key is refused. Fail closed: no key, no call.
    agent = AGENTS.get(key_hash(call.get("key", "")))
    if not agent:
        return deny(401, "unknown key")
    record.update(agent=agent["agent"], team=agent["team"])
    team = TEAMS[agent["team"]]

    # 2. Size: refuse oversized input before spending anything on it.
    if len(text) > MAX_INPUT_CHARS:
        return deny(413, f"input over {MAX_INPUT_CHARS} characters")

    # 3. Route a model call, or authorize a tool call for user AND agent.
    if call["kind"] == "model":
        if call["model"] not in team["models"]:
            return deny(403, f"model '{call['model']}' not allowed for team {agent['team']}")
    else:
        user = USERS.get(call.get("user"))
        if not user:
            return deny(403, "unknown user")
        if call["tool"] not in agent["tools"]:
            return deny(403, f"tool not registered for {agent['agent']}")
        if call["tool"] not in user["tools"]:
            return deny(403, f"tool not allowed for user {call['user']}")

    # 4. Rate limit per agent, over a rolling minute.
    if audit.recent_requests(agent["agent"], now) >= RATE_LIMIT:
        return deny(429, "rate limit")

    # 5. Budget: reserve the worst case (input plus maximum output) before forwarding.
    tokens_in = estimate_tokens(text)
    max_out = call.get("max_tokens", DEFAULT_MAX_TOKENS) if call["kind"] == "model" else 0
    used = audit.tokens_used(agent["team"], call["ts"][:10])
    if call["kind"] == "model" and used + tokens_in + max_out > team["daily_tokens"]:
        return deny(402, f"team budget: {used} of {team['daily_tokens']} tokens used today")

    # 6. Filter: mask IBAN-like strings before the text leaves the gateway.
    prompt, masked = IBAN_LIKE.subn("[IBAN]", call.get("prompt", ""))
    record["masked"] = masked

    # Forward, then meter what was really used.
    if call["kind"] == "model":
        m = MODELS[call["model"]]
        result = fake_model(m["backend"], prompt, max_out)
        record["tokens_in"], record["tokens_out"] = tokens_in, result["tokens_out"]
        record["cost_usd"] = round(tokens_in / 1000 * m["usd_per_1k_in"]
                                   + result["tokens_out"] / 1000 * m["usd_per_1k_out"], 6)
    else:
        result = FAKE_SAP[call["tool"]](call.get("args", {}))
    record.update(decision="allow", status=200, reason="ok")
    audit.write(record)
    return record, result

# --- 5. Sample traffic: one morning at Calder (fictional) ---

KEY_AP = "calder-lab-key-invoice-triage"
KEY_PL = "calder-lab-key-supply-watch"
START = datetime(2026, 11, 3, 9, 0, 0, tzinfo=timezone.utc)

def at(seconds):
    return (START + timedelta(seconds=seconds)).isoformat()

def sample_calls():
    calls = [
        dict(key=KEY_AP, kind="model", model="fast", user="00100241",
             prompt="Summarize why invoice 5105600003 is blocked: price 965.00 vs PO 905.00."),
        dict(key=KEY_AP, kind="tool", tool="supplier_invoice_get", user="00100241", args={"invoice": "5105600003"}),
        dict(key=KEY_AP, kind="tool", tool="invoice_release_draft", user="00100241", args={"invoice": "5105600003"}),
        dict(key=KEY_AP, kind="tool", tool="invoice_release_draft", user="00100211", args={"invoice": "5105600003"}),
        dict(key=KEY_AP, kind="tool", tool="po_date_change_draft", user="00100241", args={"po": "4500000002"}),
        dict(key=KEY_AP, kind="model", model="strong", user="00100241", prompt="Explain the three-way match."),
        dict(key="stolen-or-made-up-key", kind="model", model="fast", user="00100241", prompt="List all suppliers."),
        dict(key=KEY_AP, kind="model", model="fast", user="00100241",
             prompt="Draft a note to Orrin: pay to DE89 3704 0044 0532 0130 00 after the price is agreed."),
        dict(key=KEY_PL, kind="tool", tool="mrp_exceptions_get", user="00100153", args={"plant": "CP11"}),
        dict(key=KEY_PL, kind="model", model="strong", user="00100153", max_tokens=400,
             prompt="Which MRP exceptions at CP11 put order 1100019 at risk?"),
    ]
    # A planning agent caught in a loop: the same question six times in six seconds.
    for _ in range(6):
        calls.append(dict(key=KEY_PL, kind="model", model="fast", user="00100153",
                          prompt="Check PO 4500000010 again."))
    calls += [
        dict(key=KEY_PL, kind="model", model="strong", user="00100153", prompt="x" * 9000),
        dict(key=KEY_PL, kind="model", model="strong", user="00100153", max_tokens=2000,
             prompt="Re-plan every ES-200 order for November. " * 120),
        dict(key=KEY_PL, kind="model", model="strong", user="00100153", max_tokens=2000,
             prompt="Re-plan every ES-300 order for November. " * 120),
    ]
    out, t = [], 0
    for i, c in enumerate(calls, start=1):
        if 11 <= i <= 16:
            t += 1      # the loop fires once a second
        elif i >= 17:
            t += 61     # the last calls each come a minute apart
        else:
            t += 20
        out.append({**c, "id": f"R{i:03d}", "ts": at(t)})
    return out

# --- 6. Commands ---

def cmd_sample():
    AUDIT_LOG.parent.mkdir(exist_ok=True)
    if AUDIT_LOG.exists():
        AUDIT_LOG.unlink()          # each sample run starts a fresh day
    audit = Audit(AUDIT_LOG)
    print(f"{'id':5} {'agent':17} {'user':9} {'target':22} {'result':6} {'code':4} reason")
    for call in sample_calls():
        rec, _ = gateway(audit, call)
        print(f"{rec['request_id']:5} {str(rec['agent']):17} {str(rec['user']):9} "
              f"{rec['target']:22} {rec['decision']:6} {rec['status']:<4} {rec['reason']}"
              + (f" ({rec['masked']} masked)" if rec["masked"] else ""))
    print(f"\nWrote {len(audit.records)} audit records to {AUDIT_LOG}")

def cmd_report():
    audit = Audit(AUDIT_LOG)
    if not audit.records:
        print(f"No audit records in {AUDIT_LOG}. Run with --sample first.")
        return
    teams = {}
    for r in audit.records:
        key = (r["team"] or "(unauthenticated)", r["agent"] or "-")
        t = teams.setdefault(key, {"allow": 0, "deny": 0, "tokens": 0, "cost": 0.0})
        t[r["decision"]] += 1
        t["tokens"] += r["tokens_in"] + r["tokens_out"]
        t["cost"] += r["cost_usd"]
    print(f"{'team':18} {'agent':17} {'allowed':>7} {'denied':>6} {'tokens':>7} {'cost USD':>9}")
    for (team, agent), t in sorted(teams.items()):
        print(f"{team:18} {agent:17} {t['allow']:>7} {t['deny']:>6} {t['tokens']:>7} {t['cost']:>9.4f}")
    print("\nDenials by reason:")
    reasons = {}
    for r in audit.records:
        if r["decision"] == "deny":
            label = r["reason"].split(":")[0]
            reasons[label] = reasons.get(label, 0) + 1
    for reason, n in sorted(reasons.items(), key=lambda x: -x[1]):
        print(f"  {n}  {reason}")

def cmd_verify():
    audit = Audit(AUDIT_LOG)
    prev = "0" * 64
    for i, r in enumerate(audit.records, start=1):
        body = {k: v for k, v in r.items() if k != "hash"}
        if r["prev"] != prev or hashlib.sha256(json.dumps(body, sort_keys=True).encode()).hexdigest() != r["hash"]:
            print(f"BROKEN at line {i} ({r.get('request_id')}): the record was changed, removed or reordered")
            return 1
        prev = r["hash"]
    print(f"OK: {len(audit.records)} records, chain unbroken")
    return 0

# Expected decisions for the policy. Run after every policy change.
MATRIX = [
    ("AP agent, AP user, invoice read", dict(key=KEY_AP, kind="tool", tool="supplier_invoice_get", user="00100241", args={}), "allow"),
    ("AP agent, buyer, release draft", dict(key=KEY_AP, kind="tool", tool="invoice_release_draft", user="00100211", args={}), "deny"),
    ("AP agent, AP user, PO date change", dict(key=KEY_AP, kind="tool", tool="po_date_change_draft", user="00100241", args={}), "deny"),
    ("Planning agent, planner, PO date change", dict(key=KEY_PL, kind="tool", tool="po_date_change_draft", user="00100153", args={}), "allow"),
    ("Planning agent, AP user, invoice read", dict(key=KEY_PL, kind="tool", tool="supplier_invoice_get", user="00100241", args={}), "deny"),
    ("AP agent, strong model", dict(key=KEY_AP, kind="model", model="strong", user="00100241", prompt="hi"), "deny"),
    ("Planning agent, strong model", dict(key=KEY_PL, kind="model", model="strong", user="00100153", prompt="hi"), "allow"),
    ("Unknown key", dict(key="", kind="model", model="fast", user="00100241", prompt="hi"), "deny"),
    ("Unknown user", dict(key=KEY_PL, kind="tool", tool="po_get", user="99999999", args={}), "deny"),
    ("Oversized input", dict(key=KEY_AP, kind="model", model="fast", user="00100241", prompt="x" * 9000), "deny"),
]

def cmd_test():
    failures = 0
    for i, (name, call, expected) in enumerate(MATRIX, start=1):
        audit = Audit(None)                  # each case on a fresh, in-memory log
        rec, _ = gateway(audit, {**call, "id": f"T{i:02d}", "ts": at(i)})
        ok = rec["decision"] == expected
        failures += not ok
        print(f"{'PASS' if ok else 'FAIL'}  {name:42} expected {expected:5} got {rec['decision']:5} {rec['reason']}")
    print(f"\n{len(MATRIX) - failures} of {len(MATRIX)} passed")
    return 1 if failures else 0

def main():
    p = argparse.ArgumentParser(description="A small AI gateway for model and agent traffic (Unit 11).")
    p.add_argument("command", nargs="?", choices=["report", "verify", "test"])
    p.add_argument("--sample", action="store_true", help="run the sample calls and write a fresh audit log")
    a = p.parse_args()
    if a.sample:
        cmd_sample()
    elif a.command == "report":
        cmd_report()
    elif a.command == "verify":
        sys.exit(cmd_verify())
    elif a.command == "test":
        sys.exit(cmd_test())
    else:
        p.print_help()

if __name__ == "__main__":
    main()

#Step 4: Run the sample traffic

1. Run:

   Copy

   python unit11/ai_gateway.py --sample

2. You should see exactly this:

id    agent             user      target                 result code reason
R001  AGT_INV_TRIAGE    00100241  fast                   allow  200  ok
R002  AGT_INV_TRIAGE    00100241  supplier_invoice_get   allow  200  ok
R003  AGT_INV_TRIAGE    00100241  invoice_release_draft  allow  200  ok
R004  AGT_INV_TRIAGE    00100211  invoice_release_draft  deny   403  tool not allowed for user 00100211
R005  AGT_INV_TRIAGE    00100241  po_date_change_draft   deny   403  tool not registered for AGT_INV_TRIAGE
R006  AGT_INV_TRIAGE    00100241  strong                 deny   403  model 'strong' not allowed for team accounts-payable
R007  None              00100241  fast                   deny   401  unknown key
R008  AGT_INV_TRIAGE    00100241  fast                   allow  200  ok (1 masked)
R009  AGT_SUPPLY_WATCH  00100153  mrp_exceptions_get     allow  200  ok
R010  AGT_SUPPLY_WATCH  00100153  strong                 allow  200  ok
R011  AGT_SUPPLY_WATCH  00100153  fast                   allow  200  ok
R012  AGT_SUPPLY_WATCH  00100153  fast                   allow  200  ok
R013  AGT_SUPPLY_WATCH  00100153  fast                   allow  200  ok
R014  AGT_SUPPLY_WATCH  00100153  fast                   deny   429  rate limit
R015  AGT_SUPPLY_WATCH  00100153  fast                   deny   429  rate limit
R016  AGT_SUPPLY_WATCH  00100153  fast                   deny   429  rate limit
R017  AGT_SUPPLY_WATCH  00100153  strong                 deny   413  input over 8000 characters
R018  AGT_SUPPLY_WATCH  00100153  strong                 allow  200  ok
R019  AGT_SUPPLY_WATCH  00100153  strong                 deny   402  team budget: 1718 of 4000 tokens used today

Wrote 19 audit records to unit11/gateway_audit.jsonl

Read the rows in groups:

• R001 to R003: the invoice triage agent works for an AP accountant (00100241): a model call, an invoice read and a release draft. Drafts wait for a person, as in SAP tools for agents.
• R004: the same agent, for a buyer (00100211). The agent may draft releases; the buyer may not have one drafted. The user's list says no.
• R005: the triage agent tries to change a purchase order date. That tool isn't registered for it. This is what stops a poisoned tool description from widening what an agent can do.
• R006: accounts payable may only use the fast model alias.
• R007: an unknown key: refused before anything else runs.
• R008: allowed, but the gateway masked one bank account number in the prompt before it left.
• R011 to R016: the supply watch agent loops. With R009 and R010, five calls sit inside one minute, so R014 to R016 are refused with 429.
• R017: 9,000 characters of input, refused before any tokens are counted.
• R018 and R019: two large re-planning prompts. R018 fits. For R019 the gateway reserves the input (1,230 tokens) plus the caller's maximum output (2,000) on top of the 1,718 already used. That passes the 4,000-token team budget, so it is refused before the model is called.

#Step 5: Read the spend report

1. Run:

   Copy

   python unit11/ai_gateway.py report

2. You should see:

team               agent             allowed denied  tokens  cost USD
(unauthenticated)  -                       0      1       0    0.0000
accounts-payable   AGT_INV_TRIAGE          4      3     127    0.0001
planning           AGT_SUPPLY_WATCH        6      5    1718    0.0113

Denials by reason:
  3  rate limit
  1  tool not allowed for user 00100211
  1  tool not registered for AGT_INV_TRIAGE
  1  model 'strong' not allowed for team accounts-payable
  1  unknown key
  1  input over 8000 characters
  1  team budget

This is the view a budget owner and an auditor need: per team and agent, what was allowed, what was refused and why, and what it cost. Your model provider can't produce it, because it doesn't know your teams, agents or users.

#Step 6: Look at one audit record

1. In VS Code, open unit11/gateway_audit.jsonl. Each line is one decision. The first line looks like this:

{"agent": "AGT_INV_TRIAGE", "cost_usd": 7.4e-05, "decision": "allow", "hash": "2dfaaa0e0b6bc546e6b59067a804bf4c845ed6b71badf976d52a953115886037", "input_sha256": "6341c0110a6e9364", "kind": "model", "masked": 0, "prev": "0000000000000000000000000000000000000000000000000000000000000000", "reason": "ok", "request_id": "R001", "status": 200, "target": "fast", "team": "accounts-payable", "tokens_in": 19, "tokens_out": 43, "ts": "2026-11-03T09:00:20+00:00", "user": "00100241"}

2. Notice what is there: who, what, decision, reason, tokens, cost, and input_sha256, a short fingerprint of the input. Notice what isn't: the prompt itself.
3. prev is the hash of the record before; hash covers this record including prev. That links the lines into a chain.

#Step 7: Prove the log is tamper-evident

1. Run:

   Copy

   python unit11/ai_gateway.py verify

   You should see:

   Copy

   OK: 19 records, chain unbroken

2. In unit11/gateway_audit.jsonl, on the third line, change "user": "00100241" to "user": "00100211". Save.

3. Run verify again. You should see:

   Copy

   BROKEN at line 3 (R003): the record was changed, removed or reordered

4. Restore a clean log by running the sample again:

   Copy

   python unit11/ai_gateway.py --sample

A hash chain shows that something changed. It doesn't stop someone who can rewrite the whole file from rebuilding the chain. In production, write the log to storage that can't be edited and keep a copy outside the gateway's control.

#Step 8: Run the policy test matrix

1. Run:

   Copy

   python unit11/ai_gateway.py test

2. You should see:

PASS  AP agent, AP user, invoice read            expected allow got allow ok
PASS  AP agent, buyer, release draft             expected deny  got deny  tool not allowed for user 00100211
PASS  AP agent, AP user, PO date change          expected deny  got deny  tool not registered for AGT_INV_TRIAGE
PASS  Planning agent, planner, PO date change    expected allow got allow ok
PASS  Planning agent, AP user, invoice read      expected deny  got deny  tool not registered for AGT_SUPPLY_WATCH
PASS  AP agent, strong model                     expected deny  got deny  model 'strong' not allowed for team accounts-payable
PASS  Planning agent, strong model               expected allow got allow ok
PASS  Unknown key                                expected deny  got deny  unknown key
PASS  Unknown user                               expected deny  got deny  unknown user
PASS  Oversized input                            expected deny  got deny  input over 8000 characters

10 of 10 passed

Run this after every policy change. A gateway's policy is proven by the calls it refuses, so most rows expect deny.

#Step 9: Save your work in Git

git add unit11/ai_gateway.py .gitignore
git commit -m "Unit 11: AI gateway lab"

#What each part of the script does

#If something goes wrong

#The SAP way

As of 3 November 2026, SAP has four pieces that do parts of a gateway's job. They were announced at different times and are at different stages, so check each one's status in your region and edition before you design around it.

#Agent Gateway: outside agents calling Joule agents

SAP's Architecture Center announced general availability of inbound A2A interoperability through the Agent Gateway in October 2026. What it describes:

• The gateway is a single governed entry point. It validates incoming requests, applies access restrictions, routes calls and supports audit, without exposing SAP systems directly.
• Each outside client is registered as an Agent Gateway Consumer through SAP for Me and onboarded in three steps SAP calls formations: an identity in SAP Cloud Identity Services, trust with the gateway, and an access control list of the SAP agents it may invoke.
• Discovery is separate from permission: an agent can find a Joule agent's description without being allowed to call it.
• At runtime the gateway checks the token, resolves the customer tenant and checks the access list. The Joule agent then acts within the signed-in user's permissions.
• Calls from SAP agents out to other agents or MCP servers are described as future direction.

The inbound A2A reference architecture adds audit logging through SAP Cloud ALM, and says SAP Integration Suite can delegate calls to the Agent Gateway. SAP's broader Agentic AI page describes the gateway as handling authentication, principal propagation, policy enforcement and tenancy, with the effective permission being the intersection of user and agent permissions. It also says some capabilities are not yet generally available.

#MCP Gateway: agents calling tools

SAP's July 2026 Integration Suite update describes the MCP Gateway as turning APIs into agent tools with rate limiting, IP blocking, sign-in and authorization based on OpenID Connect, payload protection, and logs and traces. The same slides list principal propagation, federation of third-party MCP servers and an AI gateway for model calls with routing, token budgets and usage controls as roadmap items. How it compares with building your own tool layer is in SAP tools for agents.

#SAP Enterprise AI Gateway: announced

SAP's 27 October 2026 release says the new SAP Enterprise AI Gateway will offer a central way to manage AI traffic and will consolidate authentication, policy enforcement, routing, monitoring, usage tracking and auditing. SAP's 30 October article says it builds on SAP Integration Suite and controls traffic across MCP servers, APIs, agent-to-agent connections and large language models. Neither page gives an availability date, editions or pricing. Until SAP publishes them, treat it as announced.

#Orchestration service and Joule Studio controls

The orchestration service in generative AI hub is a managed service in SAP AI Core with one API across models, content filtering, data masking, grounding and templating. It is a pipeline for each of an application's model calls. It does the gateway's filtering and routing jobs for that application, but the SAP pages opened for this topic don't describe it as a cross-team budget or policy point.

For agents built in Joule Studio, SAP announced on 27 October 2026 observability and administrative controls for agent behavior, model usage and spend. Joule and Joule Studio covers Joule Studio itself.

#How to design for it today

Design your agents so the gateway is replaceable:

• Agents call one internal endpoint, by alias, never a provider's URL.
• Identity comes from your identity provider, not from keys the gateway invents.
• Your audit record format is yours, so it survives a change of gateway product.
• Your policy test matrix is yours too. Run it against whatever gateway you use, including SAP's when it arrives.

#Build vs. SAP

Open-source gateways cover much of the model side. LiteLLM's proxy, for example, supports budgets and rate limits per key, user and team. Read the defaults: its documentation says key budgets are not checked unless set, and need a database to be enforced. A gateway you didn't configure is a pass-through with a log.

#Architecture view

Where it sits. At Calder Pumpworks (fictional), agents run on SAP BTP and on Calder's own cloud. The gateway sits between them and everything they call. SAP's own routes stay in place: outside agents reach Joule agents through SAP's Agent Gateway, and SAP APIs still check the user's roles.

flowchart LR
  subgraph Agents
    A1[Invoice triage]
    A2[Supply watch]
  end
  A1 --> GW[AI gateway<br/>policy, limits, log]
  A2 --> GW
  GW --> M[Models<br/>orchestration service<br/>or provider]
  GW --> T[Tools: MCP Gateway<br/>or own tool layer]
  T --> S4[S/4HANA CP10<br/>roles checked]
  X[Partner agent] --> AG[SAP Agent Gateway<br/>inbound A2A]
  AG --> J[Joule agents]
  GW --> LOG[(Audit store)]
  IdP[Identity provider] -.tokens.-> GW

Non-functional requirements.

Integration and identity. The agent signs in to the identity provider and gets a token for the gateway, naming the agent and, for delegated work, the user. The gateway checks the token's audience and never forwards it. Toward models it uses its own credentials; toward SAP it uses a new token or principal propagation, so SAP checks the real user. Toward SAP, use released APIs, as in SAP tools for agents.

The decision to record. In the short form from Writing a solution design doc:

• Context: three teams build agents; each calls models directly with its own key; no shared budget or log. SAP Enterprise AI Gateway is announced with no date.
• Options: A, wait for SAP's gateway; B, an open-source gateway now, run by the platform team; C, a thin self-built gateway; D, the orchestration service per app, without a shared budget.
• Decision: B for model traffic, configured with budgets, limits and the audit format Calder owns; MCP Gateway for tools; SAP's Agent Gateway for inbound A2A. Review when SAP publishes availability.
• Consequences: Calder patches and runs the gateway; agents call aliases, so moving to SAP's gateway later is a configuration change plus a run of the policy test matrix.

What a design review will ask.

1. Can any agent reach a model or tool without passing the gateway? How do you know?
2. What happens to every agent when the gateway is down, and is that acceptable?
3. Who may change the policy, and is each change tested and logged?
4. Where are prompts stored, if at all, and who can read them?
5. What is the plan when SAP's gateway becomes available?

#Production concerns

• No bypass. Network rules, not good intentions: only the gateway can reach model providers, and model keys exist only there. Scan code and wikis for stray keys.
• Identity. Replace API keys with tokens from your identity provider. The lab trusts the user field the agent sends; production takes the user from a signed token, as in Agent permissions and SAP authorizations.
• SAP authorizations stay. The gateway's tool list is coarse. SAP's roles and business roles are the check on data. Never give the gateway a powerful technical user that makes SAP's check meaningless.
• Fail closed, fast. Policy, budget and identity lookups have timeouts. When one fails, refuse and log. Test this with the backend switched off.
• State. The lab sums budgets from its own log. A real gateway keeps counters in a shared store, so several gateway instances agree.
• Budgets with warnings. Alert owners at a share of the budget, not only at the limit. Give a named person a way to raise a budget for a day, logged.
• Prompts and logs. Keep fingerprints by default. If you need prompts for evaluation or support, store them separately, masked, with short retention. The gateway log feeds the dashboards in Observability for AI systems.
• Patching and inventory. The gateway joins your vulnerability process, next to SAP Security Notes. MCP security in 2026 shows why.
• Clean core. The gateway sits outside S/4HANA and uses released APIs. Nothing in it needs a modification of SAP code.
• Governance. The gateway's policy and log are evidence for the controls in AI governance for SAP programs: which agents exist, what they may do, and what they did.

#Pitfalls

• An optional gateway. One team calls the provider directly "just for the pilot". Now the log and budgets are incomplete, and nobody notices.
• Failing open. A timeout in the budget check lets calls through. The weekend the store is slow is the weekend the loop runs.
• Counting after the call. Metering only after the answer means one huge call can blow through the budget. Reserve the worst case first.
• Logging prompts by default. The audit log becomes the largest store of business and personal data in the program.
• Passing tokens through. The backend then sees the caller's token, rate limits keyed on the gateway's identity don't apply, and the log names the wrong caller.
• Budgets nobody owns. A limit without a named owner gets raised to "unlimited" at the first complaint.
• Treating the gateway as the authorization system. Coarse tool lists at the gateway, fine-grained data checks in SAP. Moving data rules into the gateway duplicates SAP's roles and drifts from them.
• Planning on an announcement. Building a roadmap on a product with no published date. Design for replaceability instead.

#Exercise

Calder's customer service team wants an order exception agent for blocked sales orders. Register it in your gateway and prove the policy.

1. Open unit11/ai_gateway.py.
2. In AGENTS, add a third agent: key calder-lab-key-order-exceptions, agent AGT_ORDER_EXC, team customer-service, tools sales_order_get and credit_release_request_draft. Copy the shape of the two existing entries, including key_hash(...).
3. In USERS, add "00100265", role customer service clerk, with the same two tools.
4. In TEAMS, add "customer-service" with models {"fast"} and daily_tokens 10000.
5. In FAKE_SAP, add sales_order_get returning an order with status credit block, and credit_release_request_draft returning a draft that waits for a person. Copy the shape of the existing lines.
6. In MATRIX, add three rows: the order agent reading an order for 00100265 (allow); the order agent drafting a credit release for the AP accountant 00100241 (deny); the order agent asking for the strong model (deny).
7. Run python unit11/ai_gateway.py test.
8. Run python unit11/ai_gateway.py --sample and report again, and check the existing results haven't changed.

Done when test prints 13 of 13 passed, the sample output is unchanged from Step 4, and you can say in one sentence why the AP accountant can't use the order agent to draft a credit release. Keep the file: the portfolio projects put agents behind a gateway like this one.

#Going further

For readers who want the standards and specifications behind the checks.

• OWASP LLM10:2025 Unbounded Consumption goes beyond cost: it covers model extraction through repeated queries and side-channel attacks, and lists controls such as limiting exposed probabilities and watching for unusual query patterns. Read it when you set alert rules on the gateway log.
• MCP security best practices explain the confused deputy problem for proxies that hold one client ID toward a third-party API. A gateway in front of MCP servers is such a proxy, so its consent and redirect checks apply to you. The scope minimization section shows how to grant tools step by step instead of all at once.
• LiteLLM's budget and rate-limit documentation shows a real policy model: budgets per key, user and team, budget periods that reset, separate token and request limits, and per-model limits that can fall back to another model. Compare its defaults with your policy test matrix.
• SAP's inbound A2A announcement separates discovery (agent cards and the open Agentic Resource Discovery specification) from permission (the access list). The same split is worth copying in your own registry: being listed is not being allowed.

One hard edge case: budgets across streaming and retries. A streamed answer that is cut off still used tokens, and a retry after a timeout may run twice at the provider. Meter what the provider reports, not what your client received, and give each call an idempotency key so retries can be matched.

#Check yourself