Per-User Component Isolation

Keep component drafts private to their owner, share published components, and restrict writes by owner.

With user_isolation=True, a caller can read their own components, shared unowned components, and other users' published components. Publishing shares read, run, and composition access; writes remain owner-scoped. Draft-only components are absent from runnable /agents discovery. Code-defined agents are shared, and the admin scope bypasses user scoping.

The preserved docstring's published Alice example therefore appears in Bob's lists. Use the check below to compare a private draft with a published component.

user_isolation_os.py
"""
Per-User Component Isolation
============================

Demonstrates serving components with per-user isolation, so each caller only
sees and manages the agents, teams, and workflows they created.

Components created over POST /components are stamped with the caller's JWT
subject, and the component, list, and run routes are scoped to that owner.
Admins (agent_os:admin scope) keep seeing everything.

Generate a token for a user with:

    python -c "import jwt, datetime as d; print(jwt.encode({'sub': 'alice', \
'scopes': ['components:read', 'components:write', 'components:delete', \
'agents:read', 'agents:run'], 'exp': d.datetime.now(d.UTC) + \
d.timedelta(hours=1)}, 'my-jwt-secret', algorithm='HS256'))"

Then create a component as that user:

    curl -X POST http://localhost:7777/components \
        -H "Authorization: Bearer $ALICE_TOKEN" -H "Content-Type: application/json" \
        -d '{"name": "Alice Agent", "component_type": "agent", "stage": "published",
             "config": {"name": "Alice Agent", "instructions": "You are Alice private agent."}}'

A token for a different user sees none of it:

    curl http://localhost:7777/components -H "Authorization: Bearer $BOB_TOKEN"
    curl http://localhost:7777/agents -H "Authorization: Bearer $BOB_TOKEN"
"""

from agno.db.postgres import PostgresDb
from agno.os import AgentOS
from agno.os.config import AuthorizationConfig

# ---------------------------------------------------------------------------
# Setup
# ---------------------------------------------------------------------------
db = PostgresDb(db_url="postgresql+psycopg://ai:ai@localhost:5532/ai", id="postgres_db")

# ---------------------------------------------------------------------------
# Create AgentOS App
# ---------------------------------------------------------------------------
# No agents are registered in code: only components created over the API live in
# the database and can be owned. Components passed to AgentOS(agents=[...]) are shared.
agent_os = AgentOS(
    id="user-isolation-os",
    db=db,
    authorization=True,
    authorization_config=AuthorizationConfig(
        verification_keys=["my-jwt-secret"],
        algorithm="HS256",
        user_isolation=True,
    ),
)

app = agent_os.get_app()

# ---------------------------------------------------------------------------
# Run AgentOS App
# ---------------------------------------------------------------------------
if __name__ == "__main__":
    agent_os.serve(app="user_isolation_os:app", reload=True)

Run the Example

Set up your virtual environment

uv venv --python 3.12
source .venv/bin/activate

Install dependencies

uv pip install -U "agno[os]" "psycopg[binary]" httpx pyjwt

Run PgVector

docker run -d \
  -e POSTGRES_DB=ai \
  -e POSTGRES_USER=ai \
  -e POSTGRES_PASSWORD=ai \
  -e PGDATA=/var/lib/postgresql \
  -v pgvolume:/var/lib/postgresql \
  -p 5532:5432 \
  --name pgvector \
  agnohq/pgvector:18

Run the example

Save the code above as user_isolation_os.py, then run:

python user_isolation_os.py

Compare draft and published visibility

Keep the server running. In a second terminal with the same virtual environment, save this as check_visibility.py and run python check_visibility.py. It creates two uniquely named demo records, checks Alice and Bob access, then archives both as Alice. No agent run or model key is needed. The signing key matches this local example.

check_visibility.py
from datetime import datetime, timedelta, timezone
from uuid import uuid4

import httpx
import jwt

def headers(user):
    token = jwt.encode(
        {
            "sub": user,
            "scopes": ["components:read", "components:write", "components:delete", "agents:read"],
            "exp": datetime.now(timezone.utc) + timedelta(hours=1),
        },
        "my-jwt-secret",
        algorithm="HS256",
    )
    return {"Authorization": f"Bearer {token}"}

with httpx.Client(base_url="http://localhost:7777") as client:
    created = []
    try:
        for stage, expected_bob_status in [("draft", 404), ("published", 200)]:
            name = f"Alice {stage} {uuid4().hex[:8]}"
            response = client.post("/components", headers=headers("alice"), json={
                "name": name, "component_type": "agent", "stage": stage,
                "config": {"name": name, "instructions": "Answer questions concisely."},
            })
            response.raise_for_status()
            component_id = response.json()["component_id"]
            created.append(component_id)
            path = f"/components/{component_id}"
            assert client.get(path, headers=headers("alice")).status_code == 200
            assert client.get(path, headers=headers("bob")).status_code == expected_bob_status
            if stage == "published":
                assert client.delete(path, headers=headers("bob")).status_code == 403
            print(stage, "Bob read status:", expected_bob_status)
    finally:
        for component_id in created:
            client.delete(f"/components/{component_id}", headers=headers("alice")).raise_for_status()

Full source: cookbook/93_components/user_isolation_os.py