Ctx Cognee¶

ctx_cognee — mvp.ctx_cognee

Cluster: Context & Retrieval | Type: component | MCP Tools: 26

Overview¶

Knowledge graph memory that extracts entities and relations from free text. The MCP production profile (COGNEE_MCP_PROFILE=production) uses the real Cognee SDK path by default and fails action operations until Cognee readiness passes. The local/default profile is intentionally lightweight: it uses SQLite audit logging plus capitalised noun-phrase heuristics and co-occurrence relation detection when Cognee is not enabled.

Reliability status: qualified-draft. The local/default path is useful for memory hints and auditability, but live Cognee add/cognify/search quality remains unverified without configured Cognee, graph, vector, and LLM backends. Relevance-floor claims must remain qualified until live evidence exists.

Surface contract: the direct block exposes 4 stable ops (add, query, reset, list). The MCP block exposes 25 stable data-plane tools plus block-only list_patterns introspection. status, info, and get_config expose per-op capability records with availability, fallback mode, readiness, destructive classification, and confirmation requirements.

When to use:

Building a lightweight knowledge graph from unstructured documents during a pipeline run
Querying which entities and relations are related to a given concept
Grounding agent reasoning with extracted real-world entity links

Example:

from mvp.ctx_cognee import CtxCogneeBlock, CogneeInput

block = CtxCogneeBlock(name="cognee")
block.infer(CogneeInput(op="add", text="Python is a language. Guido van Rossum created Python."))
result = block.infer(CogneeInput(op="query", query="Python"))
# result.value.entities -> ["Python", "Guido van Rossum"]; relations populated

Production profile:

set COGNEE_MCP_PROFILE=production
set LLM_API_KEY=<your-cognee-llm-key>

In production profile, call status, info, or get_config first and check metadata.cognee_readiness.ok. Action operations such as add, cognify, and search fail with a readiness error until the Cognee SDK is importable and the minimum LLM configuration is present.

Canonical degradation envelope:

Outputs preserve degraded and degradation_reason and also surface completion_state, warning_card, and evidence. SDK exception fallbacks include evidence.backend_attempted, evidence.backend_used, evidence.fallback_used, and evidence.exception_class when applicable. Local heuristic extraction/search returns completion_state="degraded_complete", warning_card.summary="heuristic local fallback, not graph-backed extraction", and evidence.backend_used="local_heuristic".

End-user caveats:

Local/default mode is useful for quick memory hints, but it is not a full knowledge graph. It relies on simple heuristics and may miss entities, create false positives, or infer weak relations.
Cognee-backed production mode can send ingested text to the configured LLM provider. Do not ingest regulated or confidential data unless the chosen provider, keys, retention policy, and deployment environment are approved for that data.
The direct CtxCogneeBlock stores fallback state in memory. The MCP block records audit metadata in SQLite, while durable graph/vector persistence belongs to Cognee's configured stores.
delete, delete_dataset, prune_data, prune_system, and delete_feedback are destructive operations. The block and MCP surfaces block them unless confirm=True and a non-empty reason are supplied; missing confirmation returns completion_state="blocked" with G6_E_COGNEE_DESTRUCTIVE_CONFIRMATION_REQUIRED.
Run a live add -> cognify -> search smoke test after changing Cognee LLM, graph, vector, or database settings.

Works well with: ctx_rag, ctx_colbert, goal_engine

Public API¶

`KGRerankDecision`¶

Validated advisory rerank verdict over a returned kg result set.

Field	Type	Default
`ordered_indices`	`tuple[int, ...]`	required
`dropped_indices`	`tuple[int, ...]`	()
`rationale`	`str`	''
`eligible_fingerprint`	`str`	''
`confidence`	`float`	0.0
`degraded`	`bool`	False
`raw_response`	`str`	''

`LLMKGRerankRuntime`¶

Provider-neutral kg-result-rerank runtime backed by G6's LLM caller.

Constructor:

Parameter	Type	Default
`llm`	`LLMCaller \\| None`	None

Methods:

`rerank(query: str, records: list[Any]) -> KGRerankDecision`¶

`CtxCogneeRerankPatternRuntime`¶

Stateless, load-bearing returned-set-ceiling enforcement.

Methods:

`enforce_eligibility(decision: KGRerankDecision, records: list[Any]) -> tuple[list[Any], bool, bool]`¶

`CtxCogneePlanner`¶

Runtime-first advisory kg-rerank facade with returned-order fallback.

Constructor:

Parameter	Type	Default
`runtime`	`KGRerankRuntime \\| None`	None
`pattern_runtime`	`CtxCogneeRerankPatternRuntime \\| None`	None

Methods:

