Build an OzBrain Shared Memory MCP Server for Cross-Agent Knowledge in 2026

One Brain, Every Agent

OzBrain solves the context drift problem: one structured knowledge base that Claude, ChatGPT, Cursor, and every MCP-compatible agent reads and writes. This FastMCP server wraps OzBrain's API into 6 tools that agents can call directly.

Architecture Overview

┌─────────────────────────────────────────┐
│         AI Agent (Claude/Cursor)          │
│  read_brain │ write_brain │ search │ sync│
└──────────────┬──────────────────────────┘
               │ MCP Protocol (JSON-RPC)
┌──────────────▼──────────────────────────┐
│        OzBrain MCP Server (FastMCP)      │
│  Tools: 6  │  Resources: 3  │  Prompts: 2│
└──────────────┬──────────────────────────┘
               │ REST API v1
┌──────────────▼──────────────────────────┐
│           OzBrain Shared Layer            │
│  Routing Index │ Version Tracker │ Dedup  │
└─────────────────────────────────────────┘

File: src/server.py

import os
import json
from fastmcp import FastMCP
import httpx

mcp = FastMCP(
    name="ozbrain-shared-memory",
    version="1.0.0",
    description="MCP server exposing OzBrain shared memory to AI agents"
)

OZBRAIN_API = os.environ.get("OZBRAIN_API_URL", "https://ozbrain.com/api/v1")
OZBRAIN_KEY = os.environ.get("OZBRAIN_API_KEY", "")

headers = {"Authorization": f"Bearer {OZBRAIN_KEY}", "Content-Type": "application/json"}

@mcp.tool()
async def read_brain(brain_id: str, query: str = "") -> str:
    """Read knowledge items from an OzBrain shared brain."""
    async with httpx.AsyncClient() as client:
        resp = await client.get(f"{OZBRAIN_API}/brains/{brain_id}/read", headers=headers, params={"q": query, "limit": 50})
        resp.raise_for_status()
        data = resp.json()
    return json.dumps({"brain_id": brain_id, "count": len(data.get("items", [])), "items": data.get("items", [])}, indent=2)

@mcp.tool()
async def write_brain(brain_id: str, title: str, content: str, category: str = "general", tags: list[str] = []) -> str:
    """Write a knowledge item to an OzBrain shared brain."""
    async with httpx.AsyncClient() as client:
        resp = await client.post(f"{OZBRAIN_API}/brains/{brain_id}/write", headers=headers, json={"title": title, "content": content, "category": category, "tags": tags})
        resp.raise_for_status()
        data = resp.json()
    return json.dumps({"success": True, "item_id": data.get("id"), "conflict": data.get("conflict")}, indent=2)

@mcp.tool()
async def search_brain(brain_id: str, query: str, top_k: int = 10) -> str:
    """Semantic search across the shared brain."""
    async with httpx.AsyncClient() as client:
        resp = await client.post(f"{OZBRAIN_API}/brains/{brain_id}/search", headers=headers, json={"query": query, "top_k": top_k})
        resp.raise_for_status()
        data = resp.json()
    return json.dumps({"query": query, "count": len(data.get("results", [])), "results": data.get("results", [])}, indent=2)

@mcp.tool()
async def sync_brain(brain_id: str, source_agent: str) -> str:
    """Sync knowledge across all connected agents."""
    async with httpx.AsyncClient() as client:
        resp = await client.post(f"{OZBRAIN_API}/brains/{brain_id}/sync", headers=headers, json={"source_agent": source_agent})
        resp.raise_for_status()
        data = resp.json()
    return json.dumps({"synced": len(data.get("synced_items", [])), "conflicts": len(data.get("conflicts", [])), "details": data}, indent=2)

@mcp.tool()
async def list_brains() -> str:
    """List all accessible OzBrains."""
    async with httpx.AsyncClient() as client:
        resp = await client.get(f"{OZBRAIN_API}/brains", headers=headers)
        resp.raise_for_status()
        data = resp.json()
    return json.dumps({"count": len(data.get("brains", [])), "brains": [{"id": b["id"], "name": b["name"], "items": b.get("item_count", 0)} for b in data.get("brains", [])]}, indent=2)

@mcp.tool()
async def delete_brain_item(brain_id: str, item_id: str) -> str:
    """Delete a knowledge item from the brain."""
    async with httpx.AsyncClient() as client:
        resp = await client.delete(f"{OZBRAIN_API}/brains/{brain_id}/items/{item_id}", headers=headers)
        resp.raise_for_status()
    return json.dumps({"success": True, "deleted": item_id}, indent=2)

@mcp.resource("ozbrain://brains/summary")
async def brains_summary() -> str:
    """Summary of all accessible brains."""
    async with httpx.AsyncClient() as client:
        resp = await client.get(f"{OZBRAIN_API}/brains", headers=headers)
        resp.raise_for_status()
        data = resp.json()
    return json.dumps({"total_brains": len(data.get("brains", [])), "brains": [b["name"] for b in data.get("brains", [])]})

if __name__ == "__main__":
    mcp.run(transport="stdio")

pip install fastmcp httpx && python src/server.py

Production Reality Check

Conflict Resolution: When two agents write to the same item, OzBrain flags the conflict and uses version tracking. The latest-writer-wins strategy with human review for critical items.

By Deepak Bagada, CEO at SaaSNext & Principal AI Architect.

Last tested: August 2026 with Python 3.12, OzBrain v1.0, FastMCP v1.2.0, and MCP 2026-07-28 specification.

Related Architecture & Implementation Resources

• Browse complementary servers and client connectors in the Daily AI World MCP Directory.
• Integrate this tool into multi-agent pipelines with our AI Workflows Blueprints.
• Review frontier LLM capabilities and token metrics on Latest AI News.