Neocarta

An end-to-end library for building a semantic layer in Neo4j — giving AI agents systemic understanding of how your data is organized, what it means, and where it lives.

Note: This library is not a Neo4j product. It is a Neo4j Labs project supported by the Neo4j field team.

What it is

Neocarta builds a semantic layer in Neo4j from your data sources and serves it to your agents through an MCP server. The graph unifies more than raw schema — it brings together:

Schema metadata — tables, columns, foreign keys, and sample values
Business glossary — terms and categories linked to the columns and tables they describe
Metrics — governed metric definitions and their expressions
Query history — real queries and the tables and columns they touch

…with more on the way. Across a growing set of database types, only the metadata crosses into Neo4j; your data stays in the source.

This gives agents systemic familiarity with the data landscape — what data exists, what it means, how it joins, and which database holds it. Agents use the graph to discover insights, ground their answers, and route queries to the right database, making Text2Query, query routing, and data discovery reliable.

Quickstart

1. Ingest — read your source's schema into the semantic graph (your data stays in the source). Use the Python library or the CLI.

Python — this is the BigQuery connector example:

import os
from google.cloud import bigquery
from neo4j import GraphDatabase
from neocarta import NodeLabel as nl
from neocarta.connectors.bigquery import BigQuerySchemaConnector
from neocarta.enrichment.embeddings import LiteLLMEmbeddingsConnector

driver = GraphDatabase.driver(
    os.getenv("NEO4J_URI"),
    auth=(os.getenv("NEO4J_USERNAME"), os.getenv("NEO4J_PASSWORD")),
)
client = bigquery.Client(project=os.getenv("GCP_PROJECT_ID"))

# Extract, transform, and load BigQuery schema metadata into Neo4j
BigQuerySchemaConnector(
    client=client,
    project_id=os.getenv("GCP_PROJECT_ID"),
    neo4j_driver=driver,
).ingest(dataset_id=os.getenv("BIGQUERY_DATASET_ID"))

# Optional: generate embeddings to turn on semantic table/column search
LiteLLMEmbeddingsConnector(
    neo4j_driver=driver,
    embedding_model="text-embedding-3-small",
).run(node_labels=[nl.DATABASE, nl.SCHEMA, nl.TABLE, nl.COLUMN])

CLI — the same ingest without writing Python (--embeddings is optional):

pip install "neocarta[cli]"
# reads NEO4J_URI / NEO4J_USERNAME / NEO4J_PASSWORD / OPENAI_API_KEY from the environment or a .env file
neocarta bigquery schema --project-id my-proj --dataset-id sales --embeddings

See the Neocarta CLI section for the full command set.

2. Serve — expose the graph to your agent as tools:

pip install "neocarta[mcp]"
# reads NEO4J_URI / NEO4J_USERNAME / NEO4J_PASSWORD from the environment or a .env file
neocarta-mcp           # or, from the unified CLI: neocarta mcp serve

The server gives the agent retrieval tools — list_schemas, list_tables_by_schema, and full-text, vector, or hybrid search over tables, columns, and business terms — each returning a table with its columns, types, example values, and foreign-key references.

3. Use — connect your agent to the neocarta MCP server plus a query-execution tool for your database. The agent searches the graph for relevant tables, follows the foreign keys to build the join, and runs the query:

Which customers placed the largest orders last quarter? → the agent calls get_context_by_table_hybrid_search, finds orders and customers, sees orders.customer_id → customers.id, writes the join, and returns the results.

A complete runnable agent (LangGraph + the MCP + a BigQuery query tool) is in run_agent.py.

The above will result in an agent architecture like below:

---
config:
    layout: dagre
---

graph LR
    
    subgraph GCP["GCP Environment"]
        BQMCP(BigQuery<br>MCP)

        subgraph DataWarehouse["Data Warehouse"]
            BQData[(BigQuery)]
        end
        
        BQMCP <--> BQData
    end

    subgraph Local["Local Environment"]
        Agent("Text2SQL Agent")
        MetadataMCP("Neocarta<br/>MCP")
        
        subgraph Graph["Database"]
            NEO[(Neo4j Graph)]
        end
        
        Agent <--> MetadataMCP
        MetadataMCP <--> NEO
    end
    
    User("User")
    
    subgraph LLM["LLM Service"]
        Model("LLM")
    end
    
    User <--> Agent
    Agent <--> Model
    Agent <--> BQMCP

Embeddings are optional: catalog and full-text tools work from schema alone, and adding embeddings turns on semantic table and column search. See Embeddings and Neocarta MCP.

Installation

pip install neocarta

Requires Python 3.10 or higher and a running Neo4j instance. Options:

Neo4j AuraDB — managed cloud (free tier available)
Neo4j Desktop — local GUI-based instance
Docker image — lightweight local instance

Performance Accelerator

Neocarta has an optional performance enhancement extra. neo4j-rust-ext replaces the pure-Python serialisation layer of the Neo4j Python Driver with a compiled Rust extension, delivering 60–90% faster throughput for bulk loads — relevant for any connector loading large schemas or datasets.

Note: This requires Python >=3.11

pip install neocarta[performance]

Metadata Graph

The metadata graph has the following schema. All connectors must convert their schema information to this graph schema to be compatible with the provided MCP server and ingestion tooling.

---
config:
    layout: elk
