The Cognee Cloud SDK is the recommended way to interact with Cognee Cloud programmatically. It is open source and provides a clean Python interface for working with the Cognee Cloud platform. In particular, it makes it easy to perform core Cognee operations—add, cognify, memify, and search—that power Cognee Cloud. You’ll need an API key to get started.
While you can call Cognee Cloud API endpoints directly, the SDK is recommended for its error handling, type safety, and overall developer experience.
# 1. Create a virtual environment (optional)
uv venv && source .venv/bin/activate
# 2. Install the SDK
uv pip install cogwit-sdk
# 3. Store your key and base URL
export COGWIT_API_KEY="<your key>"
Complete example
Let’s jump straight in with a full code example of Cognee Cloud in action. We’ll then go over the code piece by piece and explain all the relevant parts in detail.
import asyncio
import os
from cogwit_sdk import cogwit, CogwitConfig
from cogwit_sdk.responses import (
AddResponse,
CognifyResponse,
CombinedSearchResult,
SearchResult,
)
# Create configuration with your API key
cogwit_config = CogwitConfig(
api_key=os.getenv("COGWIT_API_KEY", ""),
)
# Create the client instance
cogwit_instance = cogwit(cogwit_config)
async def main():
# Add data to your dataset
result = await cogwit_instance.add(
data="Cognee Cloud automates knowledge graph creation in the cloud.",
dataset_name="demo_dataset",
)
print(f"Added data: {result.status}")
dataset_id = result.dataset_id
# Transform data into knowledge graph
cognify_result = await cogwit_instance.cognify(
dataset_ids=[dataset_id],
)
print(f"Cognify status: {cognify_result[str(dataset_id)].status}")
# Search with graph completion
search_results = await cogwit_instance.search(
query_text="What does Cognee Cloud automate?",
query_type=cogwit_instance.SearchType.GRAPH_COMPLETION,
)
for result in search_results:
print(f" {result.search_result}")
# Search for raw chunks
chunk_results = await cogwit_instance.search(
query_text="What does Cognee Cloud automate?",
query_type=cogwit_instance.SearchType.CHUNKS,
)
for result in chunk_results:
print(f" {result.search_result[0]['text']}")
asyncio.run(main())
What just happened
Client setup
cogwit_config = CogwitConfig(
api_key=os.getenv("COGWIT_API_KEY", ""),
)
cogwit_instance = cogwit(cogwit_config)
async def main() and run it with asyncio.run(main()).
Adding data
result = await cogwit_instance.add(
data="Cognee Cloud automates knowledge graph creation in the cloud.",
dataset_name="demo_dataset",
)
add operation uploads text to Cognee Cloud and schedules preprocessing. All data is organized by dataset for proper isolation. It returns an AddResponse with:
status - Whether the operation completed successfullydataset_id - Unique identifier for your dataset (save this!)dataset_name - The name you provided
Cognifying data
cognify_result = await cogwit_instance.cognify(
dataset_ids=[dataset_id],
)
cognify operation transforms your data into a knowledge graph. It returns a CognifyResponse that maps dataset IDs to their processing status. Wait for PipelineRunCompleted before searching.
Searching your data
# Graph completion search
search_results = await cogwit_instance.search(
query_text="What does Cognee Cloud automate?",
query_type=cogwit_instance.SearchType.GRAPH_COMPLETION,
)
search operation queries your knowledge graph. By default it uses GRAPH_COMPLETION, but you can use any of the Cognee search types. See also Search Basics for detailed parameter explanations.
Next steps
