Multi-Tenant AI Agents: Session Isolation Patterns

Why multi-tenancy is hard for AI agents

Multi-tenancy in a traditional web app is well-understood: put a user_id foreign key on your tables, filter every query by it, done. The blast radius of getting it wrong is a data leak — bad, but contained.

Multi-tenancy in an AI agent is harder because the contamination vector is the conversation context itself. An LLM's response is shaped by everything in its context window. If session A's history bleeds into session B's context — even a single message — the model will behave as if it knows things about user B that it learned from user A. That's not just a data leak. It's the model confidently hallucinating cross-user "knowledge" in ways that are subtle and hard to detect.

The other dimension: agents have persistent memory. A standard API endpoint is stateless — each request starts fresh. An agent accumulates context across turns. That state has to live somewhere, and wherever it lives, you need hard guarantees that one user's reads and writes can't touch another's.

Here's the architecture I use for production multi-tenant agents, built on FastAPI, Redis, and PostgreSQL.

Session isolation: the core guarantee

The isolation guarantee needs to be stated explicitly and enforced at the lowest level of the stack, not assumed to emerge from careful coding. Here's the contract:

No method can access another user's session data without an explicit admin override, regardless of what parameters it receives.

This isn't a guideline. It's an invariant enforced in the SessionManager — the single class through which all session access flows.

from dataclasses import dataclass, field
from datetime import datetime
from typing import Any

@dataclass
class AgentSession:
    user_id: str
    created_at: datetime = field(default_factory=datetime.utcnow)
    updated_at: datetime = field(default_factory=datetime.utcnow)
    messages: list[dict] = field(default_factory=list)
    memory: dict[str, Any] = field(default_factory=dict)
    preferences: dict[str, Any] = field(default_factory=dict)
    metadata: dict[str, Any] = field(default_factory=dict)

class SessionManager:
    def __init__(self, cache: RedisCache, store: PostgresStore):
        self._cache = cache
        self._store = store

    async def get(self, user_id: str) -> AgentSession:
        """Get or create a session. Always scoped to user_id."""
        # Check hot cache first
        cached = await self._cache.get(f"session:{user_id}")
        if cached:
            return AgentSession(**cached)

        # Fall through to persistent store
        stored = await self._store.get_session(user_id)
        if stored:
            await self._cache.set(f"session:{user_id}", stored, ttl=3600)
            return AgentSession(**stored)

        # Brand new session
        session = AgentSession(user_id=user_id)
        await self._persist(session)
        return session

    async def update(self, session: AgentSession) -> None:
        """Persist session. Enforces that user_id hasn't changed."""
        session.updated_at = datetime.utcnow()
        await self._cache.set(f"session:{session.user_id}", session.__dict__, ttl=3600)
        await self._store.upsert_session(session)

    async def delete(self, user_id: str) -> None:
        await self._cache.delete(f"session:{user_id}")
        await self._store.delete_session(user_id)

    # Admin-only: explicitly scoped method, requires separate auth check
    async def list_all(self, requester_is_admin: bool) -> list[str]:
        if not requester_is_admin:
            raise PermissionError("admin role required")
        return await self._store.list_user_ids()

The key design choices:

• The cache key is session:{user_id} — you can't retrieve a session without knowing the user_id, and user_id comes from the auth layer, not from the caller's request body.
• Admin operations that cross user boundaries are explicit, separately named methods that require an admin flag — not a parameter on the normal get() method.
• The SessionManager is the only path to session data. No other component queries the cache or store directly.

Redis + PostgreSQL: cache and persist

Sessions need two storage layers with different performance characteristics:

• Redis — hot cache for active sessions. Sub-millisecond reads, TTL-based eviction, no disk I/O. This is where you go on every request during an active conversation.
• PostgreSQL — durable persistence. Survives Redis restarts, supports queries (find all sessions older than 24h, find sessions by user attribute), gives you an audit trail.

import json
import redis.asyncio as redis
from app.config import settings

class RedisCache:
    def __init__(self):
        self._client = redis.from_url(settings.REDIS_URL, decode_responses=True)

    async def get(self, key: str) -> dict | None:
        raw = await self._client.get(key)
        return json.loads(raw) if raw else None

    async def set(self, key: str, value: dict, ttl: int = 3600) -> None:
        await self._client.setex(key, ttl, json.dumps(value, default=str))

    async def delete(self, key: str) -> None:
        await self._client.delete(key)

    async def extend_ttl(self, key: str, ttl: int = 3600) -> None:
        """Call this on every session access to keep active sessions warm."""
        await self._client.expire(key, ttl)

CREATE TABLE agent_sessions (
    user_id      TEXT PRIMARY KEY,
    messages     JSONB NOT NULL DEFAULT '[]',
    memory       JSONB NOT NULL DEFAULT '{}',
    preferences  JSONB NOT NULL DEFAULT '{}',
    metadata     JSONB NOT NULL DEFAULT '{}',
    created_at   TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at   TIMESTAMPTZ NOT NULL DEFAULT now()
);