---
graph LR
%% Nodes
Database("Database<br/>id: STRING | KEY<br/>name: STRING<br/>description: STRING<br/>embedding: VECTOR")
Schema("Schema<br/>id: STRING | KEY<br/>name: STRING<br/>description: STRING<br/>embedding: VECTOR")
Table("Table<br/>id: STRING | KEY<br/>name: STRING<br/>description: STRING<br/>embedding: VECTOR")
Column("Column<br/>id: STRING | KEY<br/>name: STRING<br/>description: STRING<br/>embedding: VECTOR<br/>type: STRING<br/>nullable: BOOLEAN<br/>isPrimaryKey: BOOLEAN<br/>isForeignKey: BOOLEAN")
Value("Value<br/>id: STRING | KEY<br/>value: STRING")

%% Relationships
Database -->|HAS_SCHEMA| Schema
Schema -->|HAS_TABLE| Table
Table -->|HAS_COLUMN| Column
Column -->|REFERENCES| Column
Column -->|HAS_VALUE| Value


%% Styling
classDef node_0_color fill:#e3f2fd,stroke:#1976d2,stroke-width:3px,color:#000,font-size:12px
class Database node_0_color

classDef node_1_color fill:#fff9c4,stroke:#f57f17,stroke-width:3px,color:#000,font-size:12px
class Schema node_1_color

classDef node_2_color fill:#f3e5f5,stroke:#7b1fa2,stroke-width:3px,color:#000,font-size:12px
class Table node_2_color

classDef node_3_color fill:#e8f5e8,stroke:#388e3c,stroke-width:3px,color:#000,font-size:12px
class Column node_3_color

classDef node_4_color fill:#fff3e0,stroke:#f57c00,stroke-width:3px,color:#000,font-size:12px
class Value node_4_color

Nodes

Database
Schema
Table
Column
Value

Relationships

(:Database)-[:HAS_SCHEMA]->(:Schema)
(:Schema)-[:HAS_TABLE]->(:Table)
(:Table)-[:HAS_COLUMN]->(:Column)
(:Column)-[:HAS_VALUE]->(:Value)
(:Column)-[:REFERENCES]->(:Column)

Graph Generation

This project provides connector classes that organize the ETL process into reusable components:

Extractors - Connect to source data and read metadata tables
Transformers - Transform metadata into defined Neo4j schema
Loaders - Ingest transformed data into Neo4j
Connectors - Orchestrate the extract, transform, and load process

Each connector is implemented as a class that encapsulates its extractor, transformer, and loader components, providing a clean interface for ingestion.

Connectors

OSI Connector

Bidirectional connector for the Open Semantic Interchange (OSI) spec — a YAML-based interchange format for semantic models. Unlike the other connectors (which only ingest), OSI supports both directions: load an OSI YAML spec into Neo4j, and emit an OsiSemanticModel subgraph back out as an OSI YAML file. See the OSI README for the full data model diagram and behavioral notes.

What it ingests (from an OSI YAML at a local path or HTTP(S) URL):

OsiSemanticModel (a Domain subtype) — the top-level container
OsiTable / OsiColumn — datasets and their fields (with primary keys, unique keys, labels, time-dimension flags)
Query — datasets whose source is a SQL query rather than a 3-part identifier
Metric with dialect-specific Expression definitions
Join with ordered from_columns / to_columns for composite-key relationships
OsiAiContext / OsiCustomExtensions aspects, including synonyms-derived BusinessTerm upserts (MERGE on name, so they collide cleanly with catalog-derived BTs from Dataplex etc.)

What it exports (from an :OsiSemanticModel subgraph, filtered by name):

A spec-compliant OSI YAML file with preserved column ordering, native ai_context structure, and literal-block JSON for custom extensions.

This connector only requires Neo4j credentials in .env:

NEO4J_USERNAME=neo4j-username
NEO4J_PASSWORD=neo4j-password
NEO4J_URI=neo4j-uri
NEO4J_DATABASE=neo4j-database

Sample dataset: datasets/osi/acme_semantic_model.yaml (the full 33-table ACME warehouse modeled as OSI). Runnable example: examples/osi_connector.py.

BigQuery Schema Connector

Connector for reading BigQuery Information Schema tables and ingesting schema metadata into Neo4j. Primary and foreign keys must be defined in the Information Schema tables in order for column level relationships to be created in the Neo4j graph.

What it extracts:

Database (GCP project)
Schemas (datasets)
Tables with descriptions
Columns with types, constraints, descriptions
Column references (foreign keys)
Column unique values (sample data)

This connector requires the following variables to be set in the .env file:

NEO4J_USERNAME=neo4j-username
NEO4J_PASSWORD=neo4j-password
NEO4J_URI=neo4j-uri
NEO4J_DATABASE=neo4j-database
GCP_PROJECT_ID=project-id
BIGQUERY_DATASET_ID=dataset-id

BigQuery Logs Connector

Connector for extracting query logs from BigQuery INFORMATION_SCHEMA.JOBS_BY_PROJECT, parsing SQL queries to understand table and column usage, and loading query patterns into Neo4j.

What it extracts:

SQL Queries
Tables and columns referenced in queries (discovered via SQL parsing)
Join relationships between tables (from SQL JOINs)
Query-to-table and query-to-column usage relationships

Graph schema additions:

Query nodes with properties:
- content - The query text
- query_id - Hash of query text
(:Query)-[:USES_TABLE]->(:Table) relationships
(:Query)-[:USES_COLUMN]->(:Column) relationships

