Skip to main content

​
Adding a New Graph Database to cognee

This guide describes how to integrate a new graph database engine into cognee.

​
Repository Options

Cognee used both the core and the community repositories to host graph-database adapters. 🚨 From now on every graph-database adapter – except Kuzu – will live in the cognee-community repository. NetworkX has already been migrated and the remaining adapters will follow shortly. Therefore all new adapter contributions must target the community repository.
The core repository will keep only the built-in Kuzu integration.

​
For Community Repository

To add a new adapter to cognee-community:
  1. Fork and clone the cognee-community repository
  2. Create your adapter in packages/<engine_name>/cognee_community_graph_adapter_<engine_name>/
  3. Inside that directory add __init__.py, <engine_name>_adapter.py, and register.py (see the Redis example).
  4. At the package root packages/<engine_name>/ add __init__.py, pyproject.toml, and README.md.
  5. Submit a pull request to the community repository
Below are the recommended steps in more detail.

​
Why cognee-community?

cognee-community is the extension hub for Cognee.
Anything that is not part of the core lives here—adapters for third-party databases, pipelines, community contributed additional tasks, etc.
Placing your adapter in this repository means:
  • Your code is released under the community license and can evolve independently of the core.
  • It can be installed with pip install cognee-community-graph-adapter-(engine_mane) without pulling in heavyweight drivers for users who don’t need them. For example, for NetworkX it is pip install cognee-community-graph-adapter-networkx
  • These packages can be called with cognee core package using the registration step described below.
