synapse-framework

---

✦ Overview

Synapse is an open-source Python framework engineered for building reliable, production-grade AI assistants, autonomous multi-tool agents, and enterprise backend services.

Modern AI systems face two core engineering challenges:

1. Context Grounding: Pure vector search matches semantic similarity within isolated text chunks, but fails to traverse multi-hop relationships between entities (for example, which services depend on a database managed by a specific engineering squad). Synapse solves this by coupling dense vector search with directed relational knowledge graphs into a unified Dual-Channel Hybrid RAG engine.
2. Tool Execution & Interoperability: AI frameworks often require brittle parsing glue code, non-standardized schemas, and custom network protocols. Synapse standardizes on Anthropic's Model Context Protocol (MCP), allowing Python functions and external servers to be automatically schema-validated, dynamically discovered, and executed by both backend services and desktop IDEs like Claude Desktop and Cursor.

Synapse integrates inference across self-hosted open-weights models (Ollama) and frontier cloud APIs (Google Gemini, OpenAI), pairing runtime execution with a built-in Developer Web Inspector and 3D WebGL Knowledge Graph Visualizer.

---

✦ Architecture & Request Lifecycle

flowchart TD
    subgraph Clients ["Clients & External Hosts"]
        Web["Web Inspector Dashboard (/inspector)"]
        IDE["Cursor / Claude Desktop (Stdio MCP)"]
        API["REST & SSE Clients (POST /chat)"]
    end

    subgraph SynapseApp ["Synapse Core (BaseApp)"]
        Lifespan["FastAPI Lifespan Router"]
        Tracer["Real-Time Execution Tracer"]
        
        subgraph Pipeline ["Reasoning & Orchestration Engine"]
            Dispatcher["Deterministic Tool & Skill Dispatcher"]
            RAG["Dual-Channel Hybrid RAG Engine"]
        end
    end

    subgraph Inference ["Pluggable Inference Engines"]
        OLL["Ollama (Qwen 2.5, Llama 3, Mistral)"]
        GEM["Google Gemini 1.5 / 2.0"]
        OAI["OpenAI (GPT-4o, mini) / vLLM"]
    end

    subgraph DualRAG ["Dual-Channel Retrieval Storage"]
        VectorDB[("Vector Storage: ChromaDB / Qdrant / In-Memory")]
        GraphDB[("Knowledge Graph: NetworkX / Neo4j")]
        ONNX["FastEmbed ONNX Runtime (BGE / MiniLM)"]
    end

    subgraph Tooling ["Multi-MCP Tool Ecosystem"]
        CustomTools["Dynamic Python Tools (makeconnectors)"]
        OAuthTools["Productivity Connectors (Gmail, Notion, Photos)"]
        StdioMCP["External FastMCP / Subprocess Servers"]
    end

    Web --> Lifespan
    API --> Lifespan
    IDE -.->|stdio / SSE| CustomTools

    Lifespan --> Tracer
    Lifespan --> Dispatcher
    Dispatcher <--> RAG
    RAG <--> VectorDB
    RAG <--> GraphDB
    VectorDB --- ONNX
    Dispatcher <--> Inference
    Dispatcher <--> Tooling

---

✦ Installation

Install the core framework from PyPI:

pip install synapse-framework

Or using uv:

uv add synapse-framework

Optional Extras

# MLflow experiment tracking and RAG evaluation metrics
pip install "synapse-framework[mlflow]"

# Distributed Qdrant vector database connector
pip install "synapse-framework[qdrant]"

# Complete suite of database drivers, telemetry, and optional tools
pip install "synapse-framework[all]"

---

✦ Quickstart (10 Lines)

Spin up an operational AI assistant with typed tools and an interactive developer dashboard in ten lines of Python:

from synapse import BaseApp
from synapse.connectors import makeconnectors
from synapse.models import ollama

def get_market_quote(ticker: str) -> dict:
    """Fetches real-time price, trading volume, and currency for a financial asset."""
    return {"ticker": ticker.upper(), "price": 182.45, "volume": "54.2M", "currency": "USD"}

market_tools = makeconnectors(name="market_data", tools=get_market_quote)
app = BaseApp(title="MarketAssistant", model=ollama("qwen2.5:0.5b"), connectors=[market_tools])

if __name__ == "__main__":
    app.run(port=8000)

Access Built-In Developer Interfaces

• Real-Time Web Inspector: http://127.0.0.1:8000/inspector
• Interactive 3D Knowledge Graph: http://127.0.0.1:8000/graph/export
• Interactive OpenAPI Documentation: http://127.0.0.1:8000/docs

