Multi-Tenant Agent Infrastructure: Identity, Isolation, and Rate Limits

Introduction

Building a single-agent prototype is straightforward. Building agent infrastructure that serves multiple customers—each with their own data, models, and usage patterns—is a fundamentally different problem. Multi-tenant agent systems must solve three intertwined challenges simultaneously: identity (knowing who is requesting what), isolation (ensuring tenants never see each other’s data), and rate limits (preventing one tenant from degrading another’s experience).

Whether you’re building a SaaS platform where each customer gets their own AI assistant, or an internal platform serving multiple business units, the architecture decisions you make here will determine whether your system scales or collapses under operational complexity.

Why This Matters

Most AI agent tutorials assume a single user. Production systems serve hundreds or thousands of tenants simultaneously. Without proper multi-tenancy:

• Data leakage becomes a lawsuit, not a bug report
• Cost amplification from one abusive tenant ruins margins for everyone
• Performance degradation from noisy neighbors makes your SLA meaningless
• Compliance failures across tenant boundaries can trigger regulatory action

The Google Gemini API documentation emphasizes that enterprise-grade AI applications must handle authentication, authorization, and resource management at the platform level—delegating these concerns to the application layer creates fragile, unauditable systems.

Architecture Overview

A production multi-tenant agent infrastructure has four layers:

┌─────────────────────────────────────────┐
│            API Gateway / Router          │
│      (Authentication, Routing, TLS)      │
├─────────────────────────────────────────┤
│          Identity & Access Layer         │
│   (Tenant ID, API Keys, Permissions)     │
├─────────────────────────────────────────┤
│         Agent Execution Layer            │
│  (Per-tenant isolation, model routing)   │
├─────────────────────────────────────────┤
│          Resource Management             │
│   (Rate limits, quotas, cost tracking)   │
└─────────────────────────────────────────┘

Identity Management

Every request must carry a verifiable tenant identity. We use API keys with embedded tenant context.

# identity/models.py
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
from typing import Optional
import hashlib
import secrets


class Permission(str, Enum):
    AGENT_EXECUTE = "agent:execute"
    TOOL_USE = "tool:use"
    DATA_READ = "data:read"
    DATA_WRITE = "data:write"
    ADMIN = "admin"


@dataclass
class TenantIdentity:
    tenant_id: str
    name: str
    tier: str  # "free", "pro", "enterprise"
    permissions: set[Permission]
    api_key_hash: str
    created_at: datetime = field(default_factory=datetime.utcnow)
    rate_limit_override: Optional[dict] = None
    
    def has_permission(self, permission: Permission) -> bool:
        return permission in self.permissions or Permission.ADMIN in self.permissions


@dataclass
class AuthenticatedRequest:
    tenant: TenantIdentity
    request_id: str
    timestamp: datetime = field(default_factory=datetime.utcnow)
    ip_address: Optional[str] = None
    user_agent: Optional[str] = None

The API key manager handles creation, validation, and rotation:

# identity/key_manager.py
import hashlib
import secrets
from datetime import datetime
from typing import Optional
from .models import TenantIdentity, Permission


class APIKeyManager:
    """Manages tenant API keys with rotation support."""
    
    def __init__(self, storage_backend):
        self.storage = storage_backend
        self._key_cache: dict[str, TenantIdentity] = {}
    
    def generate_api_key(self, tenant: TenantIdentity) -> str:
        """Generate a new API key for a tenant."""
        raw_key = f"sk_{secrets.token_urlsafe(32)}"
        key_hash = self._hash_key(raw_key)
        
        # Store the hash, never the raw key
        self.storage.store_key_hash(
            tenant_id=tenant.tenant_id,
            key_hash=key_hash,
            created_at=datetime.utcnow(),
        )
        
        self._key_cache[key_hash] = tenant
        return raw_key
    
    def validate_key(self, raw_key: str) -> Optional[TenantIdentity]:
        """Validate an API key and return the tenant identity."""
        key_hash = self._hash_key(raw_key)
        
        # Check cache first
        if key_hash in self._key_cache:
            return self._key_cache[key_hash]
        
        # Lookup from storage
        tenant = self.storage.get_tenant_by_key(key_hash)
        if tenant:
            self._key_cache[key_hash] = tenant
        
        return tenant
    
    def rotate_key(self, tenant_id: str, grace_period_hours: int = 24) -> str:
        """Rotate an API key with grace period for old key."""
        # Generate new key
        new_key = f"sk_{secrets.token_urlsafe(32)}"
        new_hash = self._hash_key(new_key)
        
        # Mark old key for deprecation
        self.storage.deprecate_key(
            tenant_id=tenant_id,
            expires_at=datetime.utcnow() + timedelta(hours=grace_period_hours),
        )
        
        # Store new key
        self.storage.store_key_hash(
            tenant_id=tenant_id,
            key_hash=new_hash,
            created_at=datetime.utcnow(),
        )
        
        return new_key
    
    def _hash_key(self, raw_key: str) -> str:
        return hashlib.sha256(raw_key.encode()).hexdigest()

