> ## Documentation Index > Fetch the complete documentation index at: https://docs.agno.com/llms.txt > Use this file to discover all available pages before exploring further. # Per-User Isolation: SurrealDB > Each user gets a private view of one shared knowledge base. Each user gets a private view of one shared knowledge base. Documents uploaded with a user\_id are visible only to that user; documents uploaded without one are shared with everyone. ```python surreal_db.py theme={null} """ Per-User Isolation: SurrealDB ============================= Each user gets a private view of one shared knowledge base. Documents uploaded with a user_id are visible only to that user; documents uploaded without one are shared with everyone. SurrealDB stores the owner in an option user_id field and adds user_id = $scope_user_id OR user_id = NONE to the vector search. - Search as Alice: her chunks plus shared content, never Bob's - Search as Bob: his chunks plus shared content, never Alice's - Search with user_id=None: admin view, sees everything Requirements: - ./cookbook/scripts/run_surrealdb.sh - uv pip install surrealdb - OPENAI_API_KEY """ import asyncio from typing import List from agno.agent import Agent from agno.knowledge.document import Document from agno.knowledge.knowledge import Knowledge from agno.models.openai import OpenAIResponses from agno.vectordb.surrealdb import SurrealDb from surrealdb import AsyncSurreal, Surreal # --------------------------------------------------------------------------- # Setup # --------------------------------------------------------------------------- ALICE_SALARY = "Alice's salary is $180,000. Reviewed annually in March." BOB_SALARY = "Bob's salary is $215,000. Reviewed annually in June." HOLIDAYS = "The company is closed on January 1, July 4, and December 25." SURREAL_URL = "ws://localhost:8000/rpc" SURREAL_USER = "root" SURREAL_PASSWORD = "root" SURREAL_NAMESPACE = "agno" SURREAL_DATABASE = "demo" COLLECTION_NAME = "per_user_isolation_demo" def show(label: str, results: List[Document]) -> None: """Print one search result set.""" print(f"{label} -> {len(results)} results") for d in results: print(f" - {d.content[:80]}") print() # --------------------------------------------------------------------------- # Create Knowledge Base # --------------------------------------------------------------------------- # Sync client for the constructor's exists()/create(), async client for the # searches below - it signs in on the running loop at the top of main(). client = Surreal(SURREAL_URL) client.signin({"username": SURREAL_USER, "password": SURREAL_PASSWORD}) client.use(SURREAL_NAMESPACE, SURREAL_DATABASE) async_client = AsyncSurreal(SURREAL_URL) vector_db = SurrealDb( client=client, async_client=async_client, collection=COLLECTION_NAME ) # Start clean: records left by an earlier run still carry their owner and would # show up as extra results below. if vector_db.exists(): vector_db.drop() vector_db.create() knowledge = Knowledge( name="per_user_demo", description="Per-user RAG isolation demo (SurrealDB)", vector_db=vector_db, ) # --------------------------------------------------------------------------- # Run Demo # --------------------------------------------------------------------------- if __name__ == "__main__": async def main() -> None: await async_client.signin( {"username": SURREAL_USER, "password": SURREAL_PASSWORD} ) await async_client.use(SURREAL_NAMESPACE, SURREAL_DATABASE) await knowledge.ainsert( name="alice_salary", text_content=ALICE_SALARY, user_id="alice", ) await knowledge.ainsert( name="bob_salary", text_content=BOB_SALARY, user_id="bob", ) # The last upload has no user_id, which makes it shared with everyone. await knowledge.ainsert( name="company_holidays", text_content=HOLIDAYS, ) print("\n" + "=" * 60) print("SCOPED SEARCH: three callers, one corpus") print("=" * 60 + "\n") alice_view = await knowledge.asearch(query="salary", user_id="alice") show("Alice (user_id='alice')", alice_view) alice_text = " ".join(d.content for d in alice_view) assert "180,000" in alice_text, "Alice cannot retrieve her own document" assert "January 1" in alice_text, ( "Shared content is unreachable from Alice's scoped view" ) assert "215,000" not in alice_text, ( "Isolation broken: Alice's scoped view leaked Bob's salary" ) bob_view = await knowledge.asearch(query="salary", user_id="bob") show("Bob (user_id='bob')", bob_view) bob_text = " ".join(d.content for d in bob_view) assert "215,000" in bob_text, "Bob cannot retrieve his own document" assert "January 1" in bob_text, ( "Shared content is unreachable from Bob's scoped view" ) assert "180,000" not in bob_text, ( "Isolation broken: Bob's scoped view leaked Alice's salary" ) admin_view = await knowledge.asearch(query="salary", user_id=None) show("Admin (user_id=None)", admin_view) admin_text = " ".join(d.content for d in admin_view) for expected in ("180,000", "215,000", "January 1"): assert expected in admin_text, ( f"Admin view is missing {expected}, it has to see every owner" ) assert all(d.content in admin_text for d in alice_view), ( "Admin view has to be a superset of a scoped user's view" ) print("Alice and Bob each see their own chunk plus the shared one.") print("Admin sees the whole corpus.") print("\n" + "=" * 60) print("AGENT-MEDIATED RETRIEVAL: the owner has to survive the handoff") print("=" * 60 + "\n") alice_agent = Agent( name="Alice's Assistant", model=OpenAIResponses(id="gpt-5.5"), knowledge=knowledge, search_knowledge=True, user_id="alice", instructions=[ "Answer questions using ONLY the knowledge you can retrieve.", "If you don't know, say so - do not invent salary figures.", ], markdown=True, ) response = await alice_agent.arun("What is Bob's salary?") print("Alice's agent on 'What is Bob's salary?':") print(response.content) # Assert on what retrieval returned, not on the model's prose. retrieved = " ".join( item["content"] for ref in (response.references or []) for item in (ref.references or []) if isinstance(item, dict) and item.get("content") ) # Empty references mean the agent never searched, so the check below is vacuous assert retrieved, ( "Retrieval returned no documents, so the isolation check below would pass on nothing" ) assert "215,000" not in retrieved, ( "Isolation broken: Alice's agent retrieved Bob's salary. The owner was " "dropped between the run context and the vector DB, so retrieval ran " "unscoped (user_id=None, the admin view)." ) print("\nisolation holds: Bob's salary never reached Alice's agent") print("\nDone.") asyncio.run(main()) ``` ## Run the Example ```bash theme={null} uv pip install -U agno openai surrealdb ``` ```bash Mac/Linux theme={null} export OPENAI_API_KEY="your_openai_api_key_here" ``` ```bash Windows theme={null} $Env:OPENAI_API_KEY="your_openai_api_key_here" ``` ```bash theme={null} docker run -d --rm --name surrealdb --pull always -p 8000:8000 surrealdb/surrealdb:latest start --user root --pass root ``` Save the code above as `surreal_db.py`, then run: ```bash theme={null} python surreal_db.py ``` Full source: [cookbook/07\_knowledge/04\_advanced/07\_per\_user\_isolation/surreal\_db.py](https://github.com/agno-agi/agno/blob/v3.0.4/cookbook/07_knowledge/04_advanced/07_per_user_isolation/surreal_db.py)