Query via Server-Sent Events (SSE)

curl -N -X POST http://127.0.0.1:8000/chat \
  -H "Content-Type: application/json" \
  -d '{"message": "What is the current market quote for AAPL?"}'

event: context
data: {"chunks": [], "graph_triples": ["MARKET_DATA -> CONTAINS -> GET_MARKET_QUOTE"], "tool_called": "get_market_quote"}

event: delta
data: {"token": "AAPL is currently trading at $182.45 USD with a 24-hour volume of 54.2M."}

event: done
data: {}

---

✦ Production Use Cases

Use Case 1: Enterprise Infrastructure & Incident Response Assistant

In production environments, Site Reliability Engineers (SREs) need to query running service metrics and cross-reference documentation, dependencies, and on-call schedules. Vector search alone cannot resolve dependency chains; Synapse traverses graph relations alongside operational runbooks.

from synapse import BaseApp
from synapse.connectors import makeconnectors
from synapse.database import chromadb
from synapse.graph import networkx
from synapse.models import ollama
from synapse.resource import Resource
from synapse.storage import vectorStorage

# 1. Ingest infrastructure documentation and post-mortems
runbooks = Resource("sre_runbooks")
runbooks.add("docs/runbooks/payment_service.md")
runbooks.add("docs/runbooks/redis_cluster.md")

# 2. Define operational telemetry tools
def inspect_service_health(service_name: str) -> dict:
    """Queries Prometheus metrics for error rates, latency percentiles, and pod status."""
    return {
        "service": service_name,
        "status": "degraded",
        "p99_latency_ms": 480.2,
        "error_rate_pct": 3.8,
        "active_pods": 8,
    }

def get_oncall_engineer(team: str) -> dict:
    """Retrieves current primary on-call engineer and escalation contact."""
    return {"team": team, "primary": "alex@enterprise.internal", "escalation": "sre-leads@enterprise.internal"}

ops_connectors = makeconnectors(
    name="infra_controller",
    description="Live infrastructure telemetry and on-call routing",
    tools=[inspect_service_health, get_oncall_engineer],
)

# 3. Configure storage backends and populate relational topology
vector_store = vectorStorage(database=chromadb("data/chroma", embedding="bge-small"))
graph_store = networkx("data/infra_graph.json")

graph_store.add_relation("PaymentService", "DEPENDS_ON", "RedisCluster", source_type="Service", target_type="Service")
graph_store.add_relation("PaymentService", "MAINTAINED_BY", "CheckoutTeam", source_type="Service", target_type="Team")
graph_store.add_relation("RedisCluster", "HOSTED_ON", "K8s-East-Cluster", source_type="Service", target_type="Infrastructure")

# 4. Initialize production assistant
app = BaseApp(
    title="SRE Incident Assistant",
    model=ollama("qwen2.5:0.5b"),
    connectors=[ops_connectors],
    resource=runbooks,
    vector_store=vector_store,
    graph_store=graph_store,
)

if __name__ == "__main__":
    app.run(port=8000)

When an engineer asks: "PaymentService latency is spiking. What does it depend on, and who do we page?"

1. Synapse extracts the entity PaymentService.
2. The Graph channel traverses PaymentService -> DEPENDS_ON -> RedisCluster and PaymentService -> MAINTAINED_BY -> CheckoutTeam.
3. The Vector channel retrieves relevant remediation steps from payment_service.md.
4. The Dispatcher automatically executes inspect_service_health("RedisCluster") and get_oncall_engineer("CheckoutTeam").
5. The model streams a complete diagnosis and escalation plan.

---

Use Case 2: Standalone MCP Tool Server for Claude Desktop & Cursor

When you want to expose custom tools, internal databases, or automation scripts directly to developer IDEs without running an HTTP web server or LLM backend, use conn.run_mcp():

from synapse.connectors import makeconnectors

def execute_safe_sql(query: str) -> dict:
    """Executes read-only SQL queries against the analytical data warehouse."""
    if any(keyword in query.upper() for keyword in ["DROP", "DELETE", "UPDATE", "INSERT", "ALTER"]):
        return {"error": "Write operations are forbidden by security policy."}
    return {
        "status": "success",
        "rows": [
            {"region": "us-east-1", "dau": 1420500, "conversion_rate": 0.042},
            {"region": "eu-central-1", "dau": 985200, "conversion_rate": 0.038},
        ],
    }