Tenant Isolation

Isolation ensures one tenant’s data, context, and agent state never leak to another. We enforce isolation at three boundaries: data, execution, and model access.

# isolation/tenant_context.py
from dataclasses import dataclass, field
from typing import Any
import threading


@dataclass
class TenantContext:
    """Thread-safe tenant context with data isolation."""
    tenant_id: str
    workspace_id: str
    data_namespace: str
    model_access: list[str]
    allowed_tools: list[str]
    _local = threading.local()
    
    @classmethod
    def current(cls) -> "TenantContext":
        """Get the current tenant context."""
        ctx = getattr(cls._local, "context", None)
        if ctx is None:
            raise RuntimeError("No tenant context set for this thread")
        return ctx
    
    @classmethod
    def set_current(cls, context: "TenantContext") -> None:
        """Set the current tenant context."""
        cls._local.context = context
    
    def scope_key(self, resource: str) -> str:
        """Scope a resource key to this tenant's namespace."""
        return f"{self.data_namespace}/{resource}"
    
    def validate_tool_access(self, tool_name: str) -> bool:
        """Check if this tenant can use a specific tool."""
        return tool_name in self.allowed_tools or "*" in self.allowed_tools
    
    def validate_model_access(self, model_id: str) -> bool:
        """Check if this tenant can access a specific model."""
        return model_id in self.model_access

The isolation layer wraps data access to enforce namespace boundaries:

# isolation/data_isolator.py
from typing import Any, TypeVar
from .tenant_context import TenantContext

T = TypeVar("T")


class DataIsolator:
    """Ensures data access is scoped to tenant boundaries."""
    
    def __init__(self, storage_backend):
        self.storage = storage_backend
    
    async def read(
        self, 
        resource_type: str, 
        resource_id: str,
    ) -> dict:
        """Read data with tenant scoping."""
        ctx = TenantContext.current()
        scoped_key = ctx.scope_key(f"{resource_type}/{resource_id}")
        
        data = await self.storage.get(scoped_key)
        if data is None:
            raise ResourceNotFoundError(resource_type, resource_id)
        
        # Verify ownership
        if data.get("tenant_id") != ctx.tenant_id:
            raise AccessDeniedError(
                f"Resource {scoped_key} does not belong to tenant {ctx.tenant_id}"
            )
        
        return data
    
    async def write(
        self,
        resource_type: str,
        resource_id: str,
        data: dict,
    ) -> dict:
        """Write data with tenant scoping."""
        ctx = TenantContext.current()
        scoped_key = ctx.scope_key(f"{resource_type}/{resource_id}")
        
        # Inject tenant metadata
        data["tenant_id"] = ctx.tenant_id
        data["workspace_id"] = ctx.workspace_id
        
        await self.storage.put(scoped_key, data)
        return data
    
    async def list_resources(
        self,
        resource_type: str,
        filters: dict = None,
    ) -> list[dict]:
        """List resources scoped to the current tenant."""
        ctx = TenantContext.current()
        prefix = ctx.scope_key(resource_type)
        
        results = await self.storage.list_prefix(prefix)
        
        # Additional filtering
        if filters:
            results = [
                r for r in results 
                if all(r.get(k) == v for k, v in filters.items())
            ]
        
        return results


class ResourceNotFoundError(Exception):
    pass


class AccessDeniedError(Exception):
    pass

Rate Limiting

