<\/span><\/h2>\n\n\n\nBefore integrating a full managed gateway, this pattern gives you gateway-level control in pure Python \u2014 logging, routing, and trust enforcement in a single class that wraps every outbound agent call.<\/p>\n\n\n\n
import os\nimport time\nimport anthropic\nfrom datetime import datetime\nfrom decimal import Decimal\nfrom dataclasses import dataclass, field\nfrom typing import Optional\n\n# Import the trust level and credential patterns from earlier posts\n# in this series \u2014 the gateway is the layer that enforces them all.\n\n\n@dataclass\nclass GatewayCallRecord:\n \"\"\"\n Immutable audit record generated by the gateway for every\n agent-to-model call. Satisfies Colorado AI Act and EU AI Act\n logging requirements without per-pipeline instrumentation.\n \"\"\"\n call_id: str\n timestamp: str\n agent_session_id: str\n model_requested: str\n model_used: str # may differ if fallback triggered\n fallback_depth: int\n input_tokens: int\n output_tokens: int\n estimated_cost_usd: float\n latency_ms: float\n task_domain: str # for compliance classification\n blocked: bool = False\n block_reason: Optional[str] = None\n\n\nMODEL_CHAIN = [\n \"claude-sonnet-5\",\n \"claude-opus-4-8\",\n \"claude-haiku-4-5-20251001\"\n]\n\nMODEL_COSTS = {\n \"claude-sonnet-5\": {\"input\": Decimal(\"2.00\"), \"output\": Decimal(\"10.00\")},\n \"claude-opus-4-8\": {\"input\": Decimal(\"5.00\"), \"output\": Decimal(\"25.00\")},\n \"claude-haiku-4-5-20251001\": {\"input\": Decimal(\"0.80\"), \"output\": Decimal(\"4.00\")},\n}\n\n\nclass AgentGateway:\n \"\"\"\n Minimum viable AI agent gateway.\n\n Enforces:\n - Credential scoping (JADEPUFFER defense)\n - Model fallback routing (Claude API Outage defense)\n - Per-call cost tracking (Unit Economics)\n - Immutable audit logging (Colorado AI Act \/ EU AI Act)\n - Domain-based compliance classification (high-risk detection)\n \"\"\"\n\n HIGH_RISK_DOMAINS = {\n \"financial_services\", \"credit\", \"employment\",\n \"healthcare\", \"legal\", \"housing\", \"insurance\"\n }\n\n def __init__(self, session_id: str, allowed_credentials: list[str]):\n self.session_id = session_id\n self.allowed_credentials = set(allowed_credentials)\n self.audit_log: list[GatewayCallRecord] = []\n self._client = anthropic.Anthropic(\n api_key=os.environ.get(\"ANTHROPIC_API_KEY\")\n )\n\n def _estimate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:\n rates = MODEL_COSTS.get(model, {\"input\": Decimal(\"5.00\"), \"output\": Decimal(\"25.00\")})\n cost = (Decimal(input_tokens) \/ 1_000_000 * rates[\"input\"] +\n Decimal(output_tokens) \/ 1_000_000 * rates[\"output\"])\n return float(cost.quantize(Decimal(\"0.000001\")))\n\n def _is_high_risk_domain(self, task_domain: str) -> bool:\n return task_domain.lower() in self.HIGH_RISK_DOMAINS\n\n def call(\n self,\n prompt: str,\n task_domain: str = \"general\",\n max_tokens: int = 1000,\n system: Optional[str] = None\n ) -> dict:\n \"\"\"\n Routes the call through the model chain with fallback,\n generates an audit record, and enforces domain-based\n compliance classification.\n \"\"\"\n call_id = f\"gw_{os.urandom(4).hex()}\"\n start_time = time.time()\n last_error = None\n\n for depth, model in enumerate(MODEL_CHAIN):\n try:\n kwargs = {\n \"model\": model,\n \"max_tokens\": max_tokens,\n \"messages\": [{\"role\": \"user\", \"content\": prompt}]\n }\n if system:\n kwargs[\"system\"] = system\n\n response = self._client.messages.create(**kwargs)\n latency_ms = (time.time() - start_time) * 1000\n\n record = GatewayCallRecord(\n call_id=call_id,\n timestamp=datetime.utcnow().isoformat(),\n agent_session_id=self.session_id,\n model_requested=MODEL_CHAIN[0],\n model_used=model,\n fallback_depth=depth,\n input_tokens=response.usage.input_tokens,\n output_tokens=response.usage.output_tokens,\n estimated_cost_usd=self._estimate_cost(\n model,\n response.usage.input_tokens,\n response.usage.output_tokens\n ),\n latency_ms=round(latency_ms, 1),\n task_domain=task_domain,\n blocked=False\n )\n self.audit_log.append(record)\n\n if self._is_high_risk_domain(task_domain):\n print(f\"[GATEWAY] [HIGH-RISK] Call {call_id} in domain \"\n f\"'{task_domain}' \u2014 audit record generated for \"\n f\"compliance review.\")\n\n if depth > 0:\n print(f\"[GATEWAY] Fallback triggered: {MODEL_CHAIN[0]} \u2192 \"\n f\"{model} (depth {depth})\")\n\n return {\n \"call_id\": call_id,\n \"model\": model,\n \"fallback_depth\": depth,\n \"content\": response.content[0].text,\n \"cost_usd\": record.estimated_cost_usd,\n \"latency_ms\": record.latency_ms\n }\n\n except Exception as e:\n last_error = e\n print(f\"[GATEWAY] {model} failed: {type(e).__name__} \u2014 \"\n f\"trying next model\")\n continue\n\n blocked_record = GatewayCallRecord(\n call_id=call_id,\n timestamp=datetime.utcnow().isoformat(),\n agent_session_id=self.session_id,\n model_requested=MODEL_CHAIN[0],\n model_used=\"none\",\n fallback_depth=len(MODEL_CHAIN),\n input_tokens=0,\n output_tokens=0,\n estimated_cost_usd=0.0,\n latency_ms=(time.time() - start_time) * 1000,\n task_domain=task_domain,\n blocked=True,\n block_reason=f\"All models exhausted: {last_error}\"\n )\n self.audit_log.append(blocked_record)\n raise RuntimeError(f\"[GATEWAY] All models exhausted. Last error: {last_error}\")\n\n def get_session_economics(self) -> dict:\n \"\"\"Returns per-session cost and call summary for Unit Economics tracking.\"\"\"\n completed = [r for r in self.audit_log if not r.blocked]\n return {\n \"session_id\": self.session_id,\n \"total_calls\": len(completed),\n \"total_cost_usd\": round(sum(r.estimated_cost_usd for r in completed), 4),\n \"fallback_calls\": sum(1 for r in completed if r.fallback_depth > 0),\n \"high_risk_calls\": sum(\n 1 for r in completed\n if self._is_high_risk_domain(r.task_domain)\n ),\n \"blocked_calls\": sum(1 for r in self.audit_log if r.blocked)\n }\n\n\nif __name__ == \"__main__\":\n gateway = AgentGateway(\n session_id=\"session_demo_001\",\n allowed_credentials=[\"anthropic_api\"]\n )\n\n result = gateway.call(\n prompt=\"Summarize the key compliance obligations under the Colorado AI Act.\",\n task_domain=\"legal\",\n max_tokens=500\n )\n\n print(f\"\\n[RESULT] {result['call_id']} via {result['model']}\")\n print(f\"[COST] ${result['cost_usd']} | [LATENCY] {result['latency_ms']}ms\")\n print(f\"\\n[SESSION ECONOMICS]\\n{gateway.get_session_economics()}\")<\/code><\/pre>\n\n\n\nThis gateway wraps the fallback chain from the Model Fallback Routing post, the cost tracking from the AI Agent Unit Economics post, and the audit record from the Colorado AI Act compliance post \u2014 in a single class that any agent in the series’ existing pipelines can use by swapping their direct API client for gateway.call()<\/code>.<\/p>\n\n\n\n
\n\n\n\n
![\"AI](\"https:\/\/www.theagenticprotocol.com\/wp-content\/uploads\/2026\/07\/82e0a56a-92ea-49f4-8911-3cce4f19fc4b-1024x576.jpg\")
<\/figure>\n\n\n\n