Build a Company Brain with Mem0 Platform and Supabase

-- Employees extend Supabase Auth's users; identity is auth.users.id (uuid)
create table public.employees (
  id     uuid primary key references auth.users (id) on delete cascade,
  name   text not null,
  team   text not null
);

-- The company knowledge the brain ingests. `scope` decides who can recall it.
create table public.knowledge (
  id             bigint generated always as identity primary key,
  scope          text not null,                    -- the shared agent this belongs to: 'org:acme' | 'team:payments'
  content        text not null,
  author         uuid references auth.users (id),   -- who recorded it (their user_id); null for org seed data
  created_at     timestamptz default now(),
  mem0_synced_at timestamptz                        -- null until ingested into the brain
);
create index on public.knowledge (mem0_synced_at, created_at);

-- Seed a little company knowledge to ingest.
insert into public.knowledge (scope, content) values
  ('org:acme', 'We chose Postgres over MongoDB for the core product for strong transactional guarantees and relational joins.'),
  ('org:acme', 'All production deploys go out Tuesday and Thursday; never on Fridays.'),
  ('org:acme', 'Customer data must stay in the EU region for GDPR compliance.'),
  ('org:acme', 'Billing is owned by the Payments team, and Alice is the Payments tech lead.'),
  ('team:payments', 'Stripe is our processor; webhooks are verified with PAYMENTS_WEBHOOK_SECRET.');

mkdir company-brain && cd company-brain
pip install "mem0ai>=2.0.17" supabase requests   # 2.0.17+ for agent_custom_instructions

export MEM0_API_KEY="m0-..."
export SUPABASE_URL="https://<project-ref>.supabase.co"
export SUPABASE_SERVICE_KEY="<service-role-key>"

# brain.py
import os
from mem0 import MemoryClient

client = MemoryClient(api_key=os.environ["MEM0_API_KEY"])

# Steer extraction (project-wide). Runs server-side; no LLM key needed here.
client.project.update(
    # Governs a person's OWN memories (user_id).
    custom_instructions=(
        "Extract the individual's own durable preferences, context, and how they work. "
        "Ignore greetings and one-off chatter."
    ),
    # Governs SHARED memories (agent_id); write them in the third person.
    agent_custom_instructions=(
        "Extract durable company/team knowledge in the third person "
        "(\"The company...\", \"The team...\"): decisions and their rationale, ownership "
        "(who owns what), processes, policies, tooling choices, and gotchas. "
        "Ignore greetings, scheduling, and one-off chatter."
    ),
    custom_categories=[
        {"decision": "Architectural or product decisions and why they were made"},
        {"ownership": "Who owns a system, service, or process"},
        {"policy": "Compliance, security, and process rules"},
        {"tooling": "Tools, services, and how they're configured"},
    ],
)

# Scopes. A person is a user_id; shared brains are agent_ids.
COMPANY = "org:acme"                       # agent_id: company-wide shared brain
def team(name):   return f"team:{name}"    # agent_id: a team's shared brain
def person(uid):  return uid               # user_id: an individual (Supabase auth id)

python -c "import brain; print('brain configured')"

# ingest.py
import os, time, requests
from supabase import create_client
from brain import client, person

sb = create_client(os.environ["SUPABASE_URL"], os.environ["SUPABASE_SERVICE_KEY"])
MEM0_HEADERS = {"Authorization": f"Token {os.environ['MEM0_API_KEY']}"}

def wait_for(event_id, timeout=30):
    """Platform extraction is async; poll the event until it settles."""
    for _ in range(timeout):
        r = requests.get(f"https://api.mem0.ai/v1/event/{event_id}/", headers=MEM0_HEADERS).json()
        if r.get("status") in ("SUCCEEDED", "FAILED"):
            return r["status"]
        time.sleep(1)
    return "TIMEOUT"

# 1. Read knowledge that hasn't been ingested yet
rows = sb.table("knowledge").select("*").is_("mem0_synced_at", "null").execute().data

for row in rows:
    # 2. Add it. agent_id = the shared scope (org/team); user_id = who recorded it.
    #    Mem0 routes shared facts to the agent and personal facts to the individual,
    #    so pass both when there's an author.
    add_kwargs = {
        "agent_id": row["scope"],
        "metadata": {"source": "supabase", "knowledge_id": row["id"]},
    }
    if row["author"]:
        add_kwargs["user_id"] = person(row["author"])
    res = client.add([{"role": "user", "content": row["content"]}], **add_kwargs)
    # 3. Platform returns an event_id; wait for extraction to finish
    event_id = res.get("event_id") if isinstance(res, dict) else None
    if event_id:
        wait_for(event_id)
    # 4. Mark the row synced so we never double-ingest
    sb.table("knowledge").update({"mem0_synced_at": "now()"}).eq("id", row["id"]).execute()

print(f"Ingested {len(rows)} knowledge items into the company brain.")

python ingest.py

Ingested 5 knowledge items into the company brain.

# ask.py
import sys
from brain import client, COMPANY, team, person

def ask(question: str, uid: str | None = None, user_team: str | None = None) -> str:
    scopes = [{"agent_id": COMPANY}]              # company-wide brain
    if user_team:
        scopes.append({"agent_id": team(user_team)})   # the asker's team
    if uid:
        scopes.append({"user_id": person(uid)})        # the asker's own memories
    hits = client.search(
        query=question,
        filters={"OR": scopes},   # OR, never AND (one FK per memory row)
        top_k=5,
        rerank=True,
    )
    return "\n".join(f"- {h['memory']}" for h in hits.get("results", hits))

if __name__ == "__main__":
    print(ask(" ".join(sys.argv[1:]) or "When can we deploy?"))

python ask.py "Why did we pick Postgres, and can I deploy on Friday?"

- The company chose Postgres over MongoDB for strong transactional guarantees and relational joins
- The company's production deploys go out Tuesday and Thursday, never on Fridays

python ask.py "Who should I talk to about billing?"

- Billing is owned by the Payments team
- Alice is the Payments tech lead

client.search(
    query="webhook signing secret",
    filters={"agent_id": "team:payments", "categories": {"in": ["tooling"]}},
    keyword_search=True,     # hybrid keyword + semantic
    rerank=True,
    threshold=0.3,
)

client.search(
    query="compliance rules",
    filters={"AND": [
        {"agent_id": "org:acme"},
        {"categories": {"in": ["policy"]}},
        {"created_at": {"gte": "2026-01-01"}},
    ]},
)

# onboarding.py
from brain import client, person
from ask import ask

# In a real app these come from sb.auth.get_user(jwt) and the employees table.
dana_uid, dana_team = "8f3c...-dana", "payments"

# Dana also tells the brain how *she* works. This is personal, so it goes to her
# user_id, not the shared agent, and stays scoped to her.
client.add(
    [{"role": "user", "content": "I prefer early returns over nested ifs, and I review PRs in the morning."}],
    user_id=person(dana_uid),
)

for q in [
    "Who owns billing and who do I talk to?",   # company (agent) knowledge
    "When are deploys, and are there hard rules?",
    "How do I like to write code?",             # Dana's own (user) knowledge
]:
    print(f"Q: {q}\nA: {ask(q, uid=dana_uid, user_team=dana_team)}\n")

Q: Who owns billing and who do I talk to?
A: - Billing is owned by the Payments team; Alice is the Payments tech lead

Q: When are deploys, and are there hard rules?
A: - The company's production deploys go out Tuesday and Thursday, never on Fridays

Q: How do I like to write code?
A: - User prefers early returns over nested ifs