How do I add multi-user support to self-hosted AI agents?

Mar 16, 2026

Cowrie

Dev @ Bswen

I ran into a problem when trying to share an AI agent with my team. User A asked the agent to remember an API key. User B asked what their API key was. The agent responded with User A’s key.

This is the MEMORY.md conflict problem, and it reveals why most self-hosted AI agents fail in production.

The Problem: File-Based Memory Conflicts

Most early AI agent frameworks started as personal tools. OpenClaw, for example, stores everything in a single MEMORY.md file:

/workspace/MEMORY.md  <-- All users share this file

When multiple users interact with the same agent instance:

User A: "Remember my OpenAI key is sk-proj-xxxxx"
User B: "What's my OpenAI key?"
Agent:  "Your OpenAI key is sk-proj-xxxxx"  (User A's key!)

This is catastrophic for:

• Team environments sharing an agent
• SaaS deployments with customer isolation
• Production systems with compliance requirements

I think the key reason is that early frameworks optimized for single-user experimentation, not multi-user production.

Why This Matters for Production

If you run AI agents for development teams, each developer needs an isolated workspace. For customer support, each team needs access only to their channels. For SaaS products, each customer needs complete data isolation.

For compliance (SOC2, HIPAA, GDPR), user isolation is mandatory.

The Solution: Three Architectural Patterns

To fix this, I found three patterns that work together:

1. Per-user memory isolation - Replace files with database-backed storage
2. RBAC per user per channel - Granular permissions at multiple levels
3. Encrypted secret management - User-specific credential stores

Let me show each pattern.

Pattern 1: Database-Backed Memory Per User

The fix is replacing file-based memory with database storage where every query filters by user_id.

+------------------+          +------------------+
|  OpenClaw        |          |  OpenLobster     |
|  (Single User)  |          |  (Multi-User)    |
+------------------+          +------------------+
| MEMORY.md       |    ->    | Neo4j Graph      |
| (shared file)   |          | user_id on nodes |
+------------------+          +------------------+

With a graph database, each memory node carries the user_id:

(:Memory {user_id: "user_123", content: "..."})
(:Memory {user_id: "user_456", content: "..."})
(:User {id: "user_123"})-[:HAS_MEMORY]->(:MemoryNode)

Here’s a multi-tenant memory system implementation:

memory_system.py

from dataclasses import dataclass
from datetime import datetime
from typing import Optional

@dataclass
class MemoryNode:
    id: str
    user_id: str
    content: str
    embedding: list[float]
    created_at: datetime

class MultiTenantMemory:
    """Memory system with built-in user isolation"""

    def __init__(self, db_connection):
        self.db = db_connection

    async def store(self, user_id: str, content: str) -> MemoryNode:
        """Store memory with automatic user isolation"""
        node_id = self._generate_id()
        await self.db.execute("""
            CREATE (m:Memory {
                id: $id,
                user_id: $user_id,
                content: $content,
                created_at: datetime()
            })
        """, {"id": node_id, "user_id": user_id, "content": content})
        return MemoryNode(id=node_id, user_id=user_id, content=content,
                         embedding=[], created_at=datetime.now())

    async def search(self, user_id: str, query: str) -> list[MemoryNode]:
        """Search memories - ALWAYS filtered by user_id"""
        results = await self.db.fetch_all("""
            MATCH (m:Memory {user_id: $user_id})
            RETURN m ORDER BY m.created_at DESC
        """, {"user_id": user_id})
        return [self._row_to_node(row) for row in results]

    async def delete(self, user_id: str, memory_id: str) -> bool:
        """Delete memory - user_id filter prevents cross-user deletion"""
        result = await self.db.execute("""
            MATCH (m:Memory {id: $id, user_id: $user_id})
            DELETE m
            RETURN count(m) as deleted
        """, {"id": memory_id, "user_id": user_id})
        return result["deleted"] > 0

The critical detail: every database query includes user_id as a mandatory filter.

Pattern 2: RBAC Per User Per Channel

Role-based access control needs three levels:

Level 1: User Authentication
  |
  +-- Who is this user?
  +-- Are they allowed to access the system?

Level 2: Channel Access
  |
  +-- Which channels can this user access?
  +-- User A -> [Telegram_Sales, Discord_Support]
  +-- User B -> [Slack_Dev, WhatsApp_Marketing]

Level 3: Permission Actions
  |
  +-- What can they do in each channel?
  +-- READ, WRITE, ADMIN, MANAGE_SECRETS

Here’s how to implement the permission system:

rbac_manager.py

from dataclasses import dataclass
from enum import Enum
from typing import Optional, Set
from functools import wraps

class Permission(Enum):
    READ = "read"
    WRITE = "write"
    ADMIN = "admin"
    MANAGE_SECRETS = "manage_secrets"
    EXECUTE_SKILLS = "execute_skills"
    CREATE_TASKS = "create_tasks"

@dataclass
class User:
    id: str
    email: str
    role: str  # admin, user, viewer
    is_active: bool = True

class RBACManager:
    """Role-Based Access Control for AI Agents"""

    def __init__(self, db_connection):
        self.db = db_connection
        self.role_permissions = {
            "admin": set(Permission),
            "user": {Permission.READ, Permission.WRITE,
                     Permission.EXECUTE_SKILLS, Permission.CREATE_TASKS},
            "viewer": {Permission.READ}
        }

    async def get_user_permissions(
        self,
        user_id: str,
        channel_id: str
    ) -> Set[Permission]:
        """Get all permissions for user in a channel"""
        # Check channel-specific overrides first
        channel_perms = await self.db.fetch_one("""
            SELECT permissions FROM channel_permissions
            WHERE user_id = $user_id AND channel_id = $channel_id
        """, {"user_id": user_id, "channel_id": channel_id})

        if channel_perms:
            return {Permission(p) for p in channel_perms["permissions"]}

        # Fall back to role-based permissions
        user = await self.get_user(user_id)
        return self.role_permissions.get(user.role, set())

    async def check_permission(
        self,
        user_id: str,
        channel_id: str,
        permission: Permission
    ) -> bool:
        """Check if user has specific permission in channel"""
        permissions = await self.get_user_permissions(user_id, channel_id)
        return permission in permissions

    async def grant_permission(
        self,
        user_id: str,
        channel_id: str,
        permission: Permission
    ) -> None:
        """Grant a permission to user for a channel"""
        current = await self.get_user_permissions(user_id, channel_id)
        current.add(permission)
        await self.db.execute("""
            INSERT INTO channel_permissions (user_id, channel_id, permissions)
            VALUES ($user_id, $channel_id, $permissions)
            ON CONFLICT (user_id, channel_id)
            DO UPDATE SET permissions = $permissions
        """, {"user_id": user_id, "channel_id": channel_id,
              "permissions": [p.value for p in current]})

To enforce permissions on agent actions, use a decorator:

permission_decorator.py

from functools import wraps

def require_permission(permission: Permission):
    """Decorator to enforce permission checks on agent actions"""
    def decorator(func):
        @wraps(func)
        async def wrapper(self, *args, **kwargs):
            user_id = kwargs.get("user_id")
            channel_id = kwargs.get("channel_id")

            if not user_id or not channel_id:
                raise ValueError("user_id and channel_id required")

            has_permission = await self.rbac.check_permission(
                user_id, channel_id, permission
            )

            if not has_permission:
                raise PermissionError(
                    f"User {user_id} lacks {permission.value} permission "
                    f"for channel {channel_id}"
                )

            return await func(self, *args, **kwargs)
        return wrapper
    return decorator

# Usage example
class AgentActions:
    def __init__(self, rbac: RBACManager):
        self.rbac = rbac

    @require_permission(Permission.MANAGE_SECRETS)
    async def store_api_key(self, user_id: str, channel_id: str,
                            key_name: str, key_value: str):
        """Store an API key - requires MANAGE_SECRETS permission"""
        return await self.secrets.store(user_id, key_name, key_value)

    @require_permission(Permission.EXECUTE_SKILLS)
    async def run_skill(self, user_id: str, channel_id: str,
                        skill_name: str, params: dict):
        """Execute a skill - requires EXECUTE_SKILLS permission"""
        return await self.skills.execute(skill_name, params)

Pattern 3: Encrypted Secret Management

Never store API keys in plain text files. Each user needs their own encrypted credential store.

The wrong approach (what I found in OpenClaw):

secrets.yaml

# WRONG: Plain YAML visible to all users
OPENAI_API_KEY: sk-proj-xxxxx
ANTHROPIC_API_KEY: sk-ant-xxxxx
DATABASE_URL: postgres://user:pass@host/db

The right approach:

encrypted_secrets.yaml

# RIGHT: Encrypted with user-specific keys
encrypted_secrets:
  user_123:
    openai_key: AES256:U2FsdGVkX1+vupppZksvRf5pq5g5XjFRlRpz...
    anthropic_key: AES256:U2FsdGVkX1+vupppZksvRf5pq5g5XjF...
  user_456:
    openai_key: AES256:U2FsdGVkX1+differentEncryptedBlob...

Here’s a secrets manager implementation:

secrets_manager.py

from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
import base64
import os

class SecretsManager:
    """Encrypted secrets storage with per-user isolation"""

    def __init__(self, db_connection, master_key: bytes):
        self.db = db_connection
        self.master_key = master_key

    def _derive_user_key(self, user_id: str) -> bytes:
        """Derive a unique encryption key per user"""
        kdf = PBKDF2HMAC(
            algorithm=hashes.SHA256(),
            length=32,
            salt=user_id.encode(),
            iterations=100000,
        )
        key = base64.urlsafe_b64encode(
            kdf.derive(self.master_key + user_id.encode())
        )
        return key

    async def store_secret(
        self,
        user_id: str,
        secret_name: str,
        secret_value: str
    ) -> None:
        """Encrypt and store a secret for a specific user"""
        key = self._derive_user_key(user_id)
        fernet = Fernet(key)
        encrypted = fernet.encrypt(secret_value.encode())

        await self.db.execute("""
            INSERT INTO user_secrets (user_id, secret_name, encrypted_value)
            VALUES ($user_id, $secret_name, $encrypted_value)
            ON CONFLICT (user_id, secret_name)
            DO UPDATE SET encrypted_value = $encrypted_value
        """, {"user_id": user_id, "secret_name": secret_name,
              "encrypted_value": encrypted})

    async def get_secret(
        self,
        user_id: str,
        secret_name: str
    ) -> Optional[str]:
        """Decrypt and retrieve a secret for a specific user"""
        result = await self.db.fetch_one("""
            SELECT encrypted_value FROM user_secrets
            WHERE user_id = $user_id AND secret_name = $secret_name
        """, {"user_id": user_id, "secret_name": secret_name})

        if not result:
            return None

        key = self._derive_user_key(user_id)
        fernet = Fernet(key)
        decrypted = fernet.decrypt(result["encrypted_value"])
        return decrypted.decode()

OpenLobster: A Working Example

OpenLobster demonstrates all three patterns. Here’s how it compares to OpenClaw:

+------------------+------------------+----------------------+
| Component        | OpenClaw         | OpenLobster          |
+------------------+------------------+----------------------+
| Memory Backend   | MEMORY.md files  | Neo4j graph / files  |
| User Model       | None             | Per-user RBAC        |
| Secret Storage   | Plain YAML       | Encrypted backend    |
| Authentication   | Off by default   | On by default        |
| Channel Perms    | None             | Per-user per-channel |
+------------------+------------------+----------------------+

The permission hierarchy in OpenLobster:

User
  |
  +-- Role (admin, user, viewer)
  |
  +-- Channels
  |     |
  |     +-- Telegram_Sales (READ, WRITE)
  |     +-- Discord_Support (READ)
  |     +-- Slack_Dev (ADMIN)
  |
  +-- Permissions
        |
        +-- can_use_mcp_server: true
        +-- can_manage_secrets: false
        +-- can_create_tasks: true

Summary

In this post, I showed how to add multi-user support to self-hosted AI agents. The key point is that you need three architectural pillars:

1. Per-user memory isolation - Replace file-based storage with database-backed memory where every query filters by user_id.

2. RBAC per user per channel - Implement a three-level permission model: authentication, channel access, and action permissions.

3. Encrypted secret management - Each user needs their own encrypted credential store.

These patterns solve the MEMORY.md conflict problem and make AI agents safe for production multi-user environments.

Final Words + More Resources

My intention with this article was to help others share my knowledge and experience. If you want to contact me, you can contact by email: Email me

Here are also the most important links from this article along with some further resources that will help you in this scope:

• 👨‍💻 OpenLobster GitHub Repository
• 👨‍💻 OpenLobster Documentation

Oh, and if you found these resources useful, don’t forget to support me by starring the repo on GitHub!

Comments

Tags:

• ai-agents
• ai
• performance
• security
• automation

How World Monitor Tracks Military Activity: Aircraft, Vessels, and Theater Posture

Is n8n Worth Learning for AI Automation in 2026? My Honest Take