> ## Documentation Index > Fetch the complete documentation index at: https://docs.cognee.ai/llms.txt > Use this file to discover all available pages before exploring further. # Graph Database Integration > Add a new graph database adapter to Cognee. # 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](https://github.com/topoteretes/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](https://github.com/topoteretes/cognee-community): 1. **Fork and clone** the cognee-community repository 2. **Create your adapter** in `packages//cognee_community_graph_adapter_/` 3. Inside that directory add `__init__.py`, `_adapter.py`, and `register.py` (see the Redis example). 4. At the package root `packages//` 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/*`](https://github.com/topoteretes/cognee-community/tree/main/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//cognee_community_graph_adapter_/_adapter.py` Your adapter **must** subclass [`GraphDBInterface`](https://github.com/topoteretes/cognee/blob/ef1aecd835b1a2044eb724197bbef77f6dee5d3c/cognee/infrastructure/databases/graph/graph_db_interface.py#L127), implementing all required CRUD and utility methods (e.g., `add_node`, `add_edge`, `extract_node`, etc.). Here is a sample skeleton with placeholders: ```python theme={null} """Adapter for 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 Adapter(GraphDBInterface): """Adapter for 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" 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 .""" # 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 .""" # 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](https://github.com/topoteretes/cognee/tree/main/cognee/infrastructure/databases/graph/kuzu) or the [Neo4jAdapter](https://github.com/topoteretes/cognee/tree/main/cognee/infrastructure/databases/graph/neo4j_driver) for a more comprehensive example. > **Adapter instance reuse**: Cognee's graph engine factory caches adapter instances keyed by their configuration parameters. Multiple calls with identical settings return the **same adapter object**. Design your adapter to be safe for reuse — avoid per-instance mutable state that cannot be safely shared, and prefer lazy or thread-safe initialization where state is required. When an entry is evicted (e.g. cache eviction or `cache_clear`), the factory calls your adapter's `close()` — deferred until every leased reference to that instance is released — so implement `close()` idempotently; if it raises, the error is logged and swallowed rather than propagated to the caller. > **Optional `get_id_filtered_graph_data`**: This method is **not** part of `GraphDBInterface`, so it is optional. If you implement it, graph-completion searches project only the vector-search neighborhood instead of loading the full graph via `get_graph_data()`; if you omit it, Cognee falls back to `get_graph_data()`. The contract is edge-driven and matches the built-in Ladybug, Neo4j, and Postgres adapters: given `target_ids`, return `(nodes, edges)` where `edges` are every edge with either endpoint in `target_ids`, and `nodes` are all endpoint nodes of those edges (same `(node_id, properties)` / `(source_id, target_id, relationship_label, properties)` shapes as `get_graph_data()`). Return `([], [])` when `target_ids` is empty; Cognee also falls back to the full graph if the filtered result comes back empty. > > ```python theme={null} > async def get_id_filtered_graph_data(self, target_ids: List[str]) -> Tuple[List, List]: > """Return the subgraph touching target_ids: edges with either endpoint in > the set, plus all endpoint nodes of those edges.""" > if not target_ids: > return [], [] > # ... query edges where source/target is in target_ids, then their endpoint nodes ... > return nodes, edges > ``` ## 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 `` provider, and runs basic usage checks (for example, remembering data, recalling it, and pruning isolated test state). For example: ```python theme={null} 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, remember, recall, 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. """ dataset_name = "test_dataset" await remember(text, dataset_name=dataset_name) query_text = "Tell me about NLP" search_results = await recall( query_text, SearchType.GRAPH_COMPLETION, datasets=[dataset_name], ) 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](https://github.com/topoteretes/cognee-community/blob/main/.github/workflows/test_opensearch.yml). ```yaml theme={null} name: test | 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__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 '', use: run: poetry install -E --no-interaction - name: Run 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 `` 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`**: ```toml theme={null} [tool.poetry.dependencies] python = "^3.11" cognee your-graph-db-client [tool.poetry.extras] ... ``` *** ## 6. Final Checklist 1. **Implement** your `Adapter` in `packages//cognee_community_graph_adapter_/_adapter.py`. 2. **Add** a **register** helper (`register.py`) and call it **before** configuring Cognee: ```python theme={null} from cognee.infrastructure.databases.graph import use_graph_adapter from ._adapter import Adapter use_graph_adapter("engine_name", Adapter) ``` 3. **Create** a test or example script `example.py`. 4. **Create** a test workflow: `.github/workflows/engine_name/test_.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](https://github.com/topoteretes/cognee/tree/main/cognee/infrastructure/databases) and the [community](https://github.com/topoteretes/cognee-community/tree/main/packages) repos. #### Join the Conversation! Have questions about creating custom tasks? Join our community to discuss implementation strategies and best practices!