AI features often have to satisfy two demands at once: handle requests responsibly and avoid wasting compute. Those goals can conflict. More checks can add latency; strict input limits can reject useful work; and sending every request to a large model can be expensive. A small gateway between an application and its model can make these decisions explicit and testable.
This tutorial builds a Python gateway that rejects oversized requests, blocks a simple category of sensitive data, redacts email addresses, enforces a per-request token budget, and writes structured audit events. It uses a mock model so the example runs without credentials, network access, or a vendor SDK. The checks are engineering controls, not a declaration of legal compliance: real requirements depend on the system, data, jurisdiction, and organisation.
The problem: controls need to be part of the request path
Policies written in documentation do not automatically constrain model calls. If each feature team implements its own checks, the same request may be treated differently in different parts of a product. A gateway centralises a few operational decisions: what input is accepted, what is transformed, how much work is allowed, and what gets recorded.
There is no universal token counter in this example. Model tokenisation depends on the model and its tokenizer, so counting whitespace-separated words is only a rough estimate. The gateway uses that estimate to demonstrate a budget check; a production integration should use the provider's tokenizer or usage data where appropriate. Likewise, an email regular expression is not a complete detector for personal or confidential information.
Setup
Use Python 3 and create a directory for the example. The implementation uses only the standard library. Save the main program below as gateway.py. In a real service, keep policy values in reviewed configuration, avoid committing secrets, and decide which fields may be retained in logs.
from dataclasses import dataclass
@dataclass(frozen=True)
class Policy:
max_input_chars: int = 4000
max_estimated_tokens: int = 900
max_output_tokens: int = 300
def estimate_tokens(text: str) -> int:
"""A deliberately rough estimate for this dependency-free demo."""
return len(text.split())
if __name__ == "__main__":
policy = Policy()
prompt = "Summarise the deployment checklist for the team."
estimate = estimate_tokens(prompt)
print(f"Input estimate: {estimate}")
print(f"Within budget: {estimate <= policy.max_estimated_tokens}")Run it with python gateway.py. The example defines limits as immutable settings, which makes it harder for request-handling code to accidentally mutate policy during a call. The output is a useful first check, but it does not yet reject requests or invoke a model.
Step 1: define what the gateway accepts
Validation should happen before expensive work. First reject empty or overlong input. Next, apply any domain-specific rules. The example below blocks a deliberately narrow marker, [CONFIDENTIAL], to demonstrate a policy rejection. It does not identify confidential material by itself; a real system needs controls suited to its data and threat model.
Add the following functions to the program. The exception makes a rejected request distinct from a model or programming error, allowing the caller to handle it deliberately.
class PolicyError(ValueError):
pass
def validate_input(text: str, max_chars: int) -> None:
if not isinstance(text, str) or not text.strip():
raise PolicyError("Input must be a non-empty string")
if len(text) > max_chars:
raise PolicyError("Input exceeds the character limit")
if "[CONFIDENTIAL]" in text.upper():
raise PolicyError("Input contains a blocked confidentiality marker")
if __name__ == "__main__":
try:
validate_input("Summarise a public release note.", 4000)
print("Accepted")
except PolicyError as exc:
print(f"Rejected: {exc}")This standalone check can be run as written. In the final program, the same validation runs inside the gateway, immediately before any model call. Do not rely on a user interface to enforce limits: requests can arrive through other clients, background jobs, or future application paths.
Step 2: transform data and control the budget
Blocking is not the only option. Some applications can reduce exposure by replacing specific values before sending text to a model. The next function redacts email-like strings. It is a narrow demonstration, not a general-purpose data-loss-prevention system. Keep the original prompt out of audit records unless there is a documented need and an approved retention policy.
The budget check accounts for the estimated input and configured output allowance. A real provider may count system instructions, tool schemas, conversation history, and other fields as input too. Include those in the estimate when integrating an actual API.
import re
EMAIL_PATTERN = re.compile(r"\b[\w.+-]+@[\w.-]+\.[A-Za-z]{2,}\b")
def redact_emails(text: str) -> str:
return EMAIL_PATTERN.sub("[REDACTED_EMAIL]", text)
def check_budget(text: str, max_total_tokens: int, output_tokens: int) -> int:
estimate = len(text.split())
if estimate + output_tokens > max_total_tokens:
raise ValueError("Estimated request exceeds token budget")
return estimate
if __name__ == "__main__":
prompt = "Send the summary to [email protected] after review."
safe_prompt = redact_emails(prompt)
print(safe_prompt)
print("Estimated input tokens:", check_budget(safe_prompt, 100, 20))Run this block independently to see the transformation. Redaction can change meaning, and patterns can miss unusual addresses or non-email identifiers. In production, define which data classes can be sent, which must be removed, and what happens when a detector is uncertain. Avoid treating a successful regex match as proof that the remaining text is safe.
Step 3: connect checks to a model adapter
The gateway below puts the pieces together. Its MockModel stands in for a model client and returns a predictable response. Replace it with an adapter for a chosen provider only after deciding how authentication, timeouts, retries, data handling, and provider-side logging should work.
The gateway writes one JSON object per request to a JSON Lines audit file. It records a request identifier, outcome, estimated input size, and model name, but not the prompt or response. Even metadata can be sensitive in some environments, so select fields and retention deliberately. The code records both accepted and rejected requests, while avoiding exception details that might contain user input.
import json
import re
import uuid
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Protocol
@dataclass(frozen=True)
class Policy:
max_input_chars: int = 4000
max_total_tokens: int = 1200
max_output_tokens: int = 300
class PolicyError(ValueError):
pass
class Model(Protocol):
name: str
def generate(self, prompt: str, max_output_tokens: int) -> str:
...
class MockModel:
name = "mock-model"
def generate(self, prompt: str, max_output_tokens: int) -> str:
words = prompt.split()
return "Processed request with " + str(len(words)) + " input words."
EMAIL_PATTERN = re.compile(r"\b[\w.+-]+@[\w.-]+\.[A-Za-z]{2,}\b")
def estimate_tokens(text: str) -> int:
return len(text.split())
def redact_emails(text: str) -> str:
return EMAIL_PATTERN.sub("[REDACTED_EMAIL]", text)
def validate_input(text: str, policy: Policy) -> None:
if not isinstance(text, str) or not text.strip():
raise PolicyError("Input must be a non-empty string")
if len(text) > policy.max_input_chars:
raise PolicyError("Input exceeds the character limit")
if "[CONFIDENTIAL]" in text.upper():
raise PolicyError("Input contains a blocked confidentiality marker")
def write_audit(path: Path, event: dict) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
with path.open("a", encoding="utf-8") as stream:
stream.write(json.dumps(event, sort_keys=True) + "\n")
class AIGateway:
def __init__(self, model: Model, policy: Policy, audit_path: Path):
self.model = model
self.policy = policy
self.audit_path = audit_path
def handle(self, text: str) -> str:
request_id = str(uuid.uuid4())
estimated = 0
outcome = "rejected"
try:
validate_input(text, self.policy)
safe_text = redact_emails(text)
estimated = estimate_tokens(safe_text)
if estimated + self.policy.max_output_tokens > self.policy.max_total_tokens:
raise PolicyError("Estimated request exceeds token budget")
result = self.model.generate(safe_text, self.policy.max_output_tokens)
outcome = "accepted"
return result
finally:
event = {
"request_id": request_id,
"timestamp": datetime.now(timezone.utc).isoformat(),
"outcome": outcome,
"estimated_input_tokens": estimated,
"model": self.model.name,
}
write_audit(self.audit_path, event)
if __name__ == "__main__":
gateway = AIGateway(
model=MockModel(),
policy=Policy(),
audit_path=Path("logs/ai-gateway.jsonl"),
)
try:
answer = gateway.handle("Summarise this release note for [email protected].")
print(answer)
except PolicyError as exc:
print(f"Request rejected: {exc}")Save the complete block as gateway.py and run python gateway.py. It creates logs/ai-gateway.jsonl relative to the current directory. Inspect the file to verify that it contains metadata rather than the prompt. In a deployed service, use an appropriate logging pipeline with access control, retention limits, and protection against concurrent-write issues.
Step 4: verify the policy paths
Tests should cover both accepted and rejected requests. This standard-library test module imports the gateway and checks that the mock adapter runs for valid input, while a blocked marker raises a policy error. It also checks that a rejected request is audited. Save it as test_gateway.py beside gateway.py, then run python -m unittest -v.
import json
import tempfile
import unittest
from pathlib import Path
from gateway import AIGateway, MockModel, Policy, PolicyError
class GatewayTests(unittest.TestCase):
def setUp(self):
self.temp_dir = tempfile.TemporaryDirectory()
self.audit_path = Path(self.temp_dir.name) / "audit.jsonl"
self.gateway = AIGateway(MockModel(), Policy(), self.audit_path)
def tearDown(self):
self.temp_dir.cleanup()
def test_accepts_valid_request_and_writes_event(self):
result = self.gateway.handle("Summarise the public update.")
self.assertIn("Processed request", result)
event = json.loads(self.audit_path.read_text(encoding="utf-8"))
self.assertEqual(event["outcome"], "accepted")
self.assertNotIn("prompt", event)
def test_rejects_confidential_marker_and_audits(self):
with self.assertRaises(PolicyError):
self.gateway.handle("Review [CONFIDENTIAL] draft")
event = json.loads(self.audit_path.read_text(encoding="utf-8"))
self.assertEqual(event["outcome"], "rejected")
if __name__ == "__main__":
unittest.main()The audit test intentionally checks only the basic event shape. Extend coverage for empty input, overlong input, token-budget rejection, and redaction. If the system uses tools or retrieval, add tests for those paths too: a policy applied only to the initial prompt may not govern text introduced later.
Common pitfalls
- Calling an estimate an exact token count. Word counts are a teaching shortcut. Use model-specific token accounting where available, and leave room for system instructions and conversation context.
- Assuming redaction makes data safe. The sample handles email-shaped strings only. Names, account identifiers, secrets, and combinations of details may remain identifying.
- Logging too much. Prompts and responses can contain sensitive information. Decide the purpose of each audit field, restrict access, and set retention before enabling production logging.
- Confusing a policy error with an outage. A rejected request is a deliberate decision; a timeout or provider failure is an operational failure. Handle and monitor them separately in a real adapter.
- Adding retries without limits. Retries can increase latency and consume more resources. Bound them, use timeouts, and consider whether repeating a request is safe for the operation involved.
- Treating a mock as a production integration. The mock has no network, authentication, provider-side data handling, or real token accounting. Those concerns need explicit implementation and review.
Wrap-up
A gateway gives an application one place to enforce request limits, apply transformations, check a compute budget, and record a minimal audit event. The example is intentionally small: its detectors and token estimate are not sufficient for high-assurance use. Keep the controls visible, test rejection paths, and adapt them to the actual model, data flows, and operational requirements before connecting a production service.
