Your Agents (any LLM)          ContextSynapse              Your Data
  +------------------+      +--------------------+      +----------------+
  | Claude agent     |----->|                    |<-----| Documents      |
  | GPT agent        |----->|  Shared Brain      |<-----| APIs           |
  | Llama agent      |----->|                    |<-----| Databases      |
  | Custom agent     |----->|  Remember, Recall  |<-----| Files          |
  +------------------+      |  Search, Reason    |      +----------------+
                             +--------------------+

# my_app.py — your vertical application
from contextsynapse.app_factory import create_app
from contextsynapse.security.rbac_framework import get_role_registry
from contextsynapse.security.field_encryption import get_field_encryptor
from contextsynapse.workflow.registry import get_workflow_registry

# 1. Register your roles
reg = get_role_registry()
reg.register_vertical("healthcare", {
    "doctor": {
        "permissions": ["view_patient", "write_notes", "order_tests"],
        "label": "Doctor",
        "global_access": False,
    },
    "nurse": {
        "permissions": ["view_patient", "write_vitals"],
        "label": "Nurse",
        "global_access": False,
    },
    "admin": {
        "permissions": ["view_patient", "manage_users", "view_audit"],
        "label": "Clinic Admin",
        "global_access": True,
    },
})

# 2. Register PII fields
enc = get_field_encryptor()
enc.register("patients", {
    "ssn":   {"mask": "last4", "decrypt_roles": ["admin"]},
    "phone": {"mask": "phone", "decrypt_roles": ["admin", "doctor"]},
})

# 3. Register workflows
get_workflow_registry().register("lab_order", {
    "label": "Lab Order",
    "approvers": ["doctor"],
    "auto_approve_below": 0,
    "timeout_hours": 24,
    "vertical": "healthcare",
})

# 4. Create your app
from fastapi import APIRouter
my_router = APIRouter()

@my_router.get("/patients")
def list_patients():
    return {"patients": []}

app = create_app(
    title="HealthCare App",
    verticals={
        "healthcare": {
            "register": lambda: None,  # registration done above
            "routers": [my_router],
        },
    },
)

# Run: uvicorn my_app:app

pip install contextsynapse

from contextsynapse import ContextSynapse
from contextsynapse.aiql import AIQLExecutor

# Create a graph and executor
db = ContextSynapse()
ex = AIQLExecutor(contextcore=db)

# Build a knowledge graph
ex.execute("CREATE GRAPH company")
ex.execute("USE GRAPH company")
ex.execute('CREATE NODE Person {name: "Alice", role: "Engineer", age: 30}')
ex.execute('CREATE NODE Person {name: "Bob", role: "Manager", age: 42}')
ex.execute('CREATE NODE Project {name: "Atlas", status: "active"}')
ex.execute('CREATE EDGE WORKS_ON FROM Person WHERE name = "Alice" TO Project WHERE name = "Atlas"')
ex.execute('CREATE EDGE MANAGES FROM Person WHERE name = "Bob" TO Project WHERE name = "Atlas"')

# Query it
result = ex.execute("SELECT * FROM Person")
for node in result.get("nodes", []):
    print(node.properties.get("name"), "—", node.properties.get("role"))

# Build LLM-ready context
from contextsynapse.context.hub import ContextHub
hub = ContextHub(system_prompt="You are a project analyst.")
hub.add_nodes(db.get_all_nodes())
messages = hub.to_messages()  # Ready for OpenAI/Anthropic API

python examples/demo.py

# Clone the repo
git clone https://github.com/contextsynapse/contextsynapse.git
cd contextsynapse

# Copy env file and set your keys
cp .env.example .env
# Edit .env — at minimum set CONTEXTSYNAPSE_ADMIN_KEY and CONTEXTSYNAPSE_JWT_SECRET

# Start everything (API + Redis + PostgreSQL)
docker compose up -d

# With the frontend dashboard
docker compose --profile ui up -d

# API:       http://localhost:8000
# Dashboard: http://localhost:3000
# Redis:     localhost:6379
# Postgres:  localhost:5432

# Create a virtual environment
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate

# Install with all extras
pip install -e ".[all]"

# Copy and configure environment
cp .env.example .env
# Edit .env — set at minimum:
#   CONTEXTSYNAPSE_ADMIN_KEY=your-secure-key
#   CONTEXTSYNAPSE_JWT_SECRET=your-jwt-secret

# Start the API server
uvicorn contextsynapse.api.api:app --host 0.0.0.0 --port 8000 --reload

cd frontend
npm install
npm start
# Opens at http://localhost:3000

# Redis — needed for multi-agent coordination, caching, agent registry
docker run -d --name redis -p 6379:6379 redis:7-alpine

# PostgreSQL — needed for user auth, tenant management, audit logs
docker run -d --name postgres -p 5432:5432 \
  -e POSTGRES_DB=contextsynapse \
  -e POSTGRES_USER=contextsynapse \
  -e POSTGRES_PASSWORD=contextsynapse \
  postgres:16-alpine