Rate limits protect against abuse and noisy neighbors. We implement a tiered system: per-tenant limits, per-endpoint limits, and global capacity limits.

# ratelimit/limiter.py
import time
from dataclasses import dataclass
from typing import Optional
from enum import Enum


class LimitType(str, Enum):
    REQUESTS_PER_MINUTE = "rpm"
    TOKENS_PER_DAY = "tpd"
    CONCURRENT_REQUESTS = "ccon"
    COST_PER_MONTH = "cpm"


@dataclass
class RateLimitConfig:
    requests_per_minute: int = 60
    tokens_per_day: int = 1_000_000
    concurrent_requests: int = 10
    cost_per_month_usd: float = 100.0


# Tier-specific defaults
TIER_LIMITS = {
    "free": RateLimitConfig(
        requests_per_minute=10,
        tokens_per_day=50_000,
        concurrent_requests=2,
        cost_per_month_usd=5.0,
    ),
    "pro": RateLimitConfig(
        requests_per_minute=60,
        tokens_per_day=500_000,
        concurrent_requests=10,
        cost_per_month_usd=50.0,
    ),
    "enterprise": RateLimitConfig(
        requests_per_minute=300,
        tokens_per_day=5_000_000,
        concurrent_requests=50,
        cost_per_month_usd=500.0,
    ),
}