This connector requires the following variables to be set in the .env file:

NEO4J_USERNAME=neo4j-username
NEO4J_PASSWORD=neo4j-password
NEO4J_URI=neo4j-uri
NEO4J_DATABASE=neo4j-database
GCP_PROJECT_ID=project-id
BIGQUERY_DATASET_ID=dataset-id
BIGQUERY_REGION=region-us (optional, defaults to region-us)

Connector Architecture

---
config:
    layout: elk
---
graph LR
    subgraph Schema["Graph Schema"]
        GS(Data Model Definition)
    end

    subgraph Source["Source Repository"]
        BQ(BigQuery Database)        
    end

    subgraph ETL["ETL Processes"]
        QE(Read RDBMS Schema)   
        PM(Validate + Transform<br>with Pydantic) 
    
        QE -->|Raw Data<br/>JSON| PM
    end
    
    subgraph Graph["Database"]
        NEO[(Neo4j Graph)]
    end

    BQ -->|Information Schema| QE
    GS -->|Schema Definition| PM
    PM -->|Ingest Data| NEO

Code Example - Schema Connector

import os
from neo4j import GraphDatabase
from google.cloud import bigquery
from neocarta.connectors.bigquery import BigQuerySchemaConnector

# Initialize clients
neo4j_driver = GraphDatabase.driver(
    uri=os.getenv("NEO4J_URI"),
    auth=(os.getenv("NEO4J_USERNAME"), os.getenv("NEO4J_PASSWORD")),
)
neo4j_database = os.getenv("NEO4J_DATABASE", "neo4j")
bigquery_client = bigquery.Client(project=os.getenv("GCP_PROJECT_ID"))

# Create connector instance
connector = BigQuerySchemaConnector(
    client=bigquery_client,
    project_id=os.getenv("GCP_PROJECT_ID"),
    neo4j_driver=neo4j_driver,
    database_name=neo4j_database,
)

# Run the connector to extract, transform, and load BigQuery schema metadata into Neo4j
connector.ingest(dataset_id=os.getenv("BIGQUERY_DATASET_ID"))

Code Example - Logs Connector

import os
from neo4j import GraphDatabase
from google.cloud import bigquery
from neocarta.connectors.bigquery import BigQueryLogsConnector

# Initialize clients
neo4j_driver = GraphDatabase.driver(
    uri=os.getenv("NEO4J_URI"),
    auth=(os.getenv("NEO4J_USERNAME"), os.getenv("NEO4J_PASSWORD")),
)
neo4j_database = os.getenv("NEO4J_DATABASE", "neo4j")
bigquery_client = bigquery.Client(project=os.getenv("GCP_PROJECT_ID"))

# Create connector instance
connector = BigQueryLogsConnector(
    client=bigquery_client,
    project_id=os.getenv("GCP_PROJECT_ID"),
    neo4j_driver=neo4j_driver,
    database_name=neo4j_database,
)

# Run the connector to extract query logs, parse SQL, and load into Neo4j
connector.ingest(
    dataset_id=os.getenv("BIGQUERY_DATASET_ID"),
    region="region-us",
    start_timestamp="2024-01-01 00:00:00",  # Optional
    end_timestamp="2024-01-31 23:59:59",    # Optional
    limit=100,                               # Optional, default 100
    drop_failed_queries=True,                # Optional, default True
)

Combined Usage

For the most complete picture, run both connectors:

# 1. Extract schema metadata
schema_connector = BigQuerySchemaConnector(...)
schema_connector.ingest(dataset_id=os.getenv("BIGQUERY_DATASET_ID"))

# 2. Extract query logs
logs_connector = BigQueryLogsConnector(...)
logs_connector.ingest(dataset_id=os.getenv("BIGQUERY_DATASET_ID"))

This allows you to compare declared schema vs. actual usage patterns.

GCP Dataplex Universal Catalog

Connector for reading BigQuery metadata and Glossary information from Dataplex and ingesting into Neo4j. Please see the Dataplex README for more information and caveats of using this connector.

Connector Architecture

---
config:
    layout: elk
---
graph LR
    subgraph Schema["Graph Schema"]
        GS(Data Model Definition)
    end
    subgraph Big["Google Cloud Platform"]
        subgraph SR["Source Repository"]
            BQ(BigQuery Database)  
        end      
    

        subgraph Source["GCP Dataplex Universal Catalog"]
            G(Glossaries)
            MT(Metadata Types)        
        end
    end

    subgraph ETL["ETL Processes"]
        QE(Read Data Catalog)   
        PM(Validate + Transform<br>with Pydantic) 
    
        QE -->|Raw Data<br/>JSON| PM
    end
    
    subgraph Graph["Database"]
        NEO[(Neo4j Graph)]
    end

    BQ -->|BigQuery Metadata|MT
    G -->| Glossary Content| QE
    MT -->|BigQuery Metadata| QE
    GS -->|Schema Definition| PM
    PM -->|Ingest Data| NEO

Data Model

---
config:
    layout: elk
---
graph LR
%% Database Nodes
Database("Database<br/>id: STRING | KEY<br/>name: STRING<br/>platform: STRING<br/>service: STRING")
Schema("Schema<br/>id: STRING | KEY<br/>name: STRING")
Table("Table<br/>id: STRING | KEY<br/>name: STRING<br/>description: STRING<br/>embedding: VECTOR")
Column("Column<br/>id: STRING | KEY<br/>name: STRING<br/>description: STRING<br/>embedding: VECTOR<br/>type: STRING<br/>nullable: BOOLEAN")