def calculate_mrr(active_subscriptions: int, arpu: float) -> dict:
    """Calculates Monthly Recurring Revenue and projected annual run rate."""
    mrr = round(active_subscriptions * arpu, 2)
    return {"mrr_usd": mrr, "arr_usd": round(mrr * 12, 2)}

analytics_connector = makeconnectors(
    name="enterprise_analytics",
    description="Internal data warehouse inspection and revenue analytics",
    tools=[execute_safe_sql, calculate_mrr],
)

if __name__ == "__main__":
    # Boots as a standard Model Context Protocol server over stdio
    analytics_connector.run_mcp(transport="stdio")

Connect to Claude Desktop or Cursor

Add the connector to your claude_desktop_config.json:

{
  "mcpServers": {
    "enterprise_analytics": {
      "command": "python",
      "args": ["/path/to/analytics_server.py"]
    }
  }
}

The tools are now instantly available as native function calls inside Claude Desktop and Cursor.

---

Use Case 3: Autonomous Workspace Assistant with OAuth Connectors

Aggregate personal workspace data across Gmail and Notion with automated synchronization and entity extraction:

from synapse import BaseApp
from synapse.connectors import GmailConnectors, NotionConnectors
from synapse.database import chromadb
from synapse.graph import networkx
from synapse.models import gemini
from synapse.storage import vectorStorage

app = BaseApp(
    title="Executive Workspace Assistant",
    model=gemini(model="gemini-2.0-flash", api_key="AIzaSy..."),
    connectors=[
        GmailConnectors(),    # Google Workspace OAuth connector
        NotionConnectors(),   # Notion Pages & Databases connector
    ],
    vector_store=vectorStorage(database=chromadb("data/workspace_chroma")),
    graph_store=networkx("data/workspace_graph.json"),
)

if __name__ == "__main__":
    app.run(port=8000)

• Synchronize data via background indexing endpoints: POST /sync/gmail and POST /sync/notion.
• Extracted emails and pages are indexed into ChromaDB and parsed into relational triples (e.g. Author -> SENT -> Email, Document -> MENTIONS -> Project).

---

✦ Dual-Channel Hybrid RAG Engine

Synapse couples dense vector search with a directed property graph to eliminate the retrieval failure modes of single-channel systems.

from synapse.database import chromadb
from synapse.graph import networkx
from synapse.rag import HybridRAG
from synapse.storage import vectorStorage

# 1. Initialize persistent backends
vector_db = vectorStorage(database=chromadb("data/chroma", embedding="bge-small"))
graph_db = networkx("data/knowledge_graph.json")

# 2. Add relational topology
graph_db.add_relation("Alice Chen", "DIRECTS", "Core Infrastructure", source_type="Person", target_type="Department")
graph_db.add_relation("Core Infrastructure", "OWNS", "Authentication Service", source_type="Department", target_type="System")
graph_db.add_relation("Authentication Service", "USES", "Vault Secrets Engine", source_type="System", target_type="Tool")

# 3. Query combined knowledge base
rag = HybridRAG(vector_store=vector_db, graph_store=graph_db)
context = rag.retrieve_context("Who directs the team responsible for the Authentication Service?")

print("Detected Entities:", context["entities_detected"])
print("Relational Graph Triples:", context["graph_facts"])

Retrieval Context Injected into Prompts

### RELATIONAL KNOWLEDGE GRAPH FACTS:
- Alice Chen -> DIRECTS -> Core Infrastructure
- Core Infrastructure -> OWNS -> Authentication Service
- Authentication Service -> USES -> Vault Secrets Engine

### RELEVANT DOCUMENT EXCERPTS:
1. [CORE_INFRA - Team Roster]
The Core Infrastructure department manages platform security, identity, and access management.

---

✦ Dynamic Tool Orchestration via MCP

Synapse treats tools as first-class schemas adhering to Anthropic's Model Context Protocol (MCP). When you provide Python functions to makeconnectors, Synapse:

1. Inspects function type hints and generates strict JSON Schema parameters.
2. Extracts docstrings to generate tool descriptions for LLM function calling.
3. Provides input sanitization, runtime execution, and trace logging.

from synapse.connectors import makeconnectors

def calculate_carbon_footprint(cloud_hours: float, instance_type: str = "c5.xlarge") -> dict:
    """Estimates carbon footprint (kg CO2e) for cloud compute infrastructure workloads."""
    base_factor = 0.045 if "xlarge" in instance_type else 0.020
    emissions = round(cloud_hours * base_factor, 3)
    return {"instance_type": instance_type, "hours": cloud_hours, "kg_co2e": emissions}