`rerank(query: str, records: list[Any]) -> list[Any]`¶

`CtxCogneeBlock(AIBlock[CogneeInput, CogneeOutput, dict])`¶

Knowledge graph memory.

Field	Type	Default
`name`	`str`	'ctx_cognee'
`resource_bounds`	`ResourceBounds \\| None`	None
`usage`	`ResourceUsage`	field(default_factory=ResourceUsage)
`agentic_planner`	`CtxCogneePlanner \\| None`	None

Methods:

`infer(data: CogneeInput) -> Result[CogneeOutput]`¶

`EntityRelation(BaseModel)`¶

A single entity-relation-entity triple.

Field	Type	Default
`subject`	`str`	required
`relation`	`str`	required
`obj`	`str`	required

`CogneeInput(BaseModel)`¶

Input to CtxCogneeBlock.

Field	Type	Default
`op`	`Literal['add', 'query', 'reset', 'list']`	required
`text`	`str`	''
`query`	`str`	''
`agentic_rerank`	`bool \\| None`	None
`run_mode`	`Literal['beta', 'production']`	'beta'
`reviewer_signature`	`str`	''

`CogneeOutput(BaseModel)`¶

Output from CtxCogneeBlock.

Field	Type	Default
`op`	`str`	required
`entities`	`list[str]`	Field(default_factory=list)
`relations`	`list[EntityRelation]`	Field(default_factory=list)
`message`	`str`	''
`degraded`	`bool`	False
`degradation_reason`	`str`	''
`completion_state`	`str`	'complete'
`warning_card`	`dict`	Field(default_factory=dict)
`evidence`	`dict`	Field(default_factory=dict)
`request_id`	`str`	''
`run_id`	`str`	''
`available_with`	`list[str]`	Field(default_factory=list)
`agentic_evidence`	`dict`	Field(default_factory=dict)

`CtxCogneeMCPBlock(AIBlock[MCPCogneeInput, MCPCogneeOutput, dict])`¶

Full-featured cognee knowledge graph block with SQLite audit logging.

Field	Type	Default
`name`	`str`	'ctx_cognee_mcp'
`state`	`dict \\| None`	None
`db_path`	`str`	':memory:'
`resource_bounds`	`ResourceBounds \\| None`	None
`usage`	`ResourceUsage`	field(default_factory=ResourceUsage)
`agentic_planner`	`CtxCogneePlanner \\| None`	None

Methods:

`infer(data: MCPCogneeInput) -> Result[MCPCogneeOutput]`¶

`MCPCogneeRecord(BaseModel)`¶

Field	Type	Default
`id`	`str`	required
`record_type`	`str`	required
`key`	`str`	required
`value`	`str`	required
`tags`	`list[str]`	Field(default_factory=list)
`timestamp`	`str`	required
`metadata`	`dict[str, Any]`	Field(default_factory=dict)

`MCPCogneeInput(BaseModel)`¶

Field	Type	Default
`op`	`Literal['add', 'cognify', 'search', 'memify', 'delete', 'search_graph', 'search_rag', 'search_chunks', 'search_summaries', 'search_code', 'create_dataset', 'list_datasets', 'get_dataset', 'delete_dataset', 'configure', 'get_config', 'status', 'get_graph_url', 'prune_data', 'prune_system', 'get_session', 'add_feedback', 'delete_feedback', 'search_advanced', 'info', 'list_patterns']`	required
`text`	`str`	''
`query`	`str`	''
`dataset_name`	`str`	''
`dataset_id`	`str`	''
`data_id`	`str`	''
`node_set`	`str`	''
`search_type`	`str`	''
`top_k`	`int`	10
`session_id`	`str`	''
`custom_prompt`	`str`	''
`config_key`	`str`	''
`config_value`	`str`	''
`config_dict`	`str`	''
`feedback_text`	`str`	''
`feedback_id`	`str`	''
`prune_graph`	`bool`	True
`prune_vector`	`bool`	True
`prune_metadata`	`bool`	False
`prune_cache`	`bool`	True
`confirm`	`bool`	False
`reason`	`str`	''
`tags`	`list[str]`	Field(default_factory=list)
`limit`	`int`	50
`agentic_rerank`	`bool \\| None`	None
`run_mode`	`Literal['beta', 'production']`	'beta'
`reviewer_signature`	`str`	''

`MCPCogneeOutput(BaseModel)`¶