%% Glossary Nodes
Glossary("Glossary<br/>id: STRING | KEY<br/>name: STRING<br/>description: STRING")
Category("Category<br/>id: STRING | KEY<br/>name: STRING<br/>description: STRING")
BusinessTerm("BusinessTerm<br/>id: STRING | KEY<br/>name: STRING<br/>description: STRING<br/>embedding: VECTOR")

%% Database Relationships
Database -->|HAS_SCHEMA| Schema
Schema -->|HAS_TABLE| Table
Table -->|HAS_COLUMN| Column
Column -->|REFERENCES| Column

%% Glossary Relationships
Glossary -->|HAS_CATEGORY| Category
Category -->|HAS_BUSINESS_TERM| BusinessTerm

%% Cross-domain Relationships
Column -->|TAGGED_WITH| BusinessTerm
Table -->|TAGGED_WITH| BusinessTerm

Query Logs

Connector for parsing query log JSON files into Neo4j. Please see the Query Logs README for more information and caveats of using this connector.

Connector Architecture

---
config:
    layout: elk
---
graph LR
    subgraph Schema["Graph Schema"]
        GS(Data Model Definition)
    end
    subgraph Big["Google Cloud Platform"]
        subgraph Source["Source Repository"]
            BQ(BigQuery Database)
        end

        subgraph Logging["GCP Logging"]
            GL(Logs)
        end
    end

    subgraph ETL["ETL Processes"]
        QL(Read Query Logs)
        PM(Validate + Transform<br>with Pydantic)

        QL -->|Raw Logs<br/>JSON| PM
    end

    subgraph Graph["Database"]
        NEO[(Neo4j Graph)]
    end

    BQ -->|Query Logs| GL
    GL -->|Query Logs| QL
    GS -->|Schema Definition| PM
    PM -->|Ingest Data| NEO

CSV Files

Connector for loading metadata from structured CSV files into Neo4j. This connector is useful for importing metadata from systems that don't have direct API access, or for loading curated metadata that has been manually created or exported from other tools.

The CSV connector supports selective loading - you can choose which node types and relationships to load based on what metadata is available in your CSV files.

CSV File Structure:

The connector expects CSV files in a specified directory with the following default naming convention:

database_info.csv - Database nodes
schema_info.csv - Schema nodes
table_info.csv - Table nodes
column_info.csv - Column nodes
column_references_info.csv - Foreign key relationships
value_info.csv - Sample column values
query_info.csv - Query nodes
query_table_info.csv - Query-to-table relationships
query_column_info.csv - Query-to-column relationships
glossary_info.csv - Glossary nodes
category_info.csv - Category nodes
business_term_info.csv - Business term nodes

Custom file names can be specified using the csv_file_map parameter.

ID Strategy:

Entity IDs are automatically generated from name columns using a dot-separated hierarchy (e.g., database_name.schema_name.table_name for tables, glossary_name.category_name.term_name for business terms). To use custom IDs instead, supply explicit *_id columns in every CSV file in the hierarchy. Do not mix strategies across files. If loading glossary data from both the CSV and Dataplex connectors into the same graph, you must supply explicit IDs in the CSV that match the Dataplex resource paths. See the CSV Connector README for full details.

This connector requires the following variables to be set in the .env file:

NEO4J_USERNAME=neo4j-username
NEO4J_PASSWORD=neo4j-password
NEO4J_URI=neo4j-uri
NEO4J_DATABASE=neo4j-database

Connector Architecture

---
config:
    layout: elk
---
graph LR
    subgraph Schema["Graph Schema"]
        GS(Data Model Definition)
    end

    subgraph Source["CSV Files"]
        CSV[(CSV Directory)]
    end

    subgraph ETL["ETL Processes"]
        QE(Read CSV Files)
        PM(Validate + Transform<br>with Pydantic)

        QE -->|Raw Data<br/>DataFrame| PM
    end

    subgraph Graph["Database"]
        NEO[(Neo4j Graph)]
    end

    CSV -->|CSV Files| QE
    GS -->|Schema Definition| PM
    PM -->|Ingest Data| NEO

Code Example

import os
from neo4j import GraphDatabase
from neocarta import NodeLabel as nl, RelationshipType as rt
from neocarta.connectors.csv import CSVConnector

# Initialize clients
neo4j_driver = GraphDatabase.driver(
    uri=os.getenv("NEO4J_URI"),
    auth=(os.getenv("NEO4J_USERNAME"), os.getenv("NEO4J_PASSWORD")),
)
neo4j_database = os.getenv("NEO4J_DATABASE", "neo4j")

# Create connector instance
connector = CSVConnector(
    csv_directory="datasets/csv",
    neo4j_driver=neo4j_driver,
    database_name=neo4j_database,
)

# Run the connector to load all CSV files into Neo4j
connector.ingest()

# Alternatively, load specific nodes and relationships
# Enum members are recommended, but exact string values (e.g. "Database", "HAS_SCHEMA") also work.
connector.ingest(
    include_nodes=[nl.DATABASE, nl.SCHEMA, nl.TABLE, nl.COLUMN, nl.VALUE],
    include_relationships=[rt.HAS_SCHEMA, rt.HAS_TABLE, rt.HAS_COLUMN, rt.HAS_VALUE, rt.REFERENCES]
)

