Skip to main content
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)")
)

Drivers Overview

Learn about the driver architecture

Graphiti

Main Graphiti class documentation

FalkorDB Driver

Alternative Redis-based driver

Kuzu Driver

Embedded analytics driver

Build docs developers (and LLMs) love

Get started for freeTalk to us