-- Index for admin cleanup queries (find stale sessions)
CREATE INDEX idx_sessions_updated_at ON agent_sessions (updated_at);

-- Trigger to auto-update updated_at
CREATE OR REPLACE FUNCTION update_updated_at()
RETURNS TRIGGER AS $$
BEGIN NEW.updated_at = now(); RETURN NEW; END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER sessions_updated_at
    BEFORE UPDATE ON agent_sessions
    FOR EACH ROW EXECUTE FUNCTION update_updated_at();

The agent loop

The agent loop is the core execution model: receive a message, load session context, build the LLM prompt, call tools if needed, update session, return response. Every turn goes through this loop.

from app.agent.session import SessionManager, AgentSession
from app.adapters.llm.base import LLMAdapter
from app.tools.registry import ToolRegistry

class AgentLoop:
    def __init__(
        self,
        sessions: SessionManager,
        llm: LLMAdapter,
        tools: ToolRegistry,
    ):
        self._sessions = sessions
        self._llm = llm
        self._tools = tools

    async def run(self, user_id: str, message: str) -> str:
        # 1. Load this user's session — fully isolated
        session = await self._sessions.get(user_id)

        # 2. Append the new user message to history
        session.messages.append({"role": "user", "content": message})

        # 3. Build context: system prompt + conversation history + available tools
        system = self._build_system_prompt(session)
        available_tools = self._tools.get_tools_for_user(user_id)

        # 4. Call LLM — may result in tool calls
        response = await self._llm.complete(
            system=system,
            messages=session.messages,
            tools=available_tools,
        )

        # 5. Execute tool calls if the model requested them
        while response.tool_calls:
            tool_results = []
            for call in response.tool_calls:
                result = await self._tools.execute(
                    tool_name=call.name,
                    params=call.params,
                    user_id=user_id,    # tools receive user_id for their own auth checks
                )
                tool_results.append({"tool": call.name, "result": result})

            # Feed results back to the model
            session.messages.append({"role": "assistant", "tool_calls": response.tool_calls})
            session.messages.append({"role": "tool", "content": tool_results})
            response = await self._llm.complete(
                system=system,
                messages=session.messages,
                tools=available_tools,
            )

        # 6. Append final response to history
        final_text = response.content
        session.messages.append({"role": "assistant", "content": final_text})

        # 7. Trim history to avoid context window overflow
        session.messages = self._trim_history(session.messages, max_turns=20)

        # 8. Persist updated session
        await self._sessions.update(session)

        return final_text

    def _build_system_prompt(self, session: AgentSession) -> str:
        base = "You are an AI assistant. Be helpful and accurate."
        if session.preferences.get("name"):
            base += f" The user's name is {session.preferences['name']}."
        if session.memory:
            base += f"\n\nUser context:\n{self._format_memory(session.memory)}"
        return base

    def _trim_history(self, messages: list[dict], max_turns: int) -> list[dict]:
        """Keep the last N turn pairs. Always preserve the system message."""
        # Each turn = 1 user + 1 assistant message
        max_messages = max_turns * 2
        if len(messages) > max_messages:
            return messages[-max_messages:]
        return messages

Pluggable adapter architecture

The adapter pattern is what makes this deployable to both development and production without code changes. Every external dependency — LLM, memory, auth, storage, messaging — has an abstract base class and at least two concrete implementations: local (for development, no external services required) and cloud (for production).

from abc import ABC, abstractmethod
from dataclasses import dataclass

@dataclass
class LLMResponse:
    content: str
    tool_calls: list[dict] | None = None
    usage: dict | None = None

class LLMAdapter(ABC):
    @abstractmethod
    async def complete(
        self,
        system: str,
        messages: list[dict],
        tools: list[dict] | None = None,
    ) -> LLMResponse:
        ...

import anthropic
from app.adapters.llm.base import LLMAdapter, LLMResponse
from app.config import settings

class AnthropicAdapter(LLMAdapter):
    def __init__(self):
        self._client = anthropic.AsyncAnthropic(api_key=settings.ANTHROPIC_API_KEY)

    async def complete(self, system, messages, tools=None) -> LLMResponse:
        kwargs = dict(
            model=settings.LLM_MODEL,
            max_tokens=1024,
            system=system,
            messages=messages,
        )
        if tools:
            kwargs["tools"] = tools

        response = await self._client.messages.create(**kwargs)

        tool_calls = None
        if response.stop_reason == "tool_use":
            tool_calls = [
                {"name": b.name, "params": b.input, "id": b.id}
                for b in response.content
                if b.type == "tool_use"
            ]

        text = next((b.text for b in response.content if hasattr(b, "text")), "")
        return LLMResponse(content=text, tool_calls=tool_calls)