If you are unfamiliar with the layout, have a look at the existing folders under packages/* in the community repo—each sub-folder represents a separate provider implemented in exactly the way you are about to do.

​
1. Implement the Adapter

File: packages/graph/<engine_name>/cognee_community_graph_adapter_<engine_name>/<engine_name>_adapter.py
Your adapter must subclass GraphDBInterface, implementing all required CRUD and utility methods (e.g., add_node, add_edge, extract_node, etc.). Here is a sample skeleton with placeholders: 
"""Adapter for <engine_name> graph database."""

import json
import asyncio
from typing import Dict, Any, List, Optional, Tuple

from cognee.shared.logging_utils import get_logger
from cognee.infrastructure.databases.graph.graph_db_interface import GraphDBInterface

logger = get_logger()

class <EngineName>Adapter(GraphDBInterface):
    """Adapter for <engine_name> graph database operations."""

    def __init__(
        self,
        graph_database_url: str = "",
        graph_database_username: Optional[str] = None,
        graph_database_password: Optional[str] = None,
    ):
        self.graph_database_url = graph_database_url
        self.graph_database_username = graph_database_username
        self.graph_database_password = graph_database_password
        self.connection = None

    async def query(self, query: str, params: Optional[Dict[str, Any]] = None) -> List[Tuple]:
        """Execute an async query.
        
        If your graph database library provides an async SDK, call it directly here.
        If it only provides a synchronous client, you can run the call via
        `loop.run_in_executor()` or a similar technique to avoid blocking the event loop.
        """

        loop = asyncio.get_running_loop()
        params = params or {}

        def blocking_query():
            try:
                # Example usage with your driver
                # cursor = self.connection.execute(query, params)
                # results = cursor.fetchall()
                return []
            except Exception as e:
                logger.error(f"<engine_name> query execution failed: {e}")
                raise

        return await loop.run_in_executor(self.executor, blocking_query)

    # -- Example: Add a node
    async def add_node(self, node_data: Any) -> None:
        """Add a single node to <engine_name>."""
        # Implement logic:
        # 1. Extract relevant fields (id, text, type, properties, etc.).
        # 2. Construct a CREATE query.
        # 3. Call self.query(query_str, params).
        pass

    # -- Example: Retrieve all nodes/edges
    async def get_graph_data(self) -> Tuple[List, List]:
        """Retrieve all nodes and edges from <engine_name>."""
        # Return (nodes, edges) where each node is `(node_id, properties_dict)`
        # and each edge is `(source_id, target_id, relationship_label, properties_dict)`.
        return ([], [])

    # -- Additional methods (delete_node, add_edge, etc.) ...
Keep the method signatures consistent with GraphDBInterface. Reference the KuzuAdapter or the Neo4jAdapter for a more comprehensive example.

​
2. Test with a Dedicated Script

Your contribution should have an example showcasing how this integration should be configured and used.
File: packages/graph/engine_name/examples/example.py
Create a script that loads cognee and the integration package, registers it to use your new <engine_name> provider, and runs basic usage checks (e.g., adding data, searching, pruning). For example: 
import sys
import asyncio
import pathlib
from os import path

# NOTE: Importing the register module we let cognee know it can use the Networkx adapter
import packages.graph.networkx.register

async def main():
    from cognee import config, prune, add, cognify, search, SearchType

    system_path = pathlib.Path(__file__).parent
    config.system_root_directory(path.join(system_path, ".cognee-system"))
    config.data_root_directory(path.join(system_path, ".cognee-data"))

    config.set_graph_db_config({
        "graph_database_provider": "engine-name",
    })

    await prune.prune_data()
    await prune.prune_system()

    text = """
    Natural language processing (NLP) is an interdisciplinary
    subfield of computer science and information retrieval.
    """

    await add(text)

    await cognify()

    query_text = "Tell me about NLP"

    search_results = await search(query_type=SearchType.GRAPH_COMPLETION, query_text=query_text)

    for result_text in search_results:
        print(result_text)

if __name__ == "__main__":
    asyncio.run(main())

​
3. Create a Test Workflow

File: .github/workflows/engine_name/test_engine_name.yml
Create a GitHub Actions workflow to run your integration tests. This ensures any pull requests that modify your new engine (or the shared graph code) will be tested automatically. See an example here. 
name: test | <engine_name>

on:
  workflow_dispatch:
  pull_request:
    types: [labeled, synchronize]

concurrency:
  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
  cancel-in-progress: true

env:
  RUNTIME__LOG_LEVEL: ERROR

jobs:
  run_<engine_name>_integration_test:
    name: test
    runs-on: ubuntu-22.04

    defaults:
      run:
        shell: bash

    steps:
      - name: Check out
        uses: actions/checkout@master

      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.11.x'

      - name: Install Poetry
        uses: snok/install-poetry@v1.4.1
        with:
          virtualenvs-create: true
          virtualenvs-in-project: true
          installer-parallel: true

      - name: Install dependencies
        # If your pyproject.toml has an extra named '<engine_name>', use:
        run: poetry install -E <engine_name> --no-interaction

      - name: Run <engine_name> tests
        env:
          ENV: 'dev'
          LLM_MODEL: ${{ secrets.LLM_MODEL }}
          LLM_ENDPOINT: ${{ secrets.LLM_ENDPOINT }}
          LLM_API_KEY: ${{ secrets.LLM_API_KEY }}
          LLM_API_VERSION: ${{ secrets.LLM_API_VERSION }}
          EMBEDDING_MODEL: ${{ secrets.EMBEDDING_MODEL }}
          EMBEDDING_ENDPOINT: ${{ secrets.EMBEDDING_ENDPOINT }}
          EMBEDDING_API_KEY: ${{ secrets.EMBEDDING_API_KEY }}
          EMBEDDING_API_VERSION: ${{ secrets.EMBEDDING_API_VERSION }}
          GRAPH_DATABASE_PROVIDER: ''
          GRAPH_DATABASE_URL: ''
          GRAPH_DATABASE_PASSWORD: ''
        run: poetry run python ./cognee_community_graph_adapter_engine_name/examples/example.py
        working-directory: ./packages/graph/engine_name
Tips:
  • Rename <engine_name> appropriately.
  • Ensure your pyproject.toml has an extras entry for any new dependencies.

​
5. Poetry Extras

If your new graph engine requires a special Python client or system libraries, update: pyproject.toml: 
[tool.poetry.dependencies]
python = "^3.11"
cognee
your-graph-db-client

[tool.poetry.extras]
...

​
6. Final Checklist

  1. Implement your <EngineName>Adapter in packages/<engine_name>/cognee_community_graph_adapter_<engine_name>/<engine_name>_adapter.py.
  2. Add a register helper (register.py) and call it before configuring Cognee: 
    from cognee.infrastructure.databases.graph import use_graph_adapter

from .<engine_name>_adapter import <EngineName>Adapter

use_graph_adapter("engine_name", <EngineName>Adapter)
  3. Create a test or example script example.py.
  4. Create a test workflow: .github/workflows/engine_name/test_<engine_name>.yml.
  5. Add required dependencies to pyproject.toml extras.
  6. Open a PR to verify that your new integration passes CI.
That’s all! This approach keeps cognee’s architecture flexible, allowing you to swap in any graph DB provider easily. Review the previous implementations in the core and the community repos.

​
Join the Conversation!

Have questions about creating custom tasks? Join our community to discuss implementation strategies and best practices!