pip install contextsynapse

from contextsynapse import ContextSynapse
db = ContextSynapse()
# Use directly — no API server required

  +-----------+     +-------+     +-----+     +--------+     +---------+
  | MCP Server|     | REST  |     | A2A |     | Python |     |Framework|
  | (Claude,  |     | API   |     |Proto|     | SDK    |     |Adapters |
  | Copilot)  |     |       |     |     |     |        |     |         |
  +-----------+     +-------+     +-----+     +--------+     +---------+
       |                |            |             |               |
       +--------+-------+-----+------+------+------+------+-------+
                |              |             |             |
                +-------- ContextSynapse (shared brain) ---+

# Stdio transport (default — for Claude Desktop, Claude Code)
python -m contextsynapse.mcp

# SSE transport (for web clients, remote agents)
python -m contextsynapse.mcp --transport sse --port 8100

# With namespace and auth
python -m contextsynapse.mcp --namespace myproject --api-key agent1:secret

{
  "mcpServers": {
    "contextsynapse": {
      "command": "python",
      "args": ["-m", "contextsynapse.mcp"]
    }
  }
}

claude mcp add contextsynapse -- python -m contextsynapse.mcp

uvicorn contextsynapse.api.api:app --host 0.0.0.0 --port 8000

from contextsynapse.a2a import Task, TaskState, TextPart, Message

# Create a task for another agent
task = Task(id="task-1", state=TaskState.SUBMITTED)
task.messages.append(Message(
    role="user",
    parts=[TextPart(text="Analyze TSLA earnings")]
))

pip install contextsynapse-sdk

from contextsynapse_sdk import ContextSynapseClient

# Connect to server
client = ContextSynapseClient("http://localhost:8000", api_key="your-key")

# Register an agent
agent = client.agents.register("my-bot", role="researcher")

# Create a session and work with it
session = client.sessions.create("research-project")
session.ingest({"content": "Tesla Q3 revenue was $25.2B", "type": "Fact"})
results = session.search("Tesla revenue")

# Example: LangChain retriever
from contextsynapse.adapters.langchain import AIContextDBRetriever
retriever = AIContextDBRetriever(db=my_graph, k=10)
docs = retriever.get_relevant_documents("What is TSLA outlook?")

from contextsynapse.llm import get_llm_client
llm = get_llm_client(provider="groq")

# Field-level PII encryption (vertical configures, platform encrypts)
from contextsynapse.security.field_encryption import get_field_encryptor

enc = get_field_encryptor()
enc.register("patients", {
    "ssn":   {"mask": "last4", "decrypt_roles": ["admin"]},
    "phone": {"mask": "phone", "decrypt_roles": ["admin", "doctor"]},
})

record = {"name": "Alice", "ssn": "123-45-6789", "phone": "+1-555-0123"}
admin_view = enc.process_record(record, "patients", user_roles=["admin"])
# → {"name": "Alice", "ssn": "123-45-6789", "phone": "+1-555-0123"}
nurse_view = enc.process_record(record, "patients", user_roles=["nurse"])
# → {"name": "Alice", "ssn": "XXXXX6789", "phone": "XXXXXXX0123"}

# Generic RBAC (verticals register their own roles)
from contextsynapse.security.rbac_framework import get_role_registry

reg = get_role_registry()
reg.register_vertical("myapp", {
    "editor": {"permissions": ["read", "write"], "global_access": False},
    "viewer": {"permissions": ["read"], "global_access": False},
})
perms = reg.effective_permissions(["editor", "viewer"])  # → {"read", "write"}

from contextsynapse.workflow.registry import get_workflow_registry
from contextsynapse.workflow.engine import WorkflowEngine

# Register workflow types
get_workflow_registry().register("approval", {
    "label": "Document Approval",
    "approvers": ["manager"],
    "auto_approve_below": 0,
    "timeout_hours": 48,
    "vertical": "myapp",
})

# Initiate a workflow
engine = WorkflowEngine()
task = engine.initiate("approval", initiated_by="user_123", payload={"doc": "report.pdf"})
# → status: "pending_approval"

# Approve
engine.approve(task["id"], approved_by="manager_456")
# → status: "approved"

from contextsynapse.context.assembly import ContextAssemblyEngine

engine = ContextAssemblyEngine(graph_registry)

# Doctor sees full patient context
result = engine.assemble(
    purpose="patient_review",
    requester={"user_id": "dr_123", "roles": ["doctor"], "scoped_ids": ["patient_abc"]},
    params={"patient_id": "patient_abc"},
)
# → contexts: {patient_record, lab_results, vitals, medications}
# → redacted: []

# Receptionist sees only non-sensitive data
result = engine.assemble(
    purpose="patient_review",
    requester={"user_id": "rec_456", "roles": ["receptionist"]},
    params={"patient_id": "patient_abc"},
)
# → contexts: {patient_record}   (name + appointment only)
# → redacted: ["lab_results", "vitals", "medications"]

