Neo4jDriver

Neo4j graph database driver implementation

The Neo4jDriver class provides connectivity to Neo4j graph databases with full ACID transaction support.

Overview

Neo4j is the industry-standard graph database platform. The Neo4jDriver leverages the official neo4j Python driver to provide:

• Full ACID transaction support
• Automatic index and constraint creation
• Connection pooling
• Concurrent query execution
• Health checks

Constructor

from graphiti_core.driver.neo4j_driver import Neo4jDriver

driver = Neo4jDriver(
    uri="bolt://localhost:7687",
    user="neo4j",
    password="password",
    database="neo4j"
)

Parameters

uri · str · required

The connection URI for the Neo4j database.

Supported URI schemes:

• bolt:// - Unencrypted connection
• bolt+s:// - Encrypted connection with full certificate
• bolt+ssc:// - Encrypted connection, self-signed certificate
• neo4j:// - Routing driver (for clusters)
• neo4j+s:// - Encrypted routing driver

Example: "bolt://localhost:7687"

user · str | None · required

The username for Neo4j authentication.

Can be None for databases without authentication (not recommended for production).

Example: "neo4j"

password · str | None · required

The password for Neo4j authentication.

Can be None for databases without authentication (not recommended for production).

Example: "password"

database · str · default: "neo4j"

The name of the Neo4j database to connect to.

Neo4j supports multiple databases. The default database is typically named "neo4j".

Example: "graphiti"

Methods

execute_query

Execute a Cypher query and return results.

result = await driver.execute_query(
    "MATCH (n:Entity) RETURN n LIMIT 10",
    params={"limit": 10}
)

Parameters:

• cypher_query_ (str) - The Cypher query to execute
• **kwargs - Query parameters

Returns: EagerResult - Neo4j result object

session

Create a new database session.

session = driver.session(database="neo4j")
try:
    result = await session.run("MATCH (n) RETURN count(n)")
finally:
    await session.close()

Parameters:

• database (str | None) - Optional database name override

Returns: GraphDriverSession - Session object

transaction

Context manager for ACID transactions.

async with driver.transaction() as tx:
    await tx.run("CREATE (n:Entity {uuid: $uuid})", uuid="123")
    await tx.run("CREATE (m:Entity {uuid: $uuid})", uuid="456")
    # Automatically commits on success, rolls back on exception

Returns: AsyncIterator[Transaction] - Transaction context manager

close

Close the driver and release all connections.

await driver.close()

Returns: None

build_indices_and_constraints

Create all required indices and constraints for Graphiti.

await driver.build_indices_and_constraints(delete_existing=False)

Parameters:

• delete_existing (bool) - If True, delete existing indices before creating new ones

Returns: None

This method is automatically called during driver initialization. You typically don’t need to call it manually unless you want to rebuild indices.

health_check

Verify connectivity to the Neo4j database.

try:
    await driver.health_check()
    print("Connection healthy")
except Exception as e:
    print(f"Connection failed: {e}")

Returns: None - Raises exception if connection fails

with_database

Create a shallow copy of the driver targeting a different database.

default_driver = Neo4jDriver(
    uri="bolt://localhost:7687",
    user="neo4j",
    password="password",
    database="neo4j"
)

# Create a driver for a different database
other_driver = default_driver.with_database("other_db")

Parameters:

• database (str) - The name of the database to target

Returns: GraphDriver - New driver instance with different database

Properties

provider

driver.provider  # GraphProvider.NEO4J

Returns the GraphProvider.NEO4J enum value.

Operations Properties

The driver exposes specialized operation interfaces:

driver.entity_node_ops       # EntityNodeOperations
driver.episode_node_ops      # EpisodeNodeOperations
driver.community_node_ops    # CommunityNodeOperations
driver.saga_node_ops         # SagaNodeOperations
driver.entity_edge_ops       # EntityEdgeOperations
driver.episodic_edge_ops     # EpisodicEdgeOperations
driver.community_edge_ops    # CommunityEdgeOperations
driver.has_episode_edge_ops  # HasEpisodeEdgeOperations
driver.next_episode_edge_ops # NextEpisodeEdgeOperations
driver.search_ops            # SearchOperations
driver.graph_ops             # GraphMaintenanceOperations

These properties provide access to database-specific implementations of graph operations.

Usage Examples

Basic Connection

import asyncio
from graphiti_core.driver.neo4j_driver import Neo4jDriver