# makeconnectors accepts single functions, lists, or dictionaries
cloud_tools = makeconnectors(
    name="cloud_optimizer",
    description="Cloud computing cost and sustainability inspection tools",
    tools=[calculate_carbon_footprint],
)

# Export as a FastMCP server or run as standalone MCP
mcp_server = cloud_tools.to_mcp_server()

---

✦ Pluggable Inference Providers

Synapse abstracts LLM inference behind a unified streaming interface with support for local daemons and frontier cloud providers:

from synapse.models import ollama, gemini, openai

# 1. Local open-weights model via Ollama (zero network dependencies)
local_brain = ollama(model="qwen2.5:0.5b", base_url="http://127.0.0.1:11434")

# 2. Frontier reasoning via Google Gemini
gemini_brain = gemini(model="gemini-2.0-flash", api_key="AIzaSy...")

# 3. OpenAI / compatible endpoints (vLLM, Groq, Together)
openai_brain = openai(model="gpt-4o", api_key="sk-...", base_url="https://api.openai.com/v1")

Switch models across any BaseApp without changing tool signatures, prompt templates, or storage backends.

---

✦ Pluggable Storage & Graph Architecture

Synapse provides architectural freedom with zero vendor lock-in. Pick between in-memory testing, local file persistence, or distributed enterprise clusters:

from synapse.database import in_memory, chromadb, qdrant, pgvector
from synapse.graph import in_memory_graph, networkx, neo4j
from synapse.storage import vectorStorage

# Tier 1: Ephemeral In-Memory (Zero disk writes, pure RAM execution)
ram_vector = vectorStorage(database=in_memory(embedding="bge-small"))
ram_graph = in_memory_graph()

# Tier 2: Persistent File-Based (ChromaDB + NetworkX JSON)
file_vector = vectorStorage(database=chromadb("data/chroma", embedding="bge-small"))
file_graph = networkx("data/knowledge_graph.json")

# Tier 3: Distributed Enterprise Cluster (Qdrant + Neo4j)
cluster_vector = vectorStorage(database=qdrant(url="https://qdrant.internal.net", api_key="sk-..."))
cluster_graph = neo4j(uri="bolt://neo4j.internal.net:7687", auth=("neo4j", "password"))

---

✦ Agent Skills & Personas

Skills encapsulate domain knowledge, behavioral instructions, and evaluation guidelines into standardized SKILL.md documents. Registered skills are automatically exposed as callable MCP Prompts.

from synapse import BaseApp
from synapse.skills import load_premade_skill, list_premade_skills, makeskills

# 1. Inspect bundled premade skills
print(list_premade_skills())
# Output: ['rag-analyst', 'knowledge-graph', 'mcp-orchestrator', 'mlflow-evaluator', 'template']

# 2. Load pre-packaged skill sets
rag_analyst = load_premade_skill("rag-analyst")
kg_specialist = load_premade_skill("knowledge-graph")

# 3. Create a specialized domain persona from scratch
security_reviewer = makeskills(
    name="Application Security Auditor",
    description="Audits source code and API configurations for OWASP Top 10 vulnerabilities",
    prompt="""
    You are an expert Application Security Auditor.
    When analyzing code or architectures:
    1. Check for command injection, broken authentication, and SSRF.
    2. Flag insecure cryptography or hardcoded secrets.
    3. Require parameterized queries for all database operations.
    """.strip(),
    tags=["security", "audit", "owasp"],
)

# 4. Attach skills to BaseApp
app = BaseApp(skills=[rag_analyst, kg_specialist, security_reviewer])

---

✦ MLflow Experiment Tracking & Telemetry

Monitor production latency, evaluate retrieval quality, and track prompts, token throughput, and graph snapshots directly within MLflow:

from synapse import BaseApp
from synapse.telemetry import MLflowTracker

# Automated tracking on all /chat turns:
app = BaseApp(
    title="Customer Operations Assistant",
    mlflow_tracking=True,
    mlflow_experiment="customer_ops_production",
)

# Standalone experiment evaluation:
tracker = MLflowTracker(experiment_name="rag_chunk_evaluation")

with tracker.start_run(run_name="bge_small_top4"):
    tracker.log_params({"embedding": "bge-small", "top_k": 4, "model": "qwen2.5:0.5b"})
    tracker.log_metrics({"latency_ms": 32.4, "chunks_retrieved": 4, "graph_triples": 3})
    tracker.log_artifact("data/knowledge_graph.json")