contextsynapse/
├── core/           # Graph storage engine (CSR, Redis, LMDB)
├── aiql/           # AIQL query language (grammar, parser, compiler, executor)
├── api/            # FastAPI REST API
├── mcp/            # MCP server for Claude/Copilot
├── context/        # ContextHub — LLM context building + assembled contexts
├── storage/        # Storage backends + WAL + namespace store
├── vector/         # Vector DB integration (NumPy, FAISS, Qdrant, Chroma)
├── ingestion/      # Universal ingestion pipeline (URL, file, API)
├── search/         # Graph-enhanced search + RAG + full-text
├── extraction/     # Entity/fact extraction from documents
├── security/       # RBAC, RLS, encryption, PII detection, audit
├── shield/         # AgentShield — behavioral auth + trust engine
├── governance/     # Data governance layer
├── a2a/            # Agent-to-agent protocol
├── adapters/       # Framework adapters (LangChain, CrewAI, etc.)
├── llm/            # Multi-provider LLM client (12+ providers)
├── rules/          # Rule engine with temporal + aggregate evaluators
├── workspace/      # Git/GitHub workspace connectors
└── plugins/        # Plugin system for vertical applications

frontend/           # React dashboard (graph explorer, playground, sessions)
plugins/            # Domain plugins (installed separately)
verticals/          # Vertical applications (private, not included in package)
sdk/                # Python SDK for REST API access

# Redis backend
CONTEXTSYNAPSE_GRAPH_BACKEND=redis
CONTEXTSYNAPSE_REDIS_URL=redis://localhost:6379

# LMDB backend
CONTEXTSYNAPSE_STORAGE_BACKEND=lmdb

# AWS S3
CONTEXTSYNAPSE_STORAGE_BACKEND=s3
CONTEXTSYNAPSE_S3_BUCKET=my-context-graphs
CONTEXTSYNAPSE_S3_REGION=us-east-1

# Google Cloud Storage
CONTEXTSYNAPSE_STORAGE_BACKEND=gcs
CONTEXTSYNAPSE_GCS_BUCKET=my-context-graphs

# Azure Blob Storage
CONTEXTSYNAPSE_STORAGE_BACKEND=azure
CONTEXTSYNAPSE_AZURE_CONTAINER=context-graphs
CONTEXTSYNAPSE_AZURE_CONNECTION_STRING=DefaultEndpointsProtocol=https;AccountName=...

embeddings:
  provider: "local"          # "local" (sentence-transformers) or "ollama"
  model: "all-MiniLM-L6-v2"
  dimension: 384

CONTEXTSYNAPSE_VECTOR_DB_BACKEND=qdrant  # or: custom, faiss, chroma

from contextsynapse.plugins import VerticalPlugin, Sensor, PluginSchema

class MySensor(Sensor):
    name = "my_data_feed"
    interval_seconds = 300

    async def collect(self, db):
        return [{"id": "item_1", "type": "MyType", "properties": {"value": 42}}]

class MyVertical(VerticalPlugin):
    name = "my_domain"
    version = "0.1.0"

    def sensors(self):
        return [MySensor()]

    def schemas(self):
        return [PluginSchema(name="my_schema", node_types={...})]

[project.entry-points."contextsynapse.plugins"]
my_domain = "my_package.plugin:MyVertical"

pip install contextsynapse-sdk

from contextsynapse_sdk import ContextSynapseClient

client = ContextSynapseClient("http://localhost:8000", api_key="your-key")
client.add_node("person_1", "Person", {"name": "Alice"})
results = client.search("Alice")

from contextsynapse.context.agent_memory import AgentMemory
from contextsynapse.core.registry import GraphRegistry

mem = AgentMemory(GraphRegistry(), namespace="shared_brain")

# Agent stores a fact
mem.remember("agent-1", "Revenue grew 8% YoY", tags=["fact", "revenue"], confidence=0.95)

# Another agent recalls it
memories = mem.recall("agent-2", query="revenue", limit=5)

# Build LLM context from agent's memories
hub = mem.build_context("agent-1", system_prompt="You are an analyst.")
messages = hub.to_messages()  # Ready for any LLM API

from contextsynapse.search.rag import hybrid_retrieve, plain_search, topic_scan

# Fast keyword search (no LLM needed, ~1ms)
results = plain_search(db, "connection pool incident", k=10)

# Topic clustering (no LLM needed)
topics = topic_scan(db, graph_name="knowledge_base", max_topics=10)

# Hybrid retrieval (vector + BM25 + keyword fusion)
top_results, sources = hybrid_retrieve(db, "What caused the outage?", k=5)

from contextsynapse.context.quality import score_node
from contextsynapse.intelligence.freshness import detect_event_date, score_freshness

# Score a node at ingest
score = score_node({"label": "Fact", "properties": {"statement": "Revenue grew 8%", "source_url": "..."}, "edge_count": 4})
# → 85 (high quality: sourced, connected, specific)

# Detect content freshness
result = detect_event_date("Tesla reported Q3 2026 results on August 15")
freshness = score_freshness(result.event_date)
# → freshness="recent", staleness_days=29