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.
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.
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 jsonimport asynciofrom typing import Dict, Any, List, Optional, Tuplefrom cognee.shared.logging_utils import get_loggerfrom cognee.infrastructure.databases.graph.graph_db_interface import GraphDBInterfacelogger = 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.
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 sysimport asyncioimport pathlibfrom os import path# NOTE: Importing the register module we let cognee know it can use the Networkx adapterimport packages.graph.networkx.registerasync 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())
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.
Implement your <EngineName>Adapter in packages/<engine_name>/cognee_community_graph_adapter_<engine_name>/<engine_name>_adapter.py.
Add a register helper (register.py) and call it before configuring Cognee:
from cognee.infrastructure.databases.graph import use_graph_adapterfrom .<engine_name>_adapter import <EngineName>Adapteruse_graph_adapter("engine_name", <EngineName>Adapter)
Create a test or example script example.py.
Create a test workflow: .github/workflows/engine_name/test_<engine_name>.yml.
Add required dependencies to pyproject.toml extras.
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.
