Understanding Multi-Region AI Agent Deployment
Multi-region AI agent deployment refers to the practice of distributing your AI agent infrastructure across multiple geographic regions simultaneously. Instead of running all agent inference, tool execution, and memory management from a single cloud region, you strategically position agent endpoints in data centers around the world. Each regional instance operates independently but remains part of a coordinated system, capable of handling user requests from the nearest geographic point of presence.
At its core, this architecture addresses two fundamental tensions in modern AI agent systems: latency—the physical time it takes for data to travel between the user and your agent—and compliance—the increasingly complex web of data sovereignty laws that govern where AI processing can legally occur. An AI agent deployed in us-east-1 might serve users in New York with sub-20ms latency, but that same agent would introduce 200-300ms of round-trip delay for users in Singapore, while potentially violating local data residency requirements in the process.
What Defines a Multi-Region Agent?
A multi-region AI agent consists of several key components distributed across regions:
- Regional inference endpoints — LLM inference servers deployed in multiple geographic zones, each capable of processing agent reasoning loops independently
- Region-aware routing layer — A traffic director that maps incoming user sessions to the optimal regional endpoint based on proximity, health, and regulatory constraints
- Distributed state management — Mechanisms for maintaining agent memory, conversation history, and tool execution state that respect data residency boundaries
- Global coordination plane — Lightweight cross-region synchronization for agent configuration, model versions, and non-sensitive metadata
Why Multi-Region Deployment Matters Now
The stakes for AI agent deployment have shifted dramatically. Users expect conversational agents to respond within human conversation cadence—typically under 300ms for the first token. Simultaneously, regulations like GDPR in Europe, the AI Act, PIPL in China, and frameworks emerging in Brazil, India, and Australia impose strict requirements on where AI inference can occur and where user data may persist. A single-region deployment creates an impossible trade-off: optimize for latency and risk compliance violations, or centralize for compliance and deliver sluggish experiences to distant users.
Multi-region deployment resolves this tension by allowing you to place agent compute physically close to users while keeping data within legally approved boundaries. The result is a system that feels responsive everywhere and satisfies auditor scrutiny simultaneously.
Core Architectural Concepts
Regional Inference Endpoints
Each region in your deployment runs a self-contained agent inference stack. This typically includes the LLM inference engine (serving models like GPT-4, Claude, or open models via vLLM or TensorRT), the agent orchestration logic (tool calling, planning, reflection loops), and a local cache for frequently accessed embeddings or retrieval indices. These endpoints are designed to operate autonomously for the duration of a user session, minimizing cross-region chatter during latency-sensitive operations.
Data Residency and Sovereignty Boundaries
Compliance requirements dictate that certain categories of data—personally identifiable information, conversation transcripts, tool call payloads containing user data—must never leave a designated legal jurisdiction. Multi-region architectures enforce this by implementing data residency pins that lock a user's session to a specific region and ensure all persistent storage, logging, and agent memory operations occur within that jurisdiction. Even if a failover occurs due to an outage, compliance-pinned sessions cannot be arbitrarily redirected to a non-compliant region.
Agent Routing Strategies
The routing layer sits at the edge of your deployment, making real-time decisions about which regional agent endpoint should handle an incoming request. Common strategies include:
- Geo-proximity routing — Route to the region with the lowest network latency to the user's IP address
- Compliance-first routing — Override proximity and force routing to a legally mandated region based on the user's detected or declared jurisdiction
- Affinity-based routing — Maintain session stickiness so a user's multi-turn conversation stays pinned to the same regional agent instance
- Health-aware routing — Automatically redirect traffic away from regions experiencing degradation, provided compliance constraints allow it
Setting Up Your First Multi-Region Agent
Let's begin with a practical foundation. We'll define a configuration-driven setup that maps regions to their respective agent endpoints and establishes the basic routing rules. The following configuration file defines three regions—US East, EU West, and Asia Pacific Southeast—each with distinct inference endpoints and compliance profiles.
# agent-regions.yaml — Multi-region agent deployment configuration
regions:
us-east:
provider: aws
zone: us-east-1
inference_endpoint: "https://agent-us-east.internal.ai/api/v1"
llm_model: "gpt-4-turbo"
compliance_domains: ["us-data", "ccpa"]
priority: 1
health_check_url: "https://agent-us-east.internal.ai/health"
eu-west:
provider: azure
zone: westeurope
inference_endpoint: "https://agent-eu-west.internal.ai/api/v1"
llm_model: "gpt-4-turbo"
compliance_domains: ["gdpr", "eu-ai-act"]
priority: 2
health_check_url: "https://agent-eu-west.internal.ai/health"
data_residency_pin: true
ap-southeast:
provider: gcp
zone: asia-southeast1
inference_endpoint: "https://agent-ap-se.internal.ai/api/v1"
llm_model: "gpt-4-turbo"
compliance_domains: ["pdpa-singapore", "pipL"]
priority: 3
health_check_url: "https://agent-ap-se.internal.ai/health"
global:
coordination_plane: "https://coordinator.global.internal.ai"
session_affinity_ttl_seconds: 3600
default_fallback_strategy: "next-healthy-region"With the configuration in place, we can build a lightweight routing service that reads this configuration and makes intelligent routing decisions at request time. The router must handle geo-IP resolution, compliance rule evaluation, and session affinity simultaneously.
# agent_router.py — Multi-region agent routing service
import yaml
import hashlib
from datetime import datetime, timedelta
from typing import Optional, Dict, Any
import geoip2.database
import httpx
import asyncio
class MultiRegionRouter:
def __init__(self, config_path: str = "agent-regions.yaml"):
with open(config_path, 'r') as f:
self.config = yaml.safe_load(f)
self.regions = self.config['regions']
self.global_config = self.config['global']
self.geo_reader = geoip2.database.Reader('GeoLite2-Country.mmdb')
self.session_pins: Dict[str, str] = {} # session_id -> region_id
self.pin_expiry: Dict[str, datetime] = {}
def resolve_user_jurisdiction(self, ip_address: str) -> str:
"""Determine jurisdiction from user IP for compliance routing."""
try:
response = self.geo_reader.country(ip_address)
country_code = response.country.iso_code
except Exception:
country_code = "US" # Fallback default
jurisdiction_map = {
"US": "us", "CA": "us",
"DE": "eu", "FR": "eu", "IT": "eu", "ES": "eu",
"GB": "eu", "NL": "eu", "BE": "eu", "IE": "eu",
"SG": "ap-southeast", "CN": "ap-southeast",
"JP": "ap-southeast", "KR": "ap-southeast",
}
return jurisdiction_map.get(country_code, "us")
async def check_region_health(self, region_id: str) -> bool:
"""Perform async health check on a regional endpoint."""
region = self.regions.get(region_id)
if not region:
return False
try:
async with httpx.AsyncClient(timeout=5.0) as client:
resp = await client.get(region['health_check_url'])
return resp.status_code == 200
except Exception:
return False
def get_compliant_regions(self, jurisdiction: str) -> list[str]:
"""Filter regions that satisfy compliance for a given jurisdiction."""
compliant = []
for region_id, region in self.regions.items():
if jurisdiction in region.get('compliance_domains', []):
compliant.append(region_id)
return compliant
async def route_request(self, session_id: str, ip_address: str) -> str:
"""Primary routing decision combining session affinity, compliance, and health."""
# Step 1: Check session affinity — existing pinned sessions
if session_id in self.session_pins:
expiry = self.pin_expiry.get(session_id, datetime.min)
if datetime.utcnow() < expiry:
pinned_region = self.session_pins[session_id]
is_healthy = await self.check_region_health(pinned_region)
if is_healthy:
return pinned_region
# Unhealthy pin — clear and fall through to fresh routing
del self.session_pins[session_id]
# Step 2: Determine compliance jurisdiction
jurisdiction = self.resolve_user_jurisdiction(ip_address)
compliant_regions = self.get_compliant_regions(jurisdiction)
# Step 3: If no compliant regions, fall back to default (with warning)
if not compliant_regions:
compliant_regions = list(self.regions.keys())
# Step 4: Health-filter compliant regions
healthy_compliant = []
for region_id in compliant_regions:
if await self.check_region_health(region_id):
healthy_compliant.append(region_id)
if not healthy_compliant:
# Last resort: try any healthy region
for region_id in self.regions:
if await self.check_region_health(region_id):
healthy_compliant.append(region_id)
# Step 5: Select lowest latency region among healthy compliant options
# In production, integrate real latency metrics; here we use priority
selected = min(healthy_compliant,
key=lambda r: self.regions[r].get('priority', 999))
# Step 6: Pin session for affinity
self.session_pins[session_id] = selected
self.pin_expiry[session_id] = datetime.utcnow() + timedelta(
seconds=self.global_config.get('session_affinity_ttl_seconds', 3600)
)
return selected
# Usage example
async def main():
router = MultiRegionRouter()
target_region = await router.route_request(
session_id="sess_abc123",
ip_address="89.30.45.100" # Example French IP
)
print(f"Routed to region: {target_region}")
# Output: Routed to region: eu-west
if __name__ == "__main__":
asyncio.run(main())Latency-Optimized Agent Routing
Latency optimization in multi-region deployments goes beyond simple geo-IP routing. Real-world networks exhibit asymmetric routing, transient congestion, and varying inference queue depths across regions. A sophisticated approach incorporates continuous latency measurement, predictive selection, and circuit breaker patterns to maintain consistently low response times.
Building a Latency-Aware Region Selector
The following implementation demonstrates a latency-aware routing component that maintains a rolling latency window for each region and uses it to make data-driven routing decisions. It integrates with the compliance router from the previous section, layering latency optimization on top of compliance guarantees.
# latency_optimizer.py — Latency-aware region selection
import numpy as np
from collections import deque
from datetime import datetime
import asyncio
import httpx
from typing import Dict, Deque
class LatencyOptimizer:
def __init__(self, window_size: int = 100, alpha: float = 0.3):
self.latency_windows: Dict[str, Deque[float]] = {}
self.ema_latencies: Dict[str, float] = {}
self.window_size = window_size
self.alpha = alpha # EMA smoothing factor
self.failure_counts: Dict[str, int] = {}
self.circuit_open_until: Dict[str, datetime] = {}
def record_latency(self, region_id: str, latency_ms: float):
"""Record a latency sample with exponential moving average."""
if region_id not in self.latency_windows:
self.latency_windows[region_id] = deque(maxlen=self.window_size)
self.latency_windows[region_id].append(latency_ms)
if region_id not in self.ema_latencies:
self.ema_latencies[region_id] = latency_ms
else:
self.ema_latencies[region_id] = (
self.alpha * latency_ms +
(1 - self.alpha) * self.ema_latencies[region_id]
)
def record_failure(self, region_id: str):
"""Track failures and potentially open circuit breaker."""
self.failure_counts[region_id] = self.failure_counts.get(region_id, 0) + 1
if self.failure_counts[region_id] >= 5:
# Open circuit for 30 seconds
from datetime import timedelta
self.circuit_open_until[region_id] = datetime.utcnow() + timedelta(seconds=30)
def record_success(self, region_id: str):
"""Reset failure counter on successful request."""
self.failure_counts[region_id] = 0
def is_circuit_closed(self, region_id: str) -> bool:
"""Check if circuit breaker allows requests to this region."""
open_until = self.circuit_open_until.get(region_id)
if open_until and datetime.utcnow() < open_until:
return False
return True
def get_predicted_latency(self, region_id: str) -> float:
"""Return EMA-smoothed latency prediction for a region."""
return self.ema_latencies.get(region_id, 999999.0)
def select_optimal_region(self, candidate_regions: list[str]) -> str:
"""Select region with lowest predicted latency among candidates,
respecting circuit breaker state."""
best_region = None
best_latency = float('inf')
for region_id in candidate_regions:
if not self.is_circuit_closed(region_id):
continue
predicted = self.get_predicted_latency(region_id)
if predicted < best_latency:
best_latency = predicted
best_region = region_id
return best_region if best_region else candidate_regions[0]
# Integration with the router
class LatencyAwareMultiRegionRouter:
def __init__(self, config_path: str = "agent-regions.yaml"):
self.base_router = MultiRegionRouter(config_path)
self.latency_optimizer = LatencyOptimizer()
async def route_request(self, session_id: str, ip_address: str) -> str:
# First, get compliant candidate regions
jurisdiction = self.base_router.resolve_user_jurisdiction(ip_address)
compliant = self.base_router.get_compliant_regions(jurisdiction)
if not compliant:
compliant = list(self.base_router.regions.keys())
# Filter healthy regions
healthy = []
for r in compliant:
if await self.base_router.check_region_health(r):
healthy.append(r)
if not healthy:
healthy = compliant
# Apply latency-aware selection on top of compliance-health filter
optimal = self.latency_optimizer.select_optimal_region(healthy)
# Pin session
self.base_router.session_pins[session_id] = optimal
return optimal
async def execute_agent_request(self, region_id: str, payload: dict) -> dict:
"""Execute request and record latency metrics."""
endpoint = self.base_router.regions[region_id]['inference_endpoint']
start = datetime.utcnow()
try:
async with httpx.AsyncClient(timeout=30.0) as client:
resp = await client.post(f"{endpoint}/agent/run", json=payload)
resp.raise_for_status()
elapsed_ms = (datetime.utcnow() - start).total_seconds() * 1000
self.latency_optimizer.record_latency(region_id, elapsed_ms)
self.latency_optimizer.record_success(region_id)
return resp.json()
except Exception as e:
self.latency_optimizer.record_failure(region_id)
raiseActive Latency Probing
Beyond passive measurement, active probing from edge locations provides predictive latency data before users even arrive. The following background task runs periodic latency probes from distributed probe points, feeding the optimizer with fresh network intelligence.
# latency_probe.py — Background active latency measurement
import asyncio
import httpx
import time
class LatencyProbeService:
def __init__(self, regions_config: dict, probe_interval_seconds: int = 10):
self.regions = regions_config
self.interval = probe_interval_seconds
self.optimizer = None # Reference to shared LatencyOptimizer
async def probe_all_regions(self):
"""Probe all configured regions and record latencies."""
for region_id, region in self.regions.items():
endpoint = region['health_check_url']
try:
start = time.perf_counter()
async with httpx.AsyncClient(timeout=5.0) as client:
resp = await client.get(endpoint)
resp.raise_for_status()
latency_ms = (time.perf_counter() - start) * 1000
self.optimizer.record_latency(region_id, latency_ms)
self.optimizer.record_success(region_id)
except Exception:
self.optimizer.record_failure(region_id)
# Record a penalty latency to deprioritize failing regions
self.optimizer.record_latency(region_id, 10000.0)
async def run_probe_loop(self):
"""Continuously probe regions at configured interval."""
while True:
await self.probe_all_regions()
await asyncio.sleep(self.interval)
# Initialize alongside the router
async def initialize_services():
optimizer = LatencyOptimizer()
probe_service = LatencyProbeService(
regions_config=load_config()['regions'],
probe_interval_seconds=10
)
probe_service.optimizer = optimizer
asyncio.create_task(probe_service.run_probe_loop())
return optimizerCompliance-Aware Deployment Patterns
Compliance in multi-region AI agent deployments is not a checkbox—it's a continuous architectural constraint that shapes every layer of the stack. The following patterns represent battle-tested approaches to satisfying data residency requirements while maintaining agent functionality.
Pattern 1: Data Classification and Tagging
Before you can enforce residency, you must classify the data flowing through your agent system. Not all data carries the same compliance weight. Agent configuration, model weights, and anonymous usage metrics may freely replicate across regions, while conversation history, user PII, and tool call arguments containing personal data require strict pinning.
# data_classification.py — Tagging data with residency requirements
from enum import Enum
from dataclasses import dataclass
from typing import Optional
class DataClassification(Enum):
GLOBAL_REPLICABLE = "global" # Safe to replicate anywhere
REGIONAL_PINNED = "regional" # Must stay in user's jurisdiction
EPHEMERAL_INFERENCE = "ephemeral" # Exists only during inference, no persistence
@dataclass
class DataTag:
classification: DataClassification
jurisdiction: Optional[str] = None # e.g., "eu", "us", "ap-southeast"
retention_policy: str = "session-duration"
class ComplianceTaggingMiddleware:
"""Tags every piece of agent data with its compliance classification."""
def tag_user_input(self, user_id: str, jurisdiction: str,
raw_input: dict) -> dict:
"""Wrap user input with compliance metadata."""
return {
"payload": raw_input,
"_compliance": {
"classification": DataClassification.REGIONAL_PINNED.value,
"jurisdiction": jurisdiction,
"user_id_hash": self._pseudonymize(user_id),
"timestamp_utc": self._utc_now(),
"retention_policy": "jurisdiction-dependent"
}
}
def tag_agent_memory(self, memory_entry: dict, jurisdiction: str) -> dict:
"""Tag agent memory entries for regional storage."""
return {
"data": memory_entry,
"_compliance": {
"classification": DataClassification.REGIONAL_PINNED.value,
"jurisdiction": jurisdiction,
"encryption_key_arn": self._resolve_kms_key(jurisdiction)
}
}
def _pseudonymize(self, identifier: str) -> str:
import hashlib
return hashlib.sha256(identifier.encode()).hexdigest()[:16]
def _utc_now(self) -> str:
from datetime import datetime, timezone
return datetime.now(timezone.utc).isoformat()
def _resolve_kms_key(self, jurisdiction: str) -> str:
"""Resolve jurisdiction-specific encryption key."""
key_map = {
"eu": "arn:aws:kms:eu-west-1:...",
"us": "arn:aws:kms:us-east-1:...",
"ap-southeast": "arn:aws:kms:ap-southeast-1:..."
}
return key_map.get(jurisdiction, key_map["us"])Pattern 2: Region-Locked Storage Backends
Agent memory and conversation persistence must be physically isolated per jurisdiction. The storage layer should enforce that data tagged for a specific region can only be written to and read from storage nodes within that region's boundaries.
# regional_storage.py — Jurisdiction-isolated persistence layer
import boto3
from typing import Dict, List, Optional
class RegionalStorageManager:
def __init__(self):
self.regional_clients = {
"us": boto3.client('dynamodb', region_name='us-east-1'),
"eu": boto3.client('dynamodb', region_name='eu-west-1'),
"ap-southeast": boto3.client('dynamodb', region_name='ap-southeast-1')
}
self.regional_buckets = {
"us": "agent-memory-us-east",
"eu": "agent-memory-eu-west",
"ap-southeast": "agent-memory-ap-se"
}
def store_session(self, session_id: str, jurisdiction: str,
data: dict) -> bool:
"""Persist session data to jurisdiction-locked storage."""
client = self.regional_clients.get(jurisdiction)
if not client:
raise ValueError(f"No storage backend for jurisdiction: {jurisdiction}")
client.put_item(
TableName=f"agent_sessions_{jurisdiction}",
Item={
'session_id': {'S': session_id},
'data': {'S': json.dumps(data)},
'jurisdiction_lock': {'S': jurisdiction},
'created_at': {'S': datetime.utcnow().isoformat()}
}
)
return True
def retrieve_session(self, session_id: str,
jurisdiction: str) -> Optional[dict]:
"""Retrieve session data ONLY from the pinned jurisdiction."""
client = self.regional_clients.get(jurisdiction)
if not client:
return None
response = client.get_item(
TableName=f"agent_sessions_{jurisdiction}",
Key={'session_id': {'S': session_id}}
)
if 'Item' in response:
return json.loads(response['Item']['data']['S'])
return None
def enforce_jurisdiction_lock(self, session_id: str,
attempted_jurisdiction: str) -> bool:
"""Verify a session is being accessed from its pinned jurisdiction."""
# Check all regional tables for the session
for jurisdiction, client in self.regional_clients.items():
response = client.get_item(
TableName=f"agent_sessions_{jurisdiction}",
Key={'session_id': {'S': session_id}}
)
if 'Item' in response:
stored_lock = response['Item']['jurisdiction_lock']['S']
return stored_lock == attempted_jurisdiction
return True # Session not found — allow creationPattern 3: Audit Trail for Compliance Verification
Regulators expect demonstrable proof that data residency was maintained. An immutable audit log captures every cross-region data transfer attempt, every jurisdiction pinning decision, and every exception event for later compliance review.
# audit_logger.py — Immutable compliance audit trail
import hashlib
import json
from datetime import datetime, timezone
from typing import Dict, Any
class ComplianceAuditLogger:
def __init__(self, audit_backend_url: str):
self.audit_endpoint = audit_backend_url
self.event_chain = [] # Local chain for hash chaining
def _hash_event(self, event: dict) -> str:
event_json = json.dumps(event, sort_keys=True)
return hashlib.sha256(event_json.encode()).hexdigest()
def log_routing_decision(self, session_id: str, user_ip: str,
detected_jurisdiction: str, routed_region: str,
decision_reason: str):
"""Log every routing decision with justification."""
previous_hash = self.event_chain[-1] if self.event_chain else "genesis"
event = {
"event_type": "routing_decision",
"session_id": hashlib.sha256(session_id.encode()).hexdigest()[:12],
"timestamp": datetime.now(timezone.utc).isoformat(),
"user_ip_prefix": self._mask_ip(user_ip),
"detected_jurisdiction": detected_jurisdiction,
"routed_region": routed_region,
"decision_reason": decision_reason,
"previous_event_hash": previous_hash
}
event_hash = self._hash_event(event)
event["event_hash"] = event_hash
self.event_chain.append(event_hash)
# Persist to append-only audit backend
self._persist_event(event)
return event_hash
def log_data_access(self, session_id: str, region: str,
data_classification: str, access_type: str):
"""Log every access to region-pinned data."""
event = {
"event_type": "data_access",
"session_id": hashlib.sha256(session_id.encode()).hexdigest()[:12],
"timestamp": datetime.now(timezone.utc).isoformat(),
"access_region": region,
"data_classification": data_classification,
"access_type": access_type, # read, write, delete
"previous_event_hash": self.event_chain[-1] if self.event_chain else "genesis"
}
event_hash = self._hash_event(event)
event["event_hash"] = event_hash
self.event_chain.append(event_hash)
self._persist_event(event)
return event_hash
def log_compliance_exception(self, session_id: str,
exception_type: str, details: str,
approving_authority: str):
"""Log any compliance exceptions with approval trail."""
event = {
"event_type": "compliance_exception",
"session_id": hashlib.sha256(session_id.encode()).hexdigest()[:12],
"timestamp": datetime.now(timezone.utc).isoformat(),
"exception_type": exception_type,
"details": details,
"approved_by": approving_authority,
"previous_event_hash": self.event_chain[-1] if self.event_chain else "genesis"
}
event_hash = self._hash_event(event)
event["event_hash"] = event_hash
self.event_chain.append(event_hash)
self._persist_event(event)
def _mask_ip(self, ip_address: str) -> str:
"""Mask IP address to preserve privacy in audit logs."""
parts = ip_address.split('.')
if len(parts) == 4:
return f"{parts[0]}.{parts[1]}.x.x"
return "x.x.x.x"
def _persist_event(self, event: dict):
"""Send event to append-only audit storage."""
import httpx
try:
httpx.post(self.audit_endpoint, json=event, timeout=2.0)
except Exception:
# Buffer locally if audit backend is unavailable
passPutting It All Together — A Complete Multi-Region Agent Deployment
The following example integrates routing, latency optimization, compliance enforcement, and audit logging into a single deployable agent server. This represents a production-grade multi-region agent endpoint that you can deploy to each of your configured regions.
# regional_agent_server.py — Deployable per-region agent instance
import asyncio
import json
from datetime import datetime, timezone
from typing import Dict, Any, Optional
import httpx
from contextlib import asynccontextmanager
# Assume previously defined classes are imported
# from agent_router import LatencyAwareMultiRegionRouter
# from regional_storage import RegionalStorageManager
# from audit_logger import ComplianceAuditLogger
# from data_classification import ComplianceTaggingMiddleware
class RegionalAgentServer:
"""Per-region agent server with full multi-region awareness."""
def __init__(self, region_id: str, config_path: str = "agent-regions.yaml"):
self.region_id = region_id
self.router = LatencyAwareMultiRegionRouter(config_path)
self.storage = RegionalStorageManager()
self.audit = ComplianceAuditLogger(
audit_backend_url="https://audit-trail.internal.ai/events"
)
self.tagger = ComplianceTaggingMiddleware()
self.agent_tools = self._load_region_tools()
def _load_region_tools(self) -> dict:
"""Load tools available in this region (may vary by jurisdiction)."""
# Some tools may be restricted in certain jurisdictions
base_tools = {
"web_search": self._tool_web_search,
"calculator": self._tool_calculator,
"document_retrieval": self._tool_document_retrieval
}
# EU regions may have restricted data export tools
if "eu" in self.region_id:
base_tools.pop("web_search", None)
return base_tools
async def handle_agent_request(self, session_id: str,
user_input: str,
user_ip: str) -> Dict[str, Any]:
"""Main agent request handler with full compliance lifecycle."""
# Step 1: Determine target region via router
target_region = await self.router.route_request(session_id, user_ip)
jurisdiction = self.router.base_router.resolve_user_jurisdiction(user_ip)
# Step 2: Log routing decision
self.audit.log_routing_decision(
session_id=session_id,
user_ip=user_ip,
detected_jurisdiction=jurisdiction,
routed_region=target_region,
decision_reason=f"Compliance-first routing for jurisdiction {jurisdiction}"
)
# Step 3: If this instance isn't the target, proxy the request
if target_region != self.region_id:
return await self._proxy_to_region(target_region, session_id,
user_input, user_ip)
# Step 4: Tag and validate data residency
tagged_input = self.tagger.tag_user_input(
user_id=session_id,
jurisdiction=jurisdiction,
raw_input={"text": user_input}
)
self.audit.log_data_access(
session_id=session_id,
region=self.region_id,
data_classification="REGIONAL_PINNED",
access_type="write"
)
# Step 5: Retrieve session memory from jurisdiction-locked storage
session_memory = self.storage.retrieve_session(session_id, jurisdiction)
if session_memory:
# Verify jurisdiction lock
lock_valid = self.storage.enforce_jurisdiction_lock(
session_id, jurisdiction
)
if not lock_valid:
raise Exception("Jurisdiction lock violation detected")
# Step 6: Execute agent reasoning loop
agent_response = await self._execute_agent_loop(
session_id=session_id,
user_input=tagged_input,
memory=session_memory,
jurisdiction=jurisdiction
)
# Step 7: Persist updated memory to jurisdiction-locked storage
updated_memory = {
"conversation_history": (session_memory or {}).get("conversation_history", []) + [
{"role": "user", "content": user_input},
{"role": "assistant", "content": agent_response["text"]}
],
"last_interaction": datetime.now(timezone.utc).isoformat()
}
self.storage.store_session(session_id, jurisdiction, updated_memory)
# Step 8: Record latency for routing optimization
# (Handled by router.execute_agent_request in production)
return {
"response": agent_response["text"],
"region": self.region_id,
"jurisdiction": jurisdiction,
"tool_calls": agent_response.get("tool_calls", []),
"latency_ms": agent_response.get("latency_ms")
}
async def _execute_agent_loop(self, session_id: str, user_input: dict,
memory: Optional[dict],
jurisdiction: str) -> dict:
"""Execute the agent reasoning and tool calling loop."""
# Build agent context with compliance