from openai import AsyncAzureOpenAI
from app.adapters.llm.base import LLMAdapter, LLMResponse
from app.config import settings

class AzureOpenAIAdapter(LLMAdapter):
    def __init__(self):
        self._client = AsyncAzureOpenAI(
            azure_endpoint=settings.AZURE_OPENAI_ENDPOINT,
            api_key=settings.AZURE_OPENAI_API_KEY,
            api_version="2024-02-01",
        )

    async def complete(self, system, messages, tools=None) -> LLMResponse:
        all_messages = [{"role": "system", "content": system}] + messages
        kwargs = dict(model=settings.AZURE_OPENAI_DEPLOYMENT, messages=all_messages)
        if tools:
            kwargs["tools"] = tools

        response = await self._client.chat.completions.create(**kwargs)
        choice = response.choices[0]

        tool_calls = None
        if choice.finish_reason == "tool_calls":
            tool_calls = [
                {"name": tc.function.name, "params": tc.function.arguments, "id": tc.id}
                for tc in choice.message.tool_calls
            ]

        return LLMResponse(content=choice.message.content or "", tool_calls=tool_calls)

Switching the LLM backend is one environment variable: ADAPTER_LLM=anthropic or ADAPTER_LLM=azure_openai. The factory reads it at startup:

from app.config import settings

def build_llm() -> LLMAdapter:
    match settings.ADAPTER_LLM:
        case "anthropic":
            from app.adapters.llm.anthropic import AnthropicAdapter
            return AnthropicAdapter()
        case "azure_openai":
            from app.adapters.llm.azure_openai import AzureOpenAIAdapter
            return AzureOpenAIAdapter()
        case _:
            raise ValueError(f"Unknown LLM adapter: {settings.ADAPTER_LLM}")

def build_auth() -> AuthAdapter:
    match settings.ADAPTER_AUTH:
        case "local":
            from app.adapters.auth.local import LocalAuthAdapter
            return LocalAuthAdapter()
        case "entra":
            from app.adapters.auth.entra import EntraAuthAdapter
            return EntraAuthAdapter()
        case _:
            raise ValueError(f"Unknown auth adapter: {settings.ADAPTER_AUTH}")

Role-based tool access

Tools are the surface where agents affect the world. Role-based access control here is not optional — you need hard enforcement that a regular user can't trigger admin-only tools, regardless of what the LLM decides to call.

from dataclasses import dataclass
from typing import Callable, Any

@dataclass
class Tool:
    name: str
    description: str
    parameters: dict          # JSON Schema
    handler: Callable
    roles: list[str]          # Which roles can use this tool
    schema: dict | None = None  # Cached MCP/function-call schema

class ToolRegistry:
    def __init__(self):
        self._tools: dict[str, Tool] = {}
        self._user_roles: dict[str, list[str]] = {}   # user_id → roles

    def register(self, tool: Tool) -> None:
        self._tools[tool.name] = tool

    def set_user_roles(self, user_id: str, roles: list[str]) -> None:
        self._user_roles[user_id] = roles

    def get_tools_for_user(self, user_id: str) -> list[dict]:
        """Return tool schemas the LLM can call for this user."""
        roles = self._user_roles.get(user_id, ["user"])
        return [
            self._to_schema(t)
            for t in self._tools.values()
            if any(role in t.roles for role in roles)
        ]

    async def execute(self, tool_name: str, params: dict, user_id: str) -> Any:
        tool = self._tools.get(tool_name)
        if not tool:
            raise ValueError(f"Unknown tool: {tool_name}")

        # Enforce role check at execution time — not just at schema time
        roles = self._user_roles.get(user_id, ["user"])
        if not any(role in tool.roles for role in roles):
            raise PermissionError(
                f"User {user_id} lacks required role for tool '{tool_name}'"
            )

        return await tool.handler(params, user_id=user_id)

    def _to_schema(self, tool: Tool) -> dict:
        return {
            "name": tool.name,
            "description": tool.description,
            "input_schema": tool.parameters,
        }

Registering tools:

from app.tools.registry import Tool, ToolRegistry

async def web_search_handler(params: dict, user_id: str) -> dict:
    query = params["query"]
    # DuckDuckGo or configured search provider
    results = await search(query, max_results=5)
    return {"results": [{"title": r.title, "url": r.url, "snippet": r.snippet} for r in results]}

WEB_SEARCH_TOOL = Tool(
    name="web_search",
    description="Search the web for current information. Use for recent events, facts, or anything that may have changed since the model's training cutoff.",
    parameters={
        "type": "object",
        "properties": {
            "query": {"type": "string", "description": "Search query"}
        },
        "required": ["query"],
    },
    handler=web_search_handler,
    roles=["user", "admin"],     # available to all roles
)