# Or use a custom file mapping (configured at construction time)
custom_file_map = {
    NodeLabel.DATABASE: "my_database.csv",
    NodeLabel.SCHEMA: "my_schema.csv",
    # ... other custom filenames
}
connector = CSVConnector(
    csv_directory="datasets/csv",
    neo4j_driver=neo4j_driver,
    database_name=neo4j_database,
    csv_file_map=custom_file_map,
)
connector.ingest()

Sample Dataset

A sample e-commerce dataset is provided in datasets/csv/ that demonstrates the expected CSV file structure and can be used for testing the CSV connector.

Embeddings

Embeddings may be generated for the description fields of the following nodes:

Database
Schema
Table
Column
BusinessTerm

Two embedding connectors are available:

LiteLLMEmbeddingsConnector — multi-provider via LiteLLM. Routes to OpenAI, Azure OpenAI, Gemini, Cohere, Bedrock, Vertex AI, Ollama, HuggingFace, and others based on the embedding_model string. Vector dimension is auto-detected from the model on first use. Use this when you want provider flexibility or are not on OpenAI.
OpenAIEmbeddingsConnector — direct OpenAI SDK. Takes a pre-built OpenAI / AsyncOpenAI client and an explicit dimensions value. Use this when you want full control over the OpenAI client (custom base URL, retry policy, proxies) or already have one wired up elsewhere in your app.

Authentication is read from provider-specific environment variables (.env file). For OpenAI:

OPENAI_API_KEY=sk-...

For other providers via LiteLLM, set the matching env var (e.g. GEMINI_API_KEY, COHERE_API_KEY, AZURE_API_KEY + AZURE_API_BASE, AWS_*). For LiteLLM Proxy or custom endpoints, pass api_key / api_base in litellm_kwargs.

Connector Architecture

---
config:
    layout: elk
---
graph LR

    subgraph D["Database Preparation"]
        VI(Create Vector Index)
    end

    subgraph ES["Embedding Provider"]
        E(OpenAI / LiteLLM)
    end

    subgraph Graph["Database"]
        NEO[(Neo4j Graph)]
    end

    subgraph EP["Embedding Process"]
        C(Create Embeddings) 
    end
    
    VI-->NEO

    C<-->E

    NEO-->|Unprocessed Node Descriptions|C
    C-->|Embeddings|NEO

Code Example — LiteLLM (multi-provider)

import asyncio
import os
from neo4j import GraphDatabase
from neocarta import NodeLabel as nl
from neocarta.enrichment.embeddings import LiteLLMEmbeddingsConnector

# Initialize Neo4j driver. The embedding provider is configured via env vars
# (e.g. OPENAI_API_KEY, GEMINI_API_KEY) consumed by LiteLLM at call time.
neo4j_driver = GraphDatabase.driver(
    uri=os.getenv("NEO4J_URI"),
    auth=(os.getenv("NEO4J_USERNAME"), os.getenv("NEO4J_PASSWORD")),
)
neo4j_database = os.getenv("NEO4J_DATABASE", "neo4j")

# Create connector instance. Vector dimension is auto-detected from the model.
connector = LiteLLMEmbeddingsConnector(
    embedding_model="text-embedding-3-small",
    neo4j_driver=neo4j_driver,
    database_name=neo4j_database,
)

# The node labels to generate embeddings for
# Enum members are recommended, but exact string values (e.g. "Database", "Table") also work.
node_labels = [nl.DATABASE, nl.TABLE, nl.COLUMN]

# Run the connector to create embeddings for the nodes
await connector.arun(node_labels=node_labels)

Code Example — OpenAI (direct SDK)

import os
from neo4j import GraphDatabase
from openai import AsyncOpenAI
from neocarta import NodeLabel as nl
from neocarta.enrichment.embeddings import OpenAIEmbeddingsConnector

neo4j_driver = GraphDatabase.driver(
    uri=os.getenv("NEO4J_URI"),
    auth=(os.getenv("NEO4J_USERNAME"), os.getenv("NEO4J_PASSWORD")),
)

# Bring your own OpenAI client — useful when you need a custom base URL,
# retry policy, or proxy configuration.
async_client = AsyncOpenAI(api_key=os.getenv("OPENAI_API_KEY"))

connector = OpenAIEmbeddingsConnector(
    neo4j_driver=neo4j_driver,
    async_client=async_client,
    embedding_model="text-embedding-3-small",
    dimensions=768,
    database_name=os.getenv("NEO4J_DATABASE", "neo4j"),
)

await connector.arun(node_labels=[nl.DATABASE, nl.TABLE, nl.COLUMN])

Full Pipeline

The full graph generation pipeline will run the BigQuery connector followed by the embedding generation connector.

It requires the following variables to be set in the .env file:

NEO4J_USERNAME=neo4j-username
NEO4J_PASSWORD=neo4j-password
NEO4J_URI=neo4j-uri
NEO4J_DATABASE=neo4j-database
GCP_PROJECT_ID=project-id
BIGQUERY_DATASET_ID=dataset-id
OPENAI_API_KEY=sk-...

Architecture

The combined BigQuery + Embeddings connector pipeline is seen below.