Field	Type	Default
`op`	`str`	required
`key`	`str`	''
`value`	`str`	''
`found`	`bool`	False
`count`	`int`	0
`records`	`list[MCPCogneeRecord]`	Field(default_factory=list)
`retrieved`	`list[str]`	Field(default_factory=list)
`scores`	`list[float]`	Field(default_factory=list)
`summary`	`str`	''
`message`	`str`	''
`results`	`list[dict[str, Any]]`	Field(default_factory=list)
`metadata`	`dict[str, Any]`	Field(default_factory=dict)
`degraded`	`bool`	False
`degradation_reason`	`str`	''
`completion_state`	`str`	'complete'
`warning_card`	`dict[str, Any]`	Field(default_factory=dict)
`evidence`	`dict[str, Any]`	Field(default_factory=dict)
`request_id`	`str`	''
`run_id`	`str`	''
`available_with`	`list[str]`	Field(default_factory=list)
`agentic_evidence`	`dict`	Field(default_factory=dict)

`CogneeStore`¶

Sync SQLite audit store with 5 tables.

Constructor:

Parameter	Type	Default
`db_path`	`str`	':memory:'

Methods:

`add_ingestion(dataset_name: str, text_preview: str, char_count: int, cognee_status: str, error: str = '') -> str`¶

`query_ingestions(dataset_name: str = '', limit: int = 50) -> list[dict[str, Any]]`¶

`add_cognify_run(dataset_name: str, status: str, duration_sec: float, error: str = '') -> str`¶

`query_cognify_runs(dataset_name: str = '', limit: int = 50) -> list[dict[str, Any]]`¶

`add_search(query: str, search_type: str, result_count: int, duration_sec: float) -> str`¶

`query_searches(limit: int = 50) -> list[dict[str, Any]]`¶

`upsert_dataset(name: str, description: str = '', item_count: int = 0, last_cognified: str = '') -> str`¶

`query_datasets(limit: int = 50) -> list[dict[str, Any]]`¶

`get_dataset(name: str) -> dict[str, Any] | None`¶

`add_session_event(session_id: str, action: str, feedback_text: str = '', metadata_json: str = '{}') -> str`¶

`query_sessions(session_id: str = '', limit: int = 50) -> list[dict[str, Any]]`¶

`text_search(query: str, top_k: int = 5) -> list[dict[str, Any]]`¶

TF-IDF search across all audit tables.

`count_all() -> dict[str, int]`¶

Functions¶

`agentic_planner_enabled(default_enabled: bool) -> bool`¶

Decide whether the agentic kg-rerank planner should be used.

`eligible_fingerprint(records: list[Any]) -> str`¶

sha256 over the returned record texts (order-sensitive).

`validate_kg_rerank_decision(decision: KGRerankDecision, n_eligible: int, expected_fingerprint: str) -> None`¶

Returned-set / anti-injection guard for a kg-rerank decision.

`applied_agentic_patterns() -> list[dict[str, Any]]`¶

Return compact metadata for ctx_cognee-applied vendored patterns.

`get_skill_catalog() -> CtxCogneeSkillCatalog`¶

MCP Tools¶

Operation	Source
`add`	cognee_mcp
`cognify`	cognee_mcp
`search`	cognee_mcp
`memify`	cognee_mcp
`delete`	cognee_mcp
`search_graph`	cognee_mcp
`search_rag`	cognee_mcp
`search_chunks`	cognee_mcp
`search_summaries`	cognee_mcp
`search_code`	cognee_mcp
`create_dataset`	cognee_mcp
`list_datasets`	cognee_mcp
`get_dataset`	cognee_mcp
`delete_dataset`	cognee_mcp
`configure`	cognee_mcp
`get_config`	cognee_mcp
`status`	cognee_mcp
`get_graph_url`	cognee_mcp
`prune_data`	cognee_mcp
`prune_system`	cognee_mcp
`get_session`	cognee_mcp
`add_feedback`	cognee_mcp
`delete_feedback`	cognee_mcp
`search_advanced`	cognee_mcp
`info`	cognee_mcp
`list_patterns`	cognee_mcp

Ctx Cognee¶

Overview¶

Public API¶

KGRerankDecision¶

LLMKGRerankRuntime¶

rerank(query: str, records: list[Any]) -> KGRerankDecision¶

CtxCogneeRerankPatternRuntime¶

enforce_eligibility(decision: KGRerankDecision, records: list[Any]) -> tuple[list[Any], bool, bool]¶

CtxCogneePlanner¶

rerank(query: str, records: list[Any]) -> list[Any]¶

CtxCogneeBlock(AIBlock[CogneeInput, CogneeOutput, dict])¶

infer(data: CogneeInput) -> Result[CogneeOutput]¶

EntityRelation(BaseModel)¶

CogneeInput(BaseModel)¶

CogneeOutput(BaseModel)¶

CtxCogneeMCPBlock(AIBlock[MCPCogneeInput, MCPCogneeOutput, dict])¶

