> ## 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, and **branch from `main`** — unlike the core Cognee repo, cognee-community has no `dev` branch. 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. **Run the shared graph conformance suite** in `packages/shared/contract_suite/graph_contract.py` against your adapter. 6. **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()`. For capacity eviction the close is deferred until every leased reference to that instance is released, but the dataset-queue teardown and the idle reaper force-close immediately even while idle references are still held (holders transparently re-resolve to a fresh instance on next use) — so implement `close()` idempotently and safe to run while stale references remain; 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 > ``` > **Declaring Cypher support**: `GraphDBInterface` declares `supports_cypher_queries: bool = True`, so adapters are assumed to speak Cypher through `query()`. Override it to `False` on your adapter class when `query()` executes something else — the built-in Postgres and Turso adapters do this because their `query()` runs SQL against the graph tables. `SearchType.CYPHER` and `SearchType.NATURAL_LANGUAGE` then raise `SearchTypeNotSupported` (naming your adapter class) instead of handing your backend a Cypher string it cannot parse. Keep the flag on the class rather than setting it in `__init__`: tests and tooling read the capability directly off the adapter class without instantiating it (no database connection needed), as Cognee's own adapter tests do. > > ```python theme={null} > class Adapter(GraphDBInterface): > # ``query()`` executes 's own query language, not Cypher. > supports_cypher_queries = False > ``` > **Optional `get_triplets_batch` — order before you paginate**: `GraphDBInterface.get_triplets_batch(offset, limit)` is an optional extension; the base implementation raises `NotImplementedError`, and only adapters that override it can back Memify's triplet-embedding pipeline (the built-in Ladybug, Neo4j, Postgres, and Turso adapters do). If you implement it, your query **must apply a total ordering before the offset/limit**, because `get_triplet_datapoints` reads the whole graph with a single offset loop, advancing the offset by each batch's size until a batch comes back short or empty, and that loop is only exhaustive if every call slices the same stable sequence. Paginating an unordered result set silently skips and duplicates rows. Sort on `(source node id, target node id, relationship name)` to match the built-in adapters. In Cypher this means putting `ORDER BY` in a `WITH` clause **ahead of** `SKIP`/`LIMIT` rather than after `RETURN`, so the skip applies to an already-ordered stream. > > ```cypher theme={null} > MATCH (start_node:Node)-[relationship:EDGE]->(end_node:Node) > WITH start_node, relationship, end_node > ORDER BY start_node.id, end_node.id, relationship.relationship_name > SKIP $offset LIMIT $limit > RETURN start_node, relationship, end_node > ``` > **Chunk bulk writes**: Do not send an entire `add_nodes` / `add_edges` payload as one statement — a large ingest (tens of thousands of nodes or edges) can then exceed a per-statement or per-call deadline and never finish. Define a module-level `_WRITE_CHUNK_SIZE` and loop over the rows in slices of that size, issuing one statement per chunk; the built-in adapters use 2000 (Ladybug), 1000 (Postgres), and 500 (Turso), so pick a bound that suits your backend. Write each chunk as an idempotent `MERGE`/upsert: chunks are separate statements, so a run can fail partway through, and idempotent writes plus Cognee's pipeline rollback ledger make the partial progress safe to re-apply. Chunking is internal to the adapter and does not change how many data points the pipeline batches per call. > > ```python theme={null} > _WRITE_CHUNK_SIZE = 2000 > > total = len(rows) > for start in range(0, total, _WRITE_CHUNK_SIZE): > chunk = rows[start : start + _WRITE_CHUNK_SIZE] > await self.query(merge_query, {"nodes": chunk}) > ``` > **Match edge endpoints through an index**: In a bulk edge write, bind each endpoint with a property-map match on the indexed id (`MATCH (from:Node {id: edge.from_id})`) rather than a cartesian `MATCH (from:Node), (to:Node) WHERE from.id = ... AND to.id = ...`. The cartesian form plans as a scan over the node table for every edge, which degrades badly as the graph grows; the property-map form is a primary-key index seek. > > Whichever form you use, and in existence checks as much as in writes, match on the `id` **property** Cognee stores — the string UUID the pipeline hands you — and never on a backend-internal node identifier such as Neo4j's `id(n)` or an autoincrementing row number. Those identifiers are a different value (and usually a different type) from the id Cognee passes, so the comparison does not error; it silently matches nothing. The `neo4j` adapter's `has_edges` compared `id(a)` against the string UUID for exactly this reason and reported every edge as absent, which made the cognify dedup step below write a fresh copy of every edge on each re-cognify. > **Let existence checks fail loudly**: In `has_edges` — and in any read or existence-check method whose empty result is meaningful — do not catch a store error and return an empty list. An empty return must mean "the backend answered, and nothing matched"; it must never mean "the query failed". `has_edges` is the batch check the cognify dedup uses to decide which edges are new, so a swallowed failure tells it that none of the edges exist, it writes all of them, those writes fail against the same broken store, and the run finishes reporting success with nothing persisted. Log the error and re-raise, as the `neo4j` adapter does, and reserve `[]` for the genuine empty-**input** short-circuit. > > A *successful* check has a shape contract too: return the **subset of the input triples that exist**, as `(source_id, target_id, relationship_name)` strings. Not booleans, and not a list aligned one-to-one with the input — the caller, `find_existing_edge_identities`, unpacks each returned item into three values and treats the result as the set of edges to skip, so a per-input boolean list is not a compatible substitute. > > ```python theme={null} > async def has_edges(self, edges: List[Tuple[str, str, str]]) -> List[Tuple[str, str, str]]: > if not edges: # empty input: an answer, not a failure > return [] > try: > results = await self.query(query, {"edges": edge_params}) > return [(str(row[0]), str(row[1]), str(row[2])) for row in results] > except Exception as e: > logger.error(f"Failed to check edges in batch: {e}") > raise # a failed existence check is not an empty existence check > ``` > **Carry edge properties out of `get_connections`**: `get_connections(node_id)` returns `(source_node, edge, target_node)` triples for every edge into or out of the node, and the middle element must carry the edge's **stored properties** alongside its `relationship_name`. Consumers read those properties by key, and a missing key does not raise — it produces a quietly wrong answer. Cognee's document-deletion path derives the `EdgeType` vector-row id of a chunk's `contains` edges from `edge["edge_text"]`; when that key is absent it falls back to the relationship name, computes a different id, and leaves the real vector rows behind. Watch for driver helpers that flatten a relationship and discard its properties on the way out: Neo4j's `result.data()` reduces a relationship to `(start_props, type, end_props)`, so the adapter has to request `properties(relation)` explicitly and merge it in. > > ```python theme={null} > edge = {"relationship_name": relationship_type} > edge.update(relationship_properties or {}) # edge_text, weights, timestamps, ... > connections.append((source_properties, edge, target_properties)) > ``` ## 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. **Run** the shared conformance suite in `packages/shared/contract_suite/graph_contract.py` against your adapter — it is the common contract every community graph adapter is expected to satisfy. 4. **Register** a dataset-database handler from the same `register.py` if your backend can isolate storage per user + dataset: ```python theme={null} from cognee.infrastructure.databases.dataset_database_handler import ( use_dataset_database_handler, ) use_dataset_database_handler("engine_name", DatasetDatabaseHandler, "engine_name") ``` Without a handler, your adapter can only be used with `ENABLE_BACKEND_ACCESS_CONTROL=false`. 5. **Create** a test or example script `example.py`. 6. **Create** a test workflow: `.github/workflows/engine_name/test_.yml`. 7. **Add** required dependencies to `pyproject.toml` extras. 8. **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!