async def main():
    driver = Neo4jDriver(
        uri="bolt://localhost:7687",
        user="neo4j",
        password="password"
    )
    
    try:
        # Verify connection
        await driver.health_check()
        print("Connected to Neo4j")
        
        # Execute a query
        result = await driver.execute_query(
            "MATCH (n:Entity) RETURN count(n) as count"
        )
        print(f"Entity count: {result.records[0]['count']}")
    finally:
        await driver.close()

asyncio.run(main())

With Graphiti

import os
from graphiti_core import Graphiti
from graphiti_core.driver.neo4j_driver import Neo4jDriver

# Create driver
driver = Neo4jDriver(
    uri=os.environ.get("NEO4J_URI", "bolt://localhost:7687"),
    user=os.environ.get("NEO4J_USER", "neo4j"),
    password=os.environ.get("NEO4J_PASSWORD", "password")
)

# Initialize Graphiti
graphiti = Graphiti(graph_driver=driver)

# Or use shorthand for Neo4j
graphiti = Graphiti(
    uri="bolt://localhost:7687",
    user="neo4j",
    password="password"
)

Using Transactions

from graphiti_core.driver.neo4j_driver import Neo4jDriver

driver = Neo4jDriver(
    uri="bolt://localhost:7687",
    user="neo4j",
    password="password"
)

try:
    async with driver.transaction() as tx:
        # All operations in this block are part of the same transaction
        await tx.run(
            "CREATE (e:Entity {uuid: $uuid, name: $name})",
            uuid="entity-1",
            name="Example Entity"
        )
        await tx.run(
            "CREATE (e:Entity {uuid: $uuid, name: $name})",
            uuid="entity-2",
            name="Another Entity"
        )
        # Transaction commits automatically on success
        # Rolls back automatically on exception
except Exception as e:
    print(f"Transaction failed: {e}")
finally:
    await driver.close()

Azure OpenAI Integration

import os
from openai import AsyncOpenAI
from graphiti_core import Graphiti
from graphiti_core.driver.neo4j_driver import Neo4jDriver
from graphiti_core.llm_client.azure_openai_client import AzureOpenAILLMClient
from graphiti_core.embedder.azure_openai import AzureOpenAIEmbedderClient
from graphiti_core.llm_client.config import LLMConfig

# Azure OpenAI setup
azure_client = AsyncOpenAI(
    base_url=f"{os.environ['AZURE_OPENAI_ENDPOINT']}/openai/v1/",
    api_key=os.environ['AZURE_OPENAI_API_KEY']
)

llm_client = AzureOpenAILLMClient(
    azure_client=azure_client,
    config=LLMConfig(
        model="gpt-4.1",
        small_model="gpt-4.1"
    )
)

embedder_client = AzureOpenAIEmbedderClient(
    azure_client=azure_client,
    model="text-embedding-3-small"
)

# Initialize with Neo4j and Azure OpenAI
graphiti = Graphiti(
    uri="bolt://localhost:7687",
    user="neo4j",
    password="password",
    llm_client=llm_client,
    embedder=embedder_client
)

Environment Variables

Common environment variables for Neo4j configuration:

NEO4J_URI=bolt://localhost:7687
NEO4J_USER=neo4j
NEO4J_PASSWORD=password
NEO4J_DATABASE=neo4j

Load in your application:

import os
from dotenv import load_dotenv
from graphiti_core.driver.neo4j_driver import Neo4jDriver

load_dotenv()

driver = Neo4jDriver(
    uri=os.environ.get("NEO4J_URI", "bolt://localhost:7687"),
    user=os.environ.get("NEO4J_USER", "neo4j"),
    password=os.environ.get("NEO4J_PASSWORD", "password"),
    database=os.environ.get("NEO4J_DATABASE", "neo4j")
)

Thread Safety

The Neo4jDriver is thread-safe and uses connection pooling. You can safely:

• Share a single driver instance across multiple coroutines
• Execute concurrent queries
• Create multiple sessions from the same driver

import asyncio

driver = Neo4jDriver(
    uri="bolt://localhost:7687",
    user="neo4j",
    password="password"
)

# Execute queries concurrently
results = await asyncio.gather(
    driver.execute_query("MATCH (n:Entity) RETURN count(n)"),
    driver.execute_query("MATCH (e:Episode) RETURN count(e)"),
    driver.execute_query("MATCH (c:Community) RETURN count(c)")
)

Related

Drivers Overview

Learn about the driver architecture

Graphiti

Main Graphiti class documentation

FalkorDB Driver

Alternative Redis-based driver

Kuzu Driver

Embedded analytics driver