Give your Claude Agent SDK agents persistent memory powered by cognee, build knowledge graphs, search with natural language, and maintain context across sessions.
Before using the integration, configure your environment variables:
Copy
export LLM_API_KEY="your-openai-api-key-here" # for cognee
The Claude Agent SDK handles authentication automatically through Cursor or via OAuth/API key on first run.
Add memory tools to your Claude agent:
Copy
import asynciofrom claude_agent_sdk import ( create_sdk_mcp_server, ClaudeAgentOptions, ClaudeSDKClient, AssistantMessage, TextBlock,)from cognee_integration_claude import add_tool, search_toolasync def main(): # Create MCP server with cognee tools server = create_sdk_mcp_server( name="memory-tools", version="1.0.0", tools=[add_tool, search_tool] ) options = ClaudeAgentOptions( mcp_servers={"tools": server}, allowed_tools=["mcp__tools__add_tool", "mcp__tools__search_tool"], ) # Store information async with ClaudeSDKClient(options=options) as client: await client.query("Remember: Acme Corp, healthcare, $1.2M contract") async for msg in client.receive_response(): pass # Retrieve information async with ClaudeSDKClient(options=options) as client: await client.query("What healthcare contracts do we have?") async for msg in client.receive_response(): if isinstance(msg, AssistantMessage): for block in msg.content: if isinstance(block, TextBlock): print(block.text)asyncio.run(main())
from cognee_integration_claude import get_sessionized_cognee_tools# Session 1: Store informationadd_tool, search_tool = get_sessionized_cognee_tools("user-123")server_1 = create_sdk_mcp_server( name="memory-tools", version="1.0.0", tools=[add_tool, search_tool])options_1 = ClaudeAgentOptions( mcp_servers={"tools": server_1}, allowed_tools=["mcp__tools__add_tool", "mcp__tools__search_tool"],)async with ClaudeSDKClient(options=options_1) as client: await client.query("I'm working on authentication") async for msg in client.receive_response(): pass# Session 2: Different instance, same memoryadd_tool, search_tool = get_sessionized_cognee_tools("user-123")server_2 = create_sdk_mcp_server( name="memory-tools", version="1.0.0", tools=[add_tool, search_tool])options_2 = ClaudeAgentOptions( mcp_servers={"tools": server_2}, allowed_tools=["mcp__tools__add_tool", "mcp__tools__search_tool"],)async with ClaudeSDKClient(options=options_2) as client: await client.query("What was I working on?") async for msg in client.receive_response(): # Returns: "authentication" pass
Build domain knowledge incrementally over multiple sessions:
Copy
# Pre-load documents into cogneeimport cogneefor doc_path in document_paths: with open(doc_path, 'r') as f: content = f.read() await cognee.add(content)await cognee.cognify()# Query across all documentsasync with ClaudeSDKClient(options=options) as client: await client.query("Find information about contract terms") async for msg in client.receive_response(): pass
Context-Aware Assistance
Maintain user context across work sessions:
Copy
# Mondayasync with ClaudeSDKClient(options=options) as client: await client.query("Debugging payment flow") async for _ in client.receive_response(): pass# Wednesdayasync with ClaudeSDKClient(options=options) as client: await client.query("What was I debugging?") async for msg in client.receive_response(): pass
Multi-Tenant Applications
Isolate data per user/organization while sharing global knowledge: