AI Agents

How to Build an AI Agent Cross-Tenant Data Residency Enforcement Layer for Jurisdiction-Compliant Foundation Model Routing in H2 2026

Scott Miller

29 Jul 2026 — 12 min read

As enterprise multi-agent workflows expand across regional boundaries in H2 2026, one architectural challenge has quietly become a board-level concern: where, exactly, does your foundation model inference actually happen? When a sales agent in Frankfurt triggers a chain of sub-agents that each call a different large language model endpoint, the data residency implications cascade in ways that most platform teams simply did not anticipate when they built their first agentic pipeline.

The EU AI Act's enforcement phase, combined with tightened interpretations of GDPR Article 46, Brazil's LGPD amendments, and the expanding patchwork of US state-level AI data laws, means that routing an LLM request to the "wrong" regional endpoint is no longer just a compliance footnote. It is a potential regulatory event. Yet most agent orchestration frameworks treat model endpoints as interchangeable. They are not.

This tutorial walks you through designing and implementing a Cross-Tenant Data Residency Enforcement Layer (CTDREL): a middleware component that sits between your agent orchestration plane and your foundation model endpoints, automatically resolving the correct jurisdiction-compliant route for every inference request at runtime, regardless of how deeply nested the agent call is.

Understanding the Problem Space Before You Build

Before writing a single line of code, it is worth being precise about what "data residency" means in the context of a multi-agent workflow. There are three distinct layers of concern:

Prompt residency: The jurisdiction in which the prompt payload (which may contain PII, confidential business data, or regulated content) is processed and temporarily stored.
Model weight residency: The physical or logical location of the foundation model being invoked. A model served from a US-based data center is subject to US jurisdiction even if the calling application is in the EU.
Inference log residency: Where the request/response logs, token usage records, and audit trails are persisted. This is the layer most teams forget entirely until an audit surfaces it.

A compliant CTDREL must enforce all three simultaneously, per tenant, per request, in real time. That is the architectural challenge. Let's build it.

Step 1: Define Your Tenant Jurisdiction Profile Schema

Every tenant in your multi-tenant platform needs a structured Jurisdiction Profile. This is the source of truth that the enforcement layer will consult at routing time. Store this in a low-latency key-value store (Redis or a regional DynamoDB table works well) rather than a relational database, because you will be reading it on the hot path of every agent invocation.

Here is a recommended schema in JSON:

{
  "tenant_id": "acme-eu-prod",
  "display_name": "Acme Corp (EU Production)",
  "primary_jurisdiction": "EU",
  "allowed_jurisdictions": ["EU", "CH"],
  "prohibited_jurisdictions": ["US", "CN"],
  "data_classification_overrides": {
    "PII": {
      "allowed_jurisdictions": ["EU"],
      "prohibited_jurisdictions": ["US", "CH", "CN"]
    },
    "CONFIDENTIAL": {
      "allowed_jurisdictions": ["EU", "CH"],
      "prohibited_jurisdictions": ["US", "CN"]
    }
  },
  "model_preferences": {
    "EU": ["azure-openai-swedencentral", "mistral-eu-west", "anthropic-claude-eu"],
    "CH": ["azure-openai-switzerland"]
  },
  "audit_log_region": "eu-west-1",
  "enforcement_mode": "strict",
  "fallback_behavior": "reject"
}

A few design decisions worth highlighting here. The enforcement_mode field supports strict (hard block on non-compliant routes), warn (log and alert but allow), and audit (log only). In H2 2026, the vast majority of enterprise tenants operating in regulated industries should be running strict. The fallback_behavior field of reject means that if no compliant endpoint is available, the request fails closed rather than routing to a non-compliant endpoint silently. This is the correct default.

Step 2: Build the Endpoint Registry

Your CTDREL needs a live registry of all foundation model endpoints your platform supports, annotated with their jurisdiction metadata. This registry must be treated as a compliance artifact, not just a configuration file. It should be version-controlled, audited on change, and refreshed on a schedule that accounts for cloud provider region expansions.

{
  "endpoints": [
    {
      "endpoint_id": "azure-openai-swedencentral",
      "provider": "Azure OpenAI",
      "model_family": "gpt-4o",
      "jurisdiction": "EU",
      "sub_region": "SE",
      "data_processing_location": "Sweden Central",
      "certifications": ["ISO27001", "SOC2", "EU-AI-Act-Annex-IV"],
      "latency_p50_ms": 210,
      "supports_streaming": true,
      "max_context_tokens": 128000,
      "endpoint_url": "https://acme-swedencentral.openai.azure.com/",
      "health_check_url": "https://acme-swedencentral.openai.azure.com/health",
      "status": "healthy"
    },
    {
      "endpoint_id": "mistral-eu-west",
      "provider": "Mistral AI",
      "model_family": "mistral-large-3",
      "jurisdiction": "EU",
      "sub_region": "FR",
      "data_processing_location": "Paris",
      "certifications": ["ISO27001", "GDPR-Article46-SCCs"],
      "latency_p50_ms": 185,
      "supports_streaming": true,
      "max_context_tokens": 131072,
      "endpoint_url": "https://api.mistral.ai/v1/",
      "health_check_url": "https://api.mistral.ai/health",
      "status": "healthy"
    }
  ]
}

Critically, include a certifications array. In H2 2026, several jurisdictions now require that endpoints used for regulated workloads carry specific certifications. Your enforcement layer should be capable of filtering on these, not just on geographic location.

Step 3: Implement the Data Classification Interceptor

Before you can route a request, you need to know what is in it. The Data Classification Interceptor sits at the entry point of your agent invocation pipeline and assigns a classification label to the prompt payload. This classification then flows through the entire agent execution graph as immutable metadata.

Here is a Python implementation of the interceptor using a lightweight classifier:

from dataclasses import dataclass, field
from enum import Enum
from typing import Optional
import re

class DataClassification(Enum):
    PUBLIC = "PUBLIC"
    INTERNAL = "INTERNAL"
    CONFIDENTIAL = "CONFIDENTIAL"
    PII = "PII"
    RESTRICTED = "RESTRICTED"

@dataclass
class ClassificationResult:
    classification: DataClassification
    confidence: float
    detected_entities: list[str] = field(default_factory=list)
    requires_human_review: bool = False

class DataClassificationInterceptor:
    
    PII_PATTERNS = [
        (r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', 'EMAIL'),
        (r'\b\d{3}-\d{2}-\d{4}\b', 'SSN'),
        (r'\b[0-9]{13,19}\b', 'PAYMENT_CARD'),
        (r'\b(?:passport|national\s+id|driving\s+licen[sc]e)\b', 'GOVT_ID'),
    ]
    
    CONFIDENTIAL_KEYWORDS = [
        'revenue', 'acquisition', 'merger', 'unreleased', 'internal only',
        'trade secret', 'proprietary', 'confidential', 'attorney-client'
    ]
    
    def __init__(self, use_llm_classifier: bool = True, classifier_endpoint: Optional[str] = None):
        self.use_llm_classifier = use_llm_classifier
        self.classifier_endpoint = classifier_endpoint

    def classify(self, payload: str) -> ClassificationResult:
        detected_entities = []
        
        # Rule-based PII detection (fast path)
        for pattern, entity_type in self.PII_PATTERNS:
            if re.search(pattern, payload, re.IGNORECASE):
                detected_entities.append(entity_type)
        
        if detected_entities:
            return ClassificationResult(
                classification=DataClassification.PII,
                confidence=0.97,
                detected_entities=detected_entities
            )
        
        # Keyword-based confidential detection
        lower_payload = payload.lower()
        matched_keywords = [kw for kw in self.CONFIDENTIAL_KEYWORDS if kw in lower_payload]
        
        if matched_keywords:
            return ClassificationResult(
                classification=DataClassification.CONFIDENTIAL,
                confidence=0.85,
                detected_entities=matched_keywords
            )
        
        # Default to INTERNAL for enterprise agent workflows
        return ClassificationResult(
            classification=DataClassification.INTERNAL,
            confidence=0.75
        )

In production, you will want to augment the rule-based classifier with a dedicated small language model (a fine-tuned 7B or 8B parameter model running locally works well here) for semantic classification of payloads that do not trigger pattern matching. The key architectural principle is that the classification interceptor must never call an external endpoint itself, since doing so would create a circular residency problem.

Step 4: Build the Core Routing Engine

With the tenant profile, endpoint registry, and classification result in hand, the routing engine can now make a compliant endpoint selection. The routing engine implements a priority-ordered resolution algorithm:

Filter endpoints to those whose jurisdiction is in the tenant's allowed_jurisdictions.
Apply data-classification-level overrides from the tenant profile.
Filter out endpoints whose jurisdiction is in the tenant's prohibited_jurisdictions.
Filter to endpoints that are currently healthy.
Apply certification requirements if the tenant's industry mandates them.
Select the optimal endpoint from the remaining candidates using a weighted score of latency and model capability.

from typing import Optional
import logging

logger = logging.getLogger(__name__)

class ResidencyRoutingEngine:

    def __init__(self, endpoint_registry: EndpointRegistry, tenant_store: TenantProfileStore):
        self.endpoint_registry = endpoint_registry
        self.tenant_store = tenant_store

    def resolve_endpoint(
        self,
        tenant_id: str,
        classification: DataClassification,
        required_certifications: list[str] = None,
        preferred_model_family: Optional[str] = None
    ) -> Optional[EndpointConfig]:
        
        profile = self.tenant_store.get(tenant_id)
        if not profile:
            raise TenantProfileNotFoundError(f"No jurisdiction profile found for tenant: {tenant_id}")

        # Determine effective allowed jurisdictions for this classification
        classification_key = classification.value
        if classification_key in profile.data_classification_overrides:
            override = profile.data_classification_overrides[classification_key]
            effective_allowed = override["allowed_jurisdictions"]
            effective_prohibited = override["prohibited_jurisdictions"]
        else:
            effective_allowed = profile.allowed_jurisdictions
            effective_prohibited = profile.prohibited_jurisdictions

        # Fetch and filter candidate endpoints
        candidates = self.endpoint_registry.get_all_healthy()
        
        candidates = [
            ep for ep in candidates
            if ep.jurisdiction in effective_allowed
            and ep.jurisdiction not in effective_prohibited
        ]

        if not candidates:
            if profile.enforcement_mode == "strict":
                raise NoCompliantEndpointError(
                    f"No compliant endpoint available for tenant {tenant_id} "
                    f"with classification {classification_key}"
                )
            elif profile.enforcement_mode == "warn":
                logger.warning(
                    "COMPLIANCE_WARNING: No compliant endpoint found for tenant %s. "
                    "Enforcement mode is WARN. Request will be blocked in strict mode.",
                    tenant_id
                )
                return None
            return None

        # Apply certification filter
        if required_certifications:
            certified_candidates = [
                ep for ep in candidates
                if all(cert in ep.certifications for cert in required_certifications)
            ]
            if certified_candidates:
                candidates = certified_candidates

        # Apply model family preference
        if preferred_model_family:
            preferred = [ep for ep in candidates if ep.model_family == preferred_model_family]
            if preferred:
                candidates = preferred

        # Score and select: lower latency = higher score, weighted by model capability tier
        candidates.sort(key=lambda ep: ep.latency_p50_ms)
        
        selected = candidates[0]
        logger.info(
            "ROUTING_DECISION tenant=%s classification=%s jurisdiction=%s endpoint=%s",
            tenant_id, classification_key, selected.jurisdiction, selected.endpoint_id
        )
        
        return selected

Step 5: Wrap the Routing Engine as Agent-Framework Middleware

The routing engine is only useful if it integrates cleanly with your agent orchestration framework. In H2 2026, the dominant enterprise agent frameworks are built around standardized invocation interfaces, and the cleanest integration point is a middleware wrapper that intercepts every model call before it leaves the orchestration plane.

Here is how to implement this as a drop-in wrapper compatible with the major agent SDK patterns:

from functools import wraps
from typing import Callable, Any
import contextvars

# Context variable that carries tenant and classification metadata
# through the entire async agent execution graph
_residency_context: contextvars.ContextVar[dict] = contextvars.ContextVar(
    'residency_context', default={}
)

class ResidencyEnforcementMiddleware:
    
    def __init__(
        self,
        routing_engine: ResidencyRoutingEngine,
        audit_logger: AuditLogger,
        base_llm_client: Any
    ):
        self.routing_engine = routing_engine
        self.audit_logger = audit_logger
        self.base_client = base_llm_client
        self.classifier = DataClassificationInterceptor()

    def set_tenant_context(self, tenant_id: str) -> None:
        ctx = _residency_context.get({}).copy()
        ctx['tenant_id'] = tenant_id
        _residency_context.set(ctx)

    async def chat_completion(self, messages: list[dict], **kwargs) -> dict:
        ctx = _residency_context.get({})
        tenant_id = ctx.get('tenant_id')
        
        if not tenant_id:
            raise MissingTenantContextError(
                "ResidencyEnforcementMiddleware requires tenant context. "
                "Call set_tenant_context() before invoking agent workflows."
            )

        # Classify the combined prompt payload
        combined_payload = " ".join(m.get("content", "") for m in messages if isinstance(m.get("content"), str))
        classification_result = self.classifier.classify(combined_payload)

        # Resolve compliant endpoint
        endpoint = self.routing_engine.resolve_endpoint(
            tenant_id=tenant_id,
            classification=classification_result.classification,
            preferred_model_family=kwargs.pop("preferred_model_family", None)
        )

        if endpoint is None:
            raise ComplianceBlockedError(
                f"Request blocked by residency enforcement for tenant {tenant_id}."
            )

        # Emit audit log entry BEFORE the request (for forensic completeness)
        audit_entry = self.audit_logger.log_routing_decision(
            tenant_id=tenant_id,
            classification=classification_result.classification.value,
            detected_entities=classification_result.detected_entities,
            selected_endpoint_id=endpoint.endpoint_id,
            selected_jurisdiction=endpoint.jurisdiction
        )

        # Execute the request against the compliant endpoint
        response = await self.base_client.chat_completion(
            endpoint_url=endpoint.endpoint_url,
            messages=messages,
            **kwargs
        )

        # Update audit log with response metadata (token counts, latency)
        self.audit_logger.finalize_entry(audit_entry.entry_id, response.usage)

        return response

Notice the use of Python's contextvars.ContextVar to propagate tenant context through the async execution graph. This is critical for multi-agent workflows where a root agent spawns multiple sub-agents concurrently. Each async task inherits the context from its parent, ensuring that deeply nested agent calls carry the correct tenant identity without requiring you to thread it manually through every function signature.

Step 6: Handle the Sub-Agent Propagation Problem

Here is the part that catches most teams off guard. When a root agent spawns sub-agents (tool-calling agents, retrieval agents, summarization agents), those sub-agents may be instantiated in contexts where the original tenant identity is not automatically available. This is the sub-agent propagation problem, and it is the most common source of residency violations in production multi-agent systems.

The solution is a Residency Propagation Token (RPT): a short-lived, signed token that encodes the tenant's jurisdiction constraints and travels with every agent invocation as a first-class parameter.

import jwt
import time
from dataclasses import dataclass

@dataclass
class ResidencyPropagationToken:
    tenant_id: str
    allowed_jurisdictions: list[str]
    prohibited_jurisdictions: list[str]
    classification_ceiling: str  # Maximum classification level for this execution context
    issued_at: float
    expires_at: float
    root_trace_id: str  # Links all sub-agent calls back to the root workflow for auditing

class RPTManager:
    
    def __init__(self, signing_secret: str, ttl_seconds: int = 300):
        self.signing_secret = signing_secret
        self.ttl_seconds = ttl_seconds

    def issue(self, tenant_id: str, profile: TenantProfile, root_trace_id: str) -> str:
        now = time.time()
        payload = {
            "tenant_id": tenant_id,
            "allowed_jurisdictions": profile.allowed_jurisdictions,
            "prohibited_jurisdictions": profile.prohibited_jurisdictions,
            "classification_ceiling": "PII",  # Most restrictive by default
            "iat": now,
            "exp": now + self.ttl_seconds,
            "root_trace_id": root_trace_id
        }
        return jwt.encode(payload, self.signing_secret, algorithm="HS256")

    def verify_and_decode(self, token: str) -> ResidencyPropagationToken:
        payload = jwt.decode(token, self.signing_secret, algorithms=["HS256"])
        return ResidencyPropagationToken(**{
            k: payload[k] for k in ResidencyPropagationToken.__dataclass_fields__
        })

Every agent invocation in your system should accept an optional rpt parameter. When present, the enforcement middleware validates the RPT and uses its jurisdiction constraints rather than fetching the full tenant profile again, which reduces latency on the hot path. The RPT's root_trace_id field is what allows your audit system to reconstruct the complete call tree for any given workflow execution.

Step 7: Build the Compliance Audit Trail

A residency enforcement layer without a comprehensive audit trail is not actually a compliance control. It is just routing logic. The audit trail is what transforms it into something you can present to a regulator, an auditor, or your own legal team.

Each audit log entry should capture the following, written to a regional audit store that itself respects the tenant's audit_log_region setting:

Timestamp (ISO 8601 with microsecond precision)
Tenant ID and root trace ID
Agent ID and invocation depth (root agent vs. sub-agent level 1, 2, etc.)
Classification result including confidence score and detected entity types (not the actual entity values)
Routing decision: the endpoint selected and why
Endpoints considered and rejected with rejection reasons
Request hash (SHA-256 of the prompt payload, not the payload itself)
Response metadata: token counts, latency, finish reason
Enforcement mode at time of request
Any compliance exceptions raised and how they were handled

Store audit logs in an append-only, tamper-evident log store. In AWS environments, CloudWatch Logs with log group resource policies and CloudTrail integration works well. In Azure environments, use Azure Monitor with immutable storage policies. The key requirement is that audit logs cannot be modified or deleted by application-layer principals, only by designated compliance administrators with MFA-protected break-glass access.

Step 8: Operationalize with Health Checks and Circuit Breakers

A routing engine that selects a compliant endpoint that is currently down is worse than no routing at all, because it may trigger a fallback to a non-compliant endpoint if your circuit breaker logic is not residency-aware. Build your health check loop to be jurisdiction-aware:

import asyncio
import aiohttp
from datetime import datetime

class JurisdictionAwareHealthChecker:
    
    def __init__(self, endpoint_registry: EndpointRegistry, check_interval_seconds: int = 30):
        self.registry = endpoint_registry
        self.check_interval = check_interval_seconds

    async def run(self):
        while True:
            await self._check_all_endpoints()
            await asyncio.sleep(self.check_interval)

    async def _check_all_endpoints(self):
        endpoints = self.registry.get_all()
        tasks = [self._check_endpoint(ep) for ep in endpoints]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        for endpoint, result in zip(endpoints, results):
            if isinstance(result, Exception):
                self.registry.mark_unhealthy(endpoint.endpoint_id, reason=str(result))
            elif result:
                self.registry.mark_healthy(endpoint.endpoint_id)
            else:
                self.registry.mark_unhealthy(endpoint.endpoint_id, reason="Health check returned non-200")

    async def _check_endpoint(self, endpoint: EndpointConfig) -> bool:
        try:
            async with aiohttp.ClientSession() as session:
                async with session.get(
                    endpoint.health_check_url,
                    timeout=aiohttp.ClientTimeout(total=5)
                ) as resp:
                    return resp.status == 200
        except Exception as e:
            raise EndpointHealthCheckError(f"Health check failed for {endpoint.endpoint_id}: {e}")

When a jurisdiction has no healthy endpoints, the enforcement layer should surface this as a JurisdictionDegradedAlert to your on-call team immediately. Do not silently fall back to a non-compliant endpoint. A degraded service is always preferable to a compliance violation in regulated industries.

Deployment Architecture for H2 2026 Enterprise Scale

When you deploy this enforcement layer at enterprise scale, the physical architecture matters as much as the code. Here are the key deployment principles for H2 2026:

Deploy the enforcement layer itself within each jurisdiction it serves. The CTDREL for EU tenants must run in EU infrastructure. An enforcement layer running in a US region that routes EU tenant requests to EU endpoints still processes the routing decision (including the prompt payload for classification) in the US. This defeats the purpose entirely.
Use a global control plane with regional data planes. The tenant profile store, endpoint registry, and RPT signing service can be managed globally, but the classification interceptor and routing engine must execute regionally. Sync jurisdiction profiles to regional replicas with sub-second replication lag.
Treat the enforcement layer as a zero-trust boundary. Every agent framework that wants to invoke a foundation model must go through the CTDREL. There should be no bypass routes, even for internal tooling or development environments. Enforce this with network policy, not just application convention.
Version your jurisdiction profiles. Regulations change. When the EU issues a new guidance document that narrows the definition of compliant processing locations, you need to update tenant profiles and have a clear record of what profile version was active at the time of any given request. Treat profile changes as compliance events with their own audit trail.

Common Pitfalls to Avoid

After walking through the full implementation, here are the failure modes that are most common in real enterprise deployments:

Classifying at the session level, not the request level. A conversation that starts with PUBLIC data may introduce PII three turns in. Classify every request independently, not just the first message in a session.
Forgetting tool call results. When an agent calls a retrieval tool and injects the results into a subsequent LLM prompt, the retrieved content may change the classification of the payload. Run classification on the fully assembled prompt, including injected tool outputs, not just the original user message.
Assuming cloud provider region names map cleanly to legal jurisdictions. "eu-west-1" in AWS is Ireland, which is EU. But "me-south-1" is Bahrain, which is not. Maintain an explicit mapping table; do not infer jurisdiction from region name strings.
Not testing the "no compliant endpoint available" path. This is the path that matters most under regulatory scrutiny. Chaos-engineer it regularly by deliberately marking all endpoints in a jurisdiction as unhealthy and verifying that the system fails closed, not open.

Conclusion

Building a Cross-Tenant Data Residency Enforcement Layer is not a glamorous engineering problem. It sits in the plumbing of your AI platform rather than in the features your users see. But in H2 2026, as enterprise multi-agent workflows scale across regional regulatory boundaries, it is one of the most consequential architectural decisions your team will make.

The implementation described in this tutorial gives you the core components: a jurisdiction profile schema, an endpoint registry, a data classification interceptor, a routing engine, a sub-agent propagation mechanism using RPTs, a compliance audit trail, and a jurisdiction-aware health check system. Each of these components is independently testable and deployable, which means you can adopt them incrementally rather than requiring a big-bang migration of your existing agent infrastructure.

The underlying principle is simple, even if the implementation is detailed: the jurisdiction of inference is a first-class property of every LLM request, not an afterthought. Build it into your agent framework's foundation now, and the regulatory complexity that is coming in 2027 and beyond becomes an operational concern rather than an architectural crisis.