flowchart LR
    subgraph Schema["Graph Schema"]
        GS(Data Model Definition)
    end

    subgraph Source["Source Repository"]
        BQ(BigQuery Database)        
    end

    subgraph ETL["ETL Processes"]
        QE(Read RDBMS Schema)   
        PM(Validate + Transform<br>with Pydantic) 
    end
    
    subgraph Graph["Database"]
        NEO[(Neo4j Graph)]
    end

    subgraph D["Database Preparation"]
        VI(Create Vector Index)
    end

    subgraph ES["Embedding Provider"]
        E(OpenAI / LiteLLM)
    end

    subgraph EP["Embedding Workflow"]
        C(Create Embeddings) 
    end

    %% BigQuery Flow
    BQ -->|Information Schema| QE
    QE -->|Raw Data JSON| PM
    GS -->|Schema Definition| PM
    PM -->|Ingest Data| NEO

    %% Embeddings Flow
    NEO -->|Database Ready| VI
    VI -->|Vector Index Created| NEO
    NEO -->|Unprocessed Node Descriptions| C
    C <--> E
    C -->|Embeddings| NEO

Running The Full Connector Pipeline

To run the full connector pipeline, use the following Make command:

make create-graph

Neocarta CLI

The Neocarta CLI is available via the optional [cli] add-on. It wraps the same connector classes covered above behind a noun-verb command grammar so you can drive ingestion without writing Python.

pip install "neocarta[cli]"

Today the CLI ships these connector commands:

Command	Wraps
`neocarta bigquery schema`	`BigQuerySchemaConnector` — load `Database`, `Schema`, `Table`, `Column` nodes
`neocarta bigquery logs`	`BigQueryLogsConnector` — load `Query`, `CTE`, and reference relationships from `INFORMATION_SCHEMA.JOBS_BY_PROJECT`
`neocarta csv ingest`	`CSVConnector` — load metadata from a directory of CSV files
`neocarta dataplex schema`	`DataplexSchemaConnector` — load BigQuery schema (`Database`, `Schema`, `Table`, `Column`) from the Dataplex catalog
`neocarta dataplex glossary`	`DataplexGlossaryConnector` — load the Dataplex business glossary (`Glossary`, `Category`, `BusinessTerm`) and `TAGGED_WITH` entry links
`neocarta osi ingest`	`OsiConnector` — load an OSI YAML semantic model from a local path or HTTP(S) URL
`neocarta osi export`	`OsiConnector` — export an OSI semantic model from Neo4j back to an OSI YAML file
`neocarta query-log ingest`	`QueryLogConnector` — parse a local query-log JSON file into `Query`, `CTE`, and reference relationships (distinct from `bigquery logs`, which reads the Cloud Logging API live)

Plus the MCP server tools, mirrored under neocarta tool <tool> so the graph can be queried straight from the shell or a non-MCP agent (read-only; need only the [cli] install):

Command	Mirrors MCP tool
`neocarta tool list-schemas`	`list_schemas` — list every schema and its database
`neocarta tool list-tables-by-schema --schema-name S`	`list_tables_by_schema` — list the tables in schema `S`
`neocarta tool get-full-metadata-schema`	`get_full_metadata_schema` — dump full table/column metadata (large)
`neocarta tool get-context-by-{table,column}-vector-search --text-content "..."`	semantic (embedding) search over table/column descriptions
`neocarta tool get-context-by-schema-and-table-vector-search --text-content "..."`	semantic search across schema + table embeddings
`neocarta tool get-context-by-{table,column}-full-text-search --text-content "..."`	full-text search over table/column name + description
`neocarta tool get-context-by-{table,column}-hybrid-search --text-content "..."`	hybrid vector + full-text search
`neocarta tool get-context-by-{table,column}-business-term-hybrid-search --text-content "..."`	hybrid search bridged through `:BusinessTerm` tags

Each neocarta tool command mirrors its tool's name, --text-content / --max-tables / --search-top-k arguments (and per-tool defaults), and help text. The catalog commands work from schema alone; the search commands need the matching vector/full-text indexes (build them with an ingest --embeddings) and, where they embed the query, an embedding-provider key (e.g. OPENAI_API_KEY). A search command run against a graph missing the required index exits 3 (not_found).

Plus one introspection verb:

Command	Purpose
`neocarta agent-context`	Emits the full CLI shape (commands, flags, exit codes, env vars) as JSON for AI agents to read

Example

# Set NEO4J_URI / NEO4J_USERNAME / NEO4J_PASSWORD / OPENAI_API_KEY in your shell or .env
neocarta bigquery schema --project-id my-proj --dataset-id sales
neocarta bigquery schema --project-id my-proj --dataset-id sales --embeddings
neocarta bigquery logs --dataset-id sales --limit 500 --json
neocarta csv ingest --csv-directory ./datasets/csv
neocarta dataplex schema --project-id my-proj --project-number 123456789 --dataplex-location us --dataset-id sales
neocarta dataplex glossary --project-id my-proj --project-number 123456789 --dataplex-location us
neocarta osi ingest --spec-source ./datasets/osi/acme_semantic_model.yaml
neocarta osi export --semantic-model-name acme_corp_model --output-path acme.yaml
neocarta query-log ingest --query-log-file ./query_logs.json

# Query the graph with the mirrored MCP tools (read-only):
neocarta tool list-schemas --json
neocarta tool get-context-by-table-vector-search --text-content "customer orders" --max-tables 5 --json

