Every Prompt Is an Export: Build an Egress Layer Before You Ship AI Features

Most teams treat an LLM call like a function call. It is not. It is a data export to a company you do not control, over a network you do not own, into logs you cannot read. When you paste a support ticket into a model prompt, you have just shipped a customer’s name, email and possibly their address across a border. Legally and operationally that is closer to an SFTP upload to a vendor than to array.map().

I build AI automations for clients who have real customers, so this comes up on every project. The question is never “is AI safe?” The question is: which fields left our perimeter, to which vendor, under which contract, and can I prove it six months from now. This post is the architecture I use to answer that without slowing delivery down.

The failure mode nobody plans for

A typical AI feature grows like this:

1. Week 1: a script summarizes support tickets. You paste the whole ticket body into the prompt because that is fastest.
2. Week 3: you add the customer record for context. Now email, phone and plan tier go out too.
3. Week 5: you add an observability tool so you can debug prompts. Full payloads are now stored in a second vendor.
4. Week 7: someone builds an agent that reads the CRM and writes to Slack. It sends the whole row, including internal notes.
5. Month 4: a client asks for a data flow diagram for their security review. Nobody can produce one.

Nothing here was malicious. Every step was a reasonable local decision. The problem is that there was no single place where data crossed the boundary, so there was no place to put a rule.

The fix is boring and it works: make the boundary explicit, make it narrow, and make it the only way out.

One gateway, no exceptions

Rule number one in every project I run: application code never talks to a model provider SDK directly. It talks to my own llm module. That module is the only code in the repo allowed to import openai, anthropic, google.generativeai and friends.

This is not architectural purity. It is the only way to enforce anything later. If there are fourteen call sites, you have fourteen chances to forget redaction. If there is one, you have one place to add a rule and one place to test it.

I enforce it with a lint rule and a CI grep:

# fails the build if a provider SDK is imported outside src/llm/
grep -rEn "from (openai|anthropic)" src \
  | grep -v "^src/llm/" \
  && { echo "provider SDK imported outside the gateway"; exit 1; }

Ugly, five lines, catches the mistake every time. On bigger codebases I use ESLint no-restricted-imports with a path exception, or an import-linter contract in Python.

Classify fields once, at the schema

You cannot write a redaction rule for “the ticket body”. You can write one for a tagged field. So I tag sensitivity at the data model level, not at the prompt level.

from dataclasses import dataclass, field

# PUBLIC   - safe to send anywhere
# INTERNAL - business data, send only to contracted vendors
# PII      - identifies a person, tokenize before sending
# SECRET   - never leaves the process

@dataclass
class Ticket:
    id: str = field(metadata={"sens": "INTERNAL"})
    subject: str = field(metadata={"sens": "INTERNAL"})
    body: str = field(metadata={"sens": "PII"})       # free text, assume the worst
    customer_email: str = field(metadata={"sens": "PII"})
    api_key_hint: str = field(metadata={"sens": "SECRET"})

Free-text fields always get PII. Users put their phone number in the message body. They always do.

The gateway then takes structured objects, not strings. That single decision is what makes everything downstream possible:

result = llm.run(
    task="ticket_triage",
    inputs={"ticket": ticket},
    model_policy="eu_contracted",
)

If the gateway receives a pre-rendered prompt string, it has no idea what is inside it and can only fall back to regex scanning. Regex scanning is a smoke detector, not a wall.

Reversible tokenization beats deletion

The naive approach is to strip PII. Then the model output says “contact the customer” instead of “email Anna at her address” and your automation becomes useless for anything that needs to write back.

What I do instead: replace PII with stable local tokens, keep the map in memory or in my own database, and rehydrate on the way back.

import re, uuid

EMAIL = re.compile(r"[\w.+-]+@[\w-]+\.[\w.]+")
PHONE = re.compile(r"\+?\d[\d\s().-]{7,}\d")

class Vault:
    def __init__(self):
        self.fwd = {}   # real value -> token
        self.rev = {}   # token -> real value

    def token(self, value, kind):
        if value in self.fwd:
            return self.fwd[value]
        t = f"<{kind}_{uuid.uuid4().hex[:6]}>"
        self.fwd[value] = t
        self.rev[t] = value
        return t

    def mask(self, text):
        text = EMAIL.sub(lambda m: self.token(m.group(), "EMAIL"), text)
        text = PHONE.sub(lambda m: self.token(m.group(), "PHONE"), text)
        return text

    def unmask(self, text):
        for t, v in self.rev.items():
            text = text.replace(t, v)
        return text