class RateLimiter:
    """Token-bucket rate limiter with per-tenant tracking."""
    
    def __init__(self, redis_client):
        self.redis = redis_client
    
    async def check_rate_limit(
        self,
        tenant_id: str,
        limit_type: LimitType,
        cost: int = 1,
    ) -> tuple[bool, dict]:
        """Check if a request is within rate limits."""
        config = await self._get_tenant_config(tenant_id)
        current_usage = await self._get_current_usage(tenant_id, limit_type)
        
        limit_value = getattr(config, limit_type.value)
        allowed = (current_usage + cost) <= limit_value
        
        remaining = max(0, limit_value - current_usage)
        reset_at = self._get_reset_time(limit_type)
        
        if allowed:
            await self._increment_usage(tenant_id, limit_type, cost)
        
        return allowed, {
            "limit": limit_value,
            "remaining": remaining,
            "reset_at": reset_at.isoformat(),
            "retry_after": None if allowed else self._calculate_retry_after(
                tenant_id, limit_type, cost
            ),
        }
    
    async def _get_tenant_config(self, tenant_id: str) -> RateLimitConfig:
        """Get rate limit config for a tenant, with overrides."""
        base_config = await self._get_base_config(tenant_id)
        overrides = await self._get_overrides(tenant_id)
        
        if overrides:
            for key, value in overrides.items():
                setattr(base_config, key, value)
        
        return base_config
    
    async def _get_current_usage(
        self,
        tenant_id: str,
        limit_type: LimitType,
    ) -> int:
        """Get current usage for a specific limit type."""
        key = f"ratelimit:{tenant_id}:{limit_type.value}"
        usage = await self.redis.get(key)
        return int(usage) if usage else 0
    
    async def _increment_usage(
        self,
        tenant_id: str,
        limit_type: LimitType,
        cost: int,
    ) -> None:
        """Increment usage counter."""
        key = f"ratelimit:{tenant_id}:{limit_type.value}"
        pipe = self.redis.pipeline()
        pipe.incrby(key, cost)
        pipe.expire(key, self._get_ttl(limit_type))
        await pipe.execute()
    
    def _get_ttl(self, limit_type: LimitType) -> int:
        """Get TTL for the rate limit window."""
        match limit_type:
            case LimitType.REQUESTS_PER_MINUTE:
                return 60
            case LimitType.TOKENS_PER_DAY:
                return 86400
            case LimitType.COST_PER_MONTH:
                return 2592000  # 30 days
            case _:
                return 3600
    
    def _get_reset_time(self, limit_type: LimitType) -> float:
        """Get when the current window resets."""
        import datetime
        now = time.time()
        ttl = self._get_ttl(limit_type)
        window_start = int(now // ttl) * ttl
        return datetime.datetime.fromtimestamp(window_start + ttl)

Agent Execution Router

The router ties everything together, directing requests to the right agent with proper context and limits.

# router/agent_router.py
from identity.key_manager import APIKeyManager
from isolation.tenant_context import TenantContext
from ratelimit.limiter import RateLimiter, LimitType


class AgentRouter:
    """Routes agent requests with identity, isolation, and rate limiting."""
    
    def __init__(
        self,
        key_manager: APIKeyManager,
        rate_limiter: RateLimiter,
        agent_registry: dict,
    ):
        self.key_manager = key_manager
        self.rate_limiter = rate_limiter
        self.agents = agent_registry
    
    async def handle_request(
        self,
        api_key: str,
        request: dict,
    ) -> dict:
        """Process an agent request with full multi-tenant handling."""
        # 1. Authenticate
        tenant = self.key_manager.validate_key(api_key)
        if tenant is None:
            return {"error": "Invalid API key", "status": 401}
        
        # 2. Check rate limits
        allowed, limit_info = await self.rate_limiter.check_rate_limit(
            tenant_id=tenant.tenant_id,
            limit_type=LimitType.REQUESTS_PER_MINUTE,
        )
        if not allowed:
            return {
                "error": "Rate limit exceeded",
                "status": 429,
                "retry_after": limit_info["retry_after"],
            }
        
        # 3. Set tenant context
        ctx = TenantContext(
            tenant_id=tenant.tenant_id,
            workspace_id=f"ws_{tenant.tenant_id}",
            data_namespace=f"tenant/{tenant.tenant_id}",
            model_access=self._get_model_access(tenant.tier),
            allowed_tools=self._get_tool_access(tenant.tier),
        )
        TenantContext.set_current(ctx)
        
        # 4. Route to appropriate agent
        agent_type = request.get("agent_type", "default")
        agent = self.agents.get(agent_type)
        
        if agent is None:
            return {"error": f"Unknown agent type: {agent_type}", "status": 404}
        
        if not tenant.has_permission("agent:execute"):
            return {"error": "Insufficient permissions", "status": 403}
        
        # 5. Execute with context
        try:
            result = await agent.execute(ctx, request["payload"])
            return {"result": result, "usage": limit_info}
        except Exception as e:
            return {"error": str(e), "status": 500}
    
    def _get_model_access(self, tier: str) -> list[str]:
        """Get allowed models based on tenant tier."""
        match tier:
            case "enterprise":
                return ["gpt-4", "gpt-4-turbo", "claude-3-opus", "gemini-pro"]
            case "pro":
                return ["gpt-4", "gpt-4-turbo", "claude-3-sonnet"]
            case _:
                return ["gpt-3.5-turbo"]
    
    def _get_tool_access(self, tier: str) -> list[str]:
        """Get allowed tools based on tenant tier."""
        match tier:
            case "enterprise":
                return ["*"]
            case "pro":
                return ["search", "calculator", "file_reader", "code_executor"]
            case _:
                return ["search", "calculator"]

Best Practices

1. Tenant ID Propagation: Thread the tenant ID through every layer using context variables. Never rely on implicit state.

2. Key Rotation: Implement automatic key rotation with grace periods. Old keys should continue working briefly during transition.

3. Rate Limit Isolation: Use separate Redis keys or namespaces per tenant. Cross-tenant rate limit contamination is a critical bug.

4. Audit Everything: Log every authenticated request with tenant ID, action, and outcome. Compliance requires this.

5. Graceful Degradation: When a tenant exceeds limits, return clear 429 responses with retry-after headers. Don’t silently drop requests.

Common Pitfalls

• Shared state between tenants: Using global variables or singleton patterns that don’t respect tenant boundaries
• Inadequate key validation: Caching API keys without expiration or rotation support
• Rate limit bypass: Allowing admin operations to skip rate limits entirely
• Missing context propagation: Forgetting to set tenant context on background tasks or async operations

Conclusion

Multi-tenant agent infrastructure requires deliberate architecture at every layer. Identity, isolation, and rate limits aren’t separate concerns—they’re interdependent systems that must work together.

The pattern is clear: authenticate first, enforce limits second, scope execution third. Get this order wrong, and you’ve built a security vulnerability, not a platform.

Next steps:

• Implement the DataIsolator with your production database
• Add OpenTelemetry instrumentation for cross-tenant observability
• Build a tenant dashboard for usage monitoring and limit management
• Review the Gemini API documentation for model-specific multi-tenant considerations