Fusion

DuckDB-powered in-memory analytics engine with LLM tool support.

Fusion connects to PostgreSQL/MySQL databases via Warp REST API, loads data into DuckDB for fast columnar analytics, and exposes 10 tools for LLMs through MCP and OpenAI Function Calling.

Features

• 10 LLM Tools — list_sources, describe_table, query_data, search_data, aggregate_data, create_view, list_views, refresh_view, load_table, cache_stats
• Dual Format — Tool definitions in both MCP (Model Context Protocol) and OpenAI Function Calling format
• 3 Access Layers — MCP Server (stdio), REST API (FastAPI/HTTP), Python SDK
• Query Pushdown — Routes queries directly to source databases when possible, avoiding unnecessary data transfer
• Lazy Loading — Only fetches table data from sources when actually referenced in queries
• SQL Guardrails — Blocks destructive SQL (DROP, DELETE, INSERT) to protect data integrity
• LRU Cache — Query result caching with configurable TTL for millisecond response times
• Materialized Views — Pre-computed aggregation tables with scheduled auto-refresh
• Cross-Source Federation — JOIN across multiple databases (PostgreSQL + MySQL) in a single query
• Auto-Discovery — Automatically discovers all databases and tables from Warp

Architecture

┌─────────────────────────────────────────────────────────────────────────────┐
│  1. Data Source Layer                                                       │
│  ┌──────────────┐    REST     ┌─────────────────┐                          │
│  │ PostgreSQL   │ ──────────► │                 │                          │
│  │ MySQL        │             │ WarpConnector    │  auto-discovery           │
│  └──────────────┘             │ (query pushdown) │  pagination, schema       │
│       Warp REST API           └────────┬────────┘                          │
└────────────────────────────────────────┼──────────────────────────────────┘
                                          │
┌─────────────────────────────────────────▼──────────────────────────────────┐
│  2. DuckDB Core Layer                                                       │
│  ┌──────────────────────────────────────────────────────────────────────┐  │
│  │ OLAPEngine                                                            │  │
│  │  • DuckDB (in-memory, columnar)   • QueryCache (LRU + TTL)            │  │
│  │  • SchemaCatalog (multi-source)   • MaterializedViewManager           │  │
│  │  • FetchStrategy (lazy load)      • SQLGuardrails (SELECT only)       │  │
│  └──────────────────────────────────────────────────────────────────────┘  │
└─────────────────────────────────────────┬──────────────────────────────────┘
                                          │
┌─────────────────────────────────────────▼──────────────────────────────────┐
│  3. LLM Tool Layer                                                          │
│  ┌─────────────┐  ┌──────────────────┐  ┌─────────────────┐                 │
│  │ ToolExecutor│  │ 10 tools         │  │ MCP / REST / SDK│                 │
│  │ (dispatch)  │─►│ query_data, etc. │─►│ → LLM → Result  │                 │
│  └─────────────┘  └──────────────────┘  └─────────────────┘                 │
└─────────────────────────────────────────────────────────────────────────────┘

Installation

pip install -e .

Optional dependencies:

pip install -e ".[mcp]"     # MCP Server support
pip install -e ".[rest]"    # REST API (FastAPI + uvicorn)
pip install -e ".[dev]"     # Development (pytest, ruff, mypy)
pip install -e ".[all]"     # Everything

Quick Start

Python SDK

from fusion import OLAPEngine

engine = OLAPEngine(memory_limit="4GB")
engine.connect_source("mydb", {
    "type": "warp",
    "base_url": "http://localhost:8000",
    "database": "mydb",
})

executor = engine.get_tool_executor()

# Discover available data
sources = executor.list_sources()

# Run an analytical query (auto-loads referenced tables)
result = executor.query_data("SELECT * FROM mydb.orders LIMIT 10")

# Aggregate data
agg = executor.aggregate_data(
    table="mydb.orders",
    group_by="status",
    agg_column="amount",
    agg_func="SUM",
)