The model sees <EMAIL_9f2c11> is asking about invoice <INV_44a0>. It reasons perfectly well about tokens. Your downstream code rehydrates before sending the reply. The vendor never held the real values.

Two details that matter in practice. Tokens must be stable within a single request so the model can see that two mentions are the same person. And never make the token guessable from the value, no hashing of the email into the token, because a leaked prompt log plus a wordlist undoes your work.

A policy table, not a vibe

The next thing clients ask is “which model can we use for this?” I answer with a table checked into the repo, not with an opinion in Slack.

policies:
  eu_contracted:
    allow_sensitivity: [PUBLIC, INTERNAL, PII_TOKENIZED]
    providers: [vendor_a_eu]
    retention_days: 0
    training_optout: true
  public_only:
    allow_sensitivity: [PUBLIC]
    providers: [vendor_a, vendor_b, local_ollama]
  local_only:
    allow_sensitivity: [PUBLIC, INTERNAL, PII]
    providers: [local_ollama]

The gateway refuses the call when the payload’s highest sensitivity is not in allow_sensitivity for the chosen policy. Not a warning. An exception that fails the job. Fail closed, always. A triage job that stops is an incident ticket. A triage job that quietly ships raw customer records to an unvetted endpoint is a breach notification.

This also gives you a real answer for the security questionnaire: here is the file, here are the vendors, here is the retention setting per route.

The egress ledger

For every outbound call I log a record that does not contain the payload:

{
  "ts": "2026-02-04T09:12:44Z",
  "task": "ticket_triage",
  "policy": "eu_contracted",
  "provider": "vendor_a_eu",
  "model": "mid-tier-v3",
  "tenant": "acme",
  "fields": ["ticket.subject:INTERNAL", "ticket.body:PII_TOKENIZED"],
  "payload_sha256": "3f9a...",
  "bytes_out": 4182,
  "tokens_masked": 3,
  "trace_id": "01HQ..."
}

Field names and sensitivity labels, never values. The hash lets you prove that a given payload was or was not sent without storing the payload. bytes_out per tenant per day is the single best leak detector I know: when someone changes a prompt to include the full order history, the number jumps by an order of magnitude and your alert fires before your client notices.

Keep the ledger in your own database with your own retention. It is the artifact you hand to an auditor.

Do not forget the second and third exits

The model API is the obvious door. These are the ones I find during reviews:

• Prompt observability tools. They store full request and response bodies by default. That is a second vendor holding the same PII. Either send masked payloads only, or self-host.
• Embeddings and vector stores. An embedding is derived from the original text and often sits in a managed cloud index. Same policy rules apply.
• Error trackers. A failed call frequently attaches the request body to the exception. Scrub the payload in your error handler, not in the dashboard settings.
• Agent tools. A file-read tool, a shell tool or a browser tool gives the model a way to pull data you never intended to classify. Whitelist paths and domains at the tool layer.
• Third-party agent skills and plugins. Installing someone’s skill pack is installing code with network access into the same process as your credentials. Read it, or run it in a container with no secrets mounted.

Tests that fail the build

Policy documents rot. Tests do not. Three I add to every project:

def test_pii_never_reaches_provider(fake_provider):
    ticket = Ticket(body="call me at +84 90 123 4567, anna@example.com", ...)
    llm.run(task="ticket_triage", inputs={"ticket": ticket},
            model_policy="eu_contracted")
    sent = fake_provider.last_request_text
    assert "anna@example.com" not in sent
    assert "90 123 4567" not in sent

def test_policy_violation_fails_closed():
    with pytest.raises(EgressPolicyError):
        llm.run(task="ticket_triage", inputs={"ticket": raw_pii_ticket},
                model_policy="public_only")

def test_every_call_writes_a_ledger_row(fake_provider, ledger):
    llm.run(task="ticket_triage", inputs={"ticket": ticket}, model_policy="eu_contracted")
    assert len(ledger.rows) == 1
    assert "anna@example.com" not in json.dumps(ledger.rows[0])

The fake provider records the exact bytes that would have gone over the wire. That is the assertion that actually protects you, because it tests the boundary rather than the intention.

What this costs you

About a day and a half on a new project if you do it at the start. Roughly a week if you retrofit it onto an AI feature that already has a dozen call sites. It buys you three things: you can pass a client security review without inventing documentation, you can switch providers in an afternoon because everything already goes through one module, and you can answer the “what exactly did you send” question with a query instead of a guess.

Start with the gateway and the grep check. Even with no redaction at all, having one exit point turns a vague risk into a fixable one. Everything else bolts onto it.