---

✦ Developer Inspector & 3D Knowledge Graph

Synapse includes a built-in browser-based developer dashboard with real-time audit tools.

1. Real-Time Developer Inspector (/inspector)

Audit prompt execution, inspect vector chunk similarity scores, trace tool execution parameters, and inspect relational triples side-by-side:

2. Interactive 3D WebGL Knowledge Graph (/graph/export)

Explore relational topologies, entity clusters, and multi-hop paths in an interactive force-directed WebGL canvas:

3. Connectors & Synchronization Hub (/connectors)

Audit registered tools, OAuth tokens, and execute manual background indexing:

4. Interactive OpenAPI Documentation (/docs)

Explore and test all REST endpoints, SSE streams, sync triggers, and MCP exports:

---

✦ Deployment Paradigms: app.run() vs conn.run_mcp()

Synapse supports two primary execution paradigms:

Architectural Dimension	app.run() (Full Assistant Service)	conn.run_mcp() (Standalone MCP Server)
System Role	Complete AI backend with brain, storage & HTTP server	Specialized tool server exposed over standard I/O
Inference Engine	Required (Ollama, Gemini, or OpenAI)	None needed (Host client acts as the brain)
Protocol Transport	HTTP REST & Server-Sent Events (SSE)	Standard I/O (stdio) or SSE via MCP protocol
Developer UI	Built-in Web Inspector (/inspector) & 3D Graph	Rendered inside client (Claude Desktop / Cursor)
Knowledge Base	Ingests documents into Hybrid RAG	Attached as MCP Resources / Prompts
Primary Target	Web applications, custom frontends, enterprise APIs	IDEs (Cursor), desktop apps (Claude Desktop, LibreChat)
Invocation	app.run(port=8000)	conn.run_mcp(transport="stdio")

---

✦ API Reference Specifications

When you boot BaseApp, Synapse mounts the following endpoints:

Method	Endpoint	Category	Description
GET	/ or /inspector	Inspector	Interactive Developer Inspector dashboard for chat, trace inspection, and tools.
POST	/chat	Chat	Server-Sent Events (SSE) streaming endpoint for agent prompts, tool calls, and grounded replies.
GET	/graph/export	Graph	Full-screen interactive 3D WebGL knowledge graph visualizer.
GET	/graph/data	Graph	JSON payload containing all graph nodes, directed links, color tokens, and stats.
GET	/connectors	Sync	Lists all registered connectors, descriptions, and active tools.
POST	/sync/{name}	Sync	Triggers background synchronization and indexing for a specific connector.
GET	/sync/status	Sync	Returns current synchronization progress, item counts, and timestamps.
GET	/mcp/export	MCP	Exports active tools into Claude Desktop config format and FastMCP manifests.
GET	/skills	Skills	Returns registered agent skills and specialized system instructions.
GET	/resources	Resources	Lists static reference documents and indexed data collections.
GET	/info	Inspector	Overview of running model provider, connector count, and framework version.
GET	/health	System	Health check endpoint returning service readiness and item counts.
GET	/docs	Docs	Interactive Swagger / OpenAPI documentation with try-it-out capabilities.

---

✦ Benchmarks & Performance Metrics

Engineered for high throughput and low latency on development laptops (Apple Silicon, x86_64) and enterprise server environments:

Operation	Standard Implementation	Synapse Architecture	Measured Performance Gain
Embedding Model Startup	PyTorch Transformer (Cold start 4.2s)	LazyFastEmbedEmbeddings ONNX runtime	17.6x faster cold start (<240ms)
Graph Traversal Lookup	Linear scan over nodes $O(N)$	Case-insensitive indexed map $O(1)$	Instant retrieval (<1ms)
In-Memory Vector Search	Full sort $O(M \log M)$ + repeated norms	Pre-computed norms + heapq.nlargest	8.4x higher query throughput
Document Batch Ingestion	Single-item inserts & disk writes	Batched vector ingestion & atomic graph flush	12.5x faster ingestion
Health Check Probe	Blind network probe (1000ms delay)	In-memory key verification + 400ms probe	< 1ms response latency

---

✦ Security & Quality Assurance

Synapse enforces strict quality, coverage, and security standards across the entire codebase:

# Run pytest test suite (31 unit tests)
pytest tests/

# Run AST-level security audit (0 vulnerabilities)
bandit -r synapse

# Check code style, types, and modern standards
ruff check synapse/ tests/

---

✦ License

Synapse is open-source software licensed under the MIT License.