See the CLI README for the full flag reference, env-var contract, exit-code map, and agent-integration details.

Neocarta MCP

The Neocarta MCP server is available via the optional [mcp] add-on. Start it with the standalone neocarta-mcp console script, or from the unified CLI with neocarta mcp serve (which requires both the cli and mcp extras). Both serve over stdio and read the same NEO4J_* / EMBEDDING_* environment configuration.

Overview

This is a metadata retrieval MCP server that provides tools to query the Neo4j semantic layer for relevant schema information using semantic similarity search and graph traversal. It is built for compatibility with the standard data model provided by the neocarta library.

Tools

list_schemas - List all schemas and their associated databases.
list_tables_by_schema - List all tables for a given schema name.
get_context_by_column_vector_search - Find tables by semantic similarity on column embeddings.
get_context_by_table_vector_search - Find tables by semantic similarity on table embeddings.
get_context_by_schema_and_table_vector_search - Find tables by semantic similarity across both schema and table embeddings.
get_context_by_column_full_text_search / get_context_by_table_full_text_search - Full-text search on column or table name/description.
get_context_by_column_hybrid_search / get_context_by_table_hybrid_search - Hybrid vector + full-text search at the column or table level.
get_context_by_column_business_term_hybrid_search / get_context_by_table_business_term_hybrid_search - Hybrid search with the full-text branch bridged through :BusinessTerm tags.
get_full_metadata_schema - Return complete metadata for all tables. Warning: expensive — use only for debugging.

The MCP server probes the target database at startup and registers, per label (Table, Column), the highest-priority retrieval tool whose indexes are present: business-term-bridged hybrid > hybrid > vector or full-text alone. Schema-level vector retrieval and catalog tools are registered independently.

Every one of these tools is also reachable from the CLI as neocarta tool <tool> (e.g. neocarta tool get-context-by-table-vector-search --text-content "...") — same names, arguments, and documentation — for shell use or non-MCP agents, without running the server or installing the [mcp] extra.

See the MCP server README for full server documentation.

Claude Desktop

To connect the neocarta-mcp server to Claude Desktop, add the following entry to your claude_desktop_config.json:

{
  "mcpServers": {
    "neocarta": {
      "command": "uvx",
      "args": [
        "--from",
        "neocarta[mcp]@0.8.0",
        "neocarta-mcp"
      ],
      "env": {
        "NEO4J_URI": "bolt://localhost:7687",
        "NEO4J_USERNAME": "neo4j",
        "NEO4J_PASSWORD": "your-password",
        "NEO4J_DATABASE": "neo4j",
        "OPENAI_API_KEY": "sk-...",
        "EMBEDDING_MODEL": "text-embedding-3-small"
      }
    }
  }
}

Contributing

Dev Setup

This project uses uv for dependency management and requires Python 3.10 or higher.

uv — Python dependency manager:

curl -LsSf https://astral.sh/uv/install.sh | sh

Neo4j — a running Neo4j instance is required. Options:

Neo4j AuraDB — managed cloud (free tier available)
Neo4j Desktop — local GUI-based instance
Docker image — lightweight instance

Install All Dependencies (Recommended)

For most users, install all dependencies to run the complete workflow:

make install

This installs all dependency groups and allows you to:

Create the metadata graph from BigQuery
Run the MCP server
Run the Text2SQL agent

Install Specific Components

If you only need specific components, you can install individual dependency groups:

Metadata Graph Only (BigQuery ETL + embeddings)

make install-metadata-graph

MCP Server Only (neocarta MCP server)

make install-mcp-server

Agent Only (Text2SQL agent with MCP servers)

make install-agent

Note: The agent group automatically includes mcp-server dependencies

Dependency Groups

The project is organized into the following dependency groups:

metadata-graph: BigQuery metadata extraction, Neo4j loading, and embedding generation
mcp: neocarta MCP server for metadata retrieval from Neo4j semantic layer
agent: Text2SQL agent with LangChain (includes mcp-server dependencies)
dev: Development tools (Jupyter notebooks)

Adding a New Connector

Every connector under neocarta/connectors/ follows a shared standard — directory layout, the extract / transform / load / ingest (and export for format connectors) public API, error/warning conventions, id generation, and a required README. The full contract is documented in connector-contract.md. Read it before designing a connector.

The repository ships a neocarta-add-source-connector Claude Code skill to build connectors against that contract. In Claude Code, run /neocarta-add-source-connector; the skill scaffolds a conformant connector package (plus its conformance test) and verifies it. The underlying tooling is also usable directly:

# List connectors and their detected kind (source/format)
uv run .claude/skills/neocarta-add-source-connector/scripts/driver.py list

# Scaffold a new source connector package + conformance test
uv run .claude/skills/neocarta-add-source-connector/scripts/driver.py scaffold <name>

# Verify a connector against the contract (static checks + conformance pytest)
uv run .claude/skills/neocarta-add-source-connector/scripts/driver.py verify <name>

A scaffolded connector is lint-clean as generated; fill in the extract / transform / load stages, then re-run verify. Connector creation and CLI integration are separate PRs.

Sample Datasets

This repository contains two sample datasets

ecommerce (4 tables)
acme dataset (33 tables)

Ensure that the following environment variable is set before running and that you are credentialed via the gcloud cli.

GCP_PROJECT_ID=project-id