async def shell_handler(params: dict, user_id: str) -> dict:
    # Admin-only: actually executes shell commands
    import asyncio
    proc = await asyncio.create_subprocess_shell(
        params["command"],
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    stdout, stderr = await proc.communicate()
    return {"stdout": stdout.decode(), "stderr": stderr.decode(), "returncode": proc.returncode}

SHELL_TOOL = Tool(
    name="shell",
    description="Execute a shell command on the server. Admin only.",
    parameters={
        "type": "object",
        "properties": {
            "command": {"type": "string", "description": "Shell command to run"}
        },
        "required": ["command"],
    },
    handler=shell_handler,
    roles=["admin"],    # admin only — never shown to regular users
)

FastAPI wiring

The API layer is where auth meets sessions. The auth adapter validates the token and returns a user_id; that user_id is the key for everything downstream.

from fastapi import FastAPI, Depends, HTTPException, Header
from pydantic import BaseModel
from app.agent.loop import AgentLoop
from app.adapters.factory import build_auth, build_llm, build_sessions
from app.tools.registry import ToolRegistry

app = FastAPI(title="Enterprise AI Agent")

# Build adapters once at startup
auth = build_auth()
sessions = build_sessions()
llm = build_llm()
tools = ToolRegistry()
# ... register tools ...
agent = AgentLoop(sessions=sessions, llm=llm, tools=tools)

async def get_current_user(authorization: str = Header(...)) -> str:
    """Extract and validate user_id from the Authorization header."""
    if not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Invalid authorization header")
    token = authorization.removeprefix("Bearer ")
    user_id = await auth.validate_token(token)
    if not user_id:
        raise HTTPException(status_code=401, detail="Invalid or expired token")
    return user_id

class ChatRequest(BaseModel):
    message: str

class ChatResponse(BaseModel):
    response: str
    session_id: str  # == user_id; returned for client-side reference

@app.post("/chat", response_model=ChatResponse)
async def chat(
    request: ChatRequest,
    user_id: str = Depends(get_current_user),
):
    # user_id from auth — caller cannot override this
    response_text = await agent.run(user_id=user_id, message=request.message)
    return ChatResponse(response=response_text, session_id=user_id)

@app.get("/session")
async def get_session(user_id: str = Depends(get_current_user)):
    session = await sessions.get(user_id)
    return {
        "user_id": user_id,
        "message_count": len(session.messages),
        "created_at": session.created_at,
        "updated_at": session.updated_at,
    }

@app.delete("/session")
async def delete_session(user_id: str = Depends(get_current_user)):
    await sessions.delete(user_id)
    return {"status": "deleted"}

# Admin endpoints — additional role check inside
@app.post("/admin/cleanup")
async def cleanup_stale_sessions(
    max_age_hours: int = 24,
    user_id: str = Depends(get_current_user),
):
    if not await auth.is_admin(user_id):
        raise HTTPException(status_code=403, detail="Admin role required")
    deleted = await sessions.cleanup_older_than(max_age_hours)
    return {"deleted_sessions": deleted}

What I'd do differently

Use a proper message queue for long agent runs. The current architecture is synchronous: the HTTP request waits until the agent finishes. For quick Q&A that's fine. For agent runs that call multiple tools and take 15-30 seconds, you want an async job queue (Celery, ARQ, or even just a background task with asyncio) so the client can poll for results rather than holding an open connection. SSE or WebSocket for streaming responses is the user-experience upgrade after that.

Rate-limit per user, not per IP. A shared AI agent endpoint is expensive. Users who send thousands of messages will blow your API budget fast. Implement per-user rate limiting in Redis (INCR user:{user_id}:requests:minute with a TTL) and return 429 when they exceed the limit. Fair use policy enforced in code beats it in a terms-of-service document.

Instrument the agent loop, not just the API endpoints. Standard APM tools trace HTTP requests well. They don't trace what happens inside an agent turn — which tools were called, how long LLM inference took, what the token counts were, whether the model triggered a tool call at all. Add structured logging in the agent loop and push it to your observability stack. You need this to debug hallucinations and unexpected tool usage patterns.

Make the adapter factory injectable, not a module-level singleton. Building adapters once at module import time makes testing awkward — you need real Redis and Postgres running to import the module. Use FastAPI's dependency injection instead: adapters are created per-request (or cached at app startup via @app.on_event("startup")) and injected into route handlers. The test suite can swap in mock adapters without touching environment variables.

The Enterprise Agent source is private, but the full architecture here is enough to build from. If you're building something similar and want to compare notes, email me or find me on LinkedIn.

That wraps the series. If there's a topic you'd like to see covered — fine-tuning for industrial data, evaluation frameworks for production RAG, or MuleSoft integration patterns for AI workloads — let me know.