infer(data: MCPCogneeInput) -> Result[MCPCogneeOutput]¶

MCPCogneeRecord(BaseModel)¶

MCPCogneeInput(BaseModel)¶

MCPCogneeOutput(BaseModel)¶

CogneeStore¶

add_ingestion(dataset_name: str, text_preview: str, char_count: int, cognee_status: str, error: str = '') -> str¶

query_ingestions(dataset_name: str = '', limit: int = 50) -> list[dict[str, Any]]¶

add_cognify_run(dataset_name: str, status: str, duration_sec: float, error: str = '') -> str¶

query_cognify_runs(dataset_name: str = '', limit: int = 50) -> list[dict[str, Any]]¶

add_search(query: str, search_type: str, result_count: int, duration_sec: float) -> str¶

query_searches(limit: int = 50) -> list[dict[str, Any]]¶

upsert_dataset(name: str, description: str = '', item_count: int = 0, last_cognified: str = '') -> str¶

query_datasets(limit: int = 50) -> list[dict[str, Any]]¶

get_dataset(name: str) -> dict[str, Any] | None¶

add_session_event(session_id: str, action: str, feedback_text: str = '', metadata_json: str = '{}') -> str¶

query_sessions(session_id: str = '', limit: int = 50) -> list[dict[str, Any]]¶

text_search(query: str, top_k: int = 5) -> list[dict[str, Any]]¶

count_all() -> dict[str, int]¶

Functions¶

agentic_planner_enabled(default_enabled: bool) -> bool¶

eligible_fingerprint(records: list[Any]) -> str¶

validate_kg_rerank_decision(decision: KGRerankDecision, n_eligible: int, expected_fingerprint: str) -> None¶

applied_agentic_patterns() -> list[dict[str, Any]]¶

get_skill_catalog() -> CtxCogneeSkillCatalog¶

MCP Tools¶

`KGRerankDecision`¶

`LLMKGRerankRuntime`¶

`rerank(query: str, records: list[Any]) -> KGRerankDecision`¶

`CtxCogneeRerankPatternRuntime`¶

`enforce_eligibility(decision: KGRerankDecision, records: list[Any]) -> tuple[list[Any], bool, bool]`¶

`CtxCogneePlanner`¶

`rerank(query: str, records: list[Any]) -> list[Any]`¶

`CtxCogneeBlock(AIBlock[CogneeInput, CogneeOutput, dict])`¶

`infer(data: CogneeInput) -> Result[CogneeOutput]`¶

`EntityRelation(BaseModel)`¶

`CogneeInput(BaseModel)`¶

`CogneeOutput(BaseModel)`¶

`CtxCogneeMCPBlock(AIBlock[MCPCogneeInput, MCPCogneeOutput, dict])`¶

`infer(data: MCPCogneeInput) -> Result[MCPCogneeOutput]`¶

`MCPCogneeRecord(BaseModel)`¶

`MCPCogneeInput(BaseModel)`¶

`MCPCogneeOutput(BaseModel)`¶

`CogneeStore`¶

`add_ingestion(dataset_name: str, text_preview: str, char_count: int, cognee_status: str, error: str = '') -> str`¶

`query_ingestions(dataset_name: str = '', limit: int = 50) -> list[dict[str, Any]]`¶

`add_cognify_run(dataset_name: str, status: str, duration_sec: float, error: str = '') -> str`¶

`query_cognify_runs(dataset_name: str = '', limit: int = 50) -> list[dict[str, Any]]`¶

`add_search(query: str, search_type: str, result_count: int, duration_sec: float) -> str`¶

`query_searches(limit: int = 50) -> list[dict[str, Any]]`¶

`upsert_dataset(name: str, description: str = '', item_count: int = 0, last_cognified: str = '') -> str`¶

`query_datasets(limit: int = 50) -> list[dict[str, Any]]`¶

`get_dataset(name: str) -> dict[str, Any] | None`¶

`add_session_event(session_id: str, action: str, feedback_text: str = '', metadata_json: str = '{}') -> str`¶

`query_sessions(session_id: str = '', limit: int = 50) -> list[dict[str, Any]]`¶

`text_search(query: str, top_k: int = 5) -> list[dict[str, Any]]`¶

`count_all() -> dict[str, int]`¶

`agentic_planner_enabled(default_enabled: bool) -> bool`¶

`eligible_fingerprint(records: list[Any]) -> str`¶

`validate_kg_rerank_decision(decision: KGRerankDecision, n_eligible: int, expected_fingerprint: str) -> None`¶

`applied_agentic_patterns() -> list[dict[str, Any]]`¶

`get_skill_catalog() -> CtxCogneeSkillCatalog`¶