To create the dataset in your BigQuery instance, you may run the following command.

uv run examples/bigquery.py --dataset=acme

# or

uv run examples/bigquery.py --dataset=ecommerce

Agent

This is the Text2SQL agent that converts natural language questions into SQL queries for BigQuery. The agent uses two MCP servers to:

Retrieve relevant database metadata from Neo4j using semantic similarity
Execute generated SQL queries against BigQuery

The agent architecture can be seen below.

---
config:
    layout: dagre
---

graph LR
    
    subgraph GCP["GCP Environment"]
        BQMCP(BigQuery<br>MCP)

        subgraph DataWarehouse["Data Warehouse"]
            BQData[(BigQuery)]
        end
        
        BQMCP <--> BQData
    end

    subgraph Local["Local Environment"]
        Agent("Text2SQL Agent")
        MetadataMCP("neocarta<br/>MCP")
        
        subgraph Graph["Database"]
            NEO[(Neo4j Graph)]
        end
        
        Agent <--> MetadataMCP
        MetadataMCP <--> NEO
    end
    
    User("User")
    
    subgraph LLM["LLM Service"]
        Model("LLM")
    end
    
    User <--> Agent
    Agent <--> Model
    Agent <--> BQMCP

How it works

User asks a natural language question about the data
Agent calls the SQL Metadata MCP server to retrieve relevant table schemas
Agent generates a SQL query via an LLM call based on the retrieved metadata context
Agent calls execute_sql from the BigQuery MCP server to run the query against BigQuery
Agent returns formatted results to the user

BigQuery MCP Server Set Up

Enable use of the Bigquery MCP server in your project.

Additional information may be found here.

gcloud beta services mcp enable bigquery.googleapis.com --project=PROJECT_ID

To disable again run:

gcloud beta services mcp disable bigquery.googleapis.com --project=PROJECT_ID

You can test the BigQuery server connection with the following curl command

curl -k \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer $(gcloud auth application-default print-access-token)" \
  -d '{
  "jsonrpc": "2.0",
  "id": 3,
  "method": "tools/call",
  "params": {
    "name": "execute_sql",
    "arguments": {
      "projectId": "<PROJECT_ID>",
      "query": "SELECT table_name FROM `<PROJECT_ID>.<DATASET_ID>.INFORMATION_SCHEMA.TABLES`"
    }
  }
}' \
  https://bigquery.googleapis.com/mcp

Running the Agent

Use this command to run the agent locally:

make agent

The agent will start an interactive chat session in the terminal where you can ask questions about your data.

BigQuery MCP Authentication

The agent uses Google Cloud Application Default Credentials for BigQuery authentication:

class GoogleAuth(httpx.Auth):
    def __init__(self):
        self.credentials, _ = default()

    def auth_flow(self, request):
        self.credentials.refresh(Request())
        request.headers["Authorization"] = f"Bearer {self.credentials.token}"
        yield request

Make sure you're authenticated with:

gcloud auth application-default login

Environment Variables

Required environment variables (add to .env file):

Neo4j Connection

NEO4J_URI - Neo4j database URI (e.g., bolt://localhost:7687)
NEO4J_USERNAME - Neo4j username (default: neo4j)
NEO4J_PASSWORD - Neo4j password
NEO4J_DATABASE - Neo4j database name (default: neo4j)

LLM - OpenAI

OPENAI_API_KEY - OpenAI API key for embeddings and LLM

Example Usage

> What are the total sales by product category?

Agent: [Calls get_context_by_column_vector_search with query about sales and categories]
Agent: [Generates SQL query using retrieved schema]
Agent: [Calls execute_sql with generated query]
Agent: Here are the total sales by product category:
- Electronics: $15,234.50
- Clothing: $8,912.30
...

Name		Name	Last commit message	Last commit date
.claude/skills		.claude/skills
.github/workflows		.github/workflows
agent		agent
assets		assets
datasets		datasets
docs		docs
eval		eval
examples		examples
neocarta		neocarta
scripts		scripts
tests		tests
.env.example		.env.example
.gitignore		.gitignore
.python-version		.python-version
CHANGELOG.md		CHANGELOG.md
LICENSE		LICENSE
Makefile		Makefile
README.md		README.md
REVIEW.md		REVIEW.md
claude.md		claude.md
pyproject.toml		pyproject.toml
run_agent.py		run_agent.py
uv.lock		uv.lock

Latest commit

History

Folders and files

Repository files navigation

Neocarta

What it is

Quickstart

Installation

Performance Accelerator

Metadata Graph

Graph Generation

Connectors

OSI Connector

BigQuery Schema Connector

BigQuery Logs Connector

Connector Architecture

Code Example - Schema Connector

Code Example - Logs Connector

Combined Usage

GCP Dataplex Universal Catalog

Connector Architecture

Data Model

Query Logs

Connector Architecture

CSV Files

Connector Architecture

Code Example

Sample Dataset

Embeddings

Connector Architecture

Code Example — LiteLLM (multi-provider)

Code Example — OpenAI (direct SDK)

Full Pipeline

Architecture

Neocarta CLI

Example

Neocarta MCP

Overview

Claude Desktop

Contributing

Dev Setup

Install All Dependencies (Recommended)

Install Specific Components

Dependency Groups

Adding a New Connector

Sample Datasets

Agent

About

Resources

Stars

Watchers

Forks

Releases

Packages

Contributors

Languages