# Create a materialized view
executor.create_view(
    name="daily_revenue",
    sql="SELECT status, SUM(amount) as total FROM mydb.orders GROUP BY status",
    refresh="hourly",
)

MCP Server (Claude Desktop / Cursor)

fusion-mcp --warp-url http://localhost:8000 --database mydb

Configure in claude_desktop_config.json:

{
  "mcpServers": {
    "fusion": {
      "command": "fusion-mcp",
      "args": ["--warp-url", "http://localhost:8000", "--database", "mydb"]
    }
  }
}

Auto-discover all databases:

fusion-mcp --warp-url http://localhost:8000 --auto-discover

REST API

fusion-rest --warp-url http://localhost:8000 --auto-discover --port 9000

Swagger UI at http://localhost:9000/docs. Key endpoints:

Endpoint	Method	Description
/sources	GET	List connected sources and tables
/tables/{source.table}/schema	GET	Table schema details
/query	POST	Execute analytical SQL query
/search	POST	Filter search on a table
/aggregate	POST	GROUP BY aggregation
/views	GET/POST	List or create materialized views
/views/{name}/refresh	POST	Refresh a materialized view
/tables/{source.table}/load	POST	Explicitly load a table
/cache/stats	GET	Cache statistics
/tools/{tool_name}	POST	Generic tool dispatch

OpenAI Function Calling

from fusion import get_openai_tools, OLAPEngine

engine = OLAPEngine()
engine.connect_source("mydb", {"type": "warp", "base_url": "http://localhost:8000"})
executor = engine.get_tool_executor()

# Get tool definitions for OpenAI Chat Completions API
tools = get_openai_tools()

# When the LLM makes a tool call:
result = executor.execute("query_data", {"sql": "SELECT ..."})

Tools

Tool	Description
list_sources	Connected sources and tables with row counts
describe_table	Table schema (columns, types, row count)
query_data	Run analytical SQL on DuckDB (SELECT only, max 100 rows)
search_data	Filter search on a table (exact match or LIKE with %)
aggregate_data	GROUP BY aggregation (SUM, AVG, COUNT, MIN, MAX)
create_view	Create a materialized view from a SELECT query
list_views	List materialized views with refresh schedule
refresh_view	Manually refresh a materialized view
load_table	Explicitly load a table from source into DuckDB
cache_stats	Query cache hit rate, entry count, memory usage

Warp Setup

Fusion uses Warp as its data source gateway:

git clone https://github.com/yasinyaman/warp.git
cd warp
docker compose up -d

Warp provides a REST API that federates access to PostgreSQL and MySQL databases.

Project Structure

fusion/
├── __init__.py              # Public API exports
├── engine.py                # OLAPEngine — main orchestration
├── cache.py                 # QueryCache (LRU + TTL)
├── catalog.py               # SchemaCatalog — multi-source metadata
├── guardrails.py            # SQLGuardrails — blocks destructive SQL
├── result.py                # QueryResult — format conversions
├── strategy.py              # FetchStrategy — smart table loading
├── exceptions.py            # Custom exception hierarchy
├── connectors/
│   ├── base.py              # BaseConnector (abstract)
│   └── warp.py              # WarpConnector (Warp REST API)
├── tools/
│   ├── definitions.py       # 10 tool schemas (OpenAI + MCP)
│   ├── executor.py          # ToolExecutor — routes tool calls
│   ├── mcp_server.py        # MCP Server (stdio transport)
│   └── rest_server.py       # REST API Server (FastAPI)
└── views/
    └── materialized.py      # MaterializedViewManager

Development

pip install -e ".[all]"
pytest tests/ -v           # 236 tests
ruff check fusion/         # Lint
python -m demo.demo        # Demo with synthetic data

Requirements

• Python 3.10+
• DuckDB 1.2+
• Warp (data source gateway)

License

Apache 2.0 — see LICENSE for details.