為 Cognee 知識圖譜構建適用於生產環境的 FastMCP 伺服器整合

Overview

Context fragmentation is one of the most critical bottlenecks when building autonomous AI agents. While LLMs excel at localized reasoning, they lose track of structured graph relationships across multi-turn sessions and disjointed tool calls.

To solve this, I built a production-ready FastMCP (Model Context Protocol) server integration for Cognee, allowing standard MCP clients (such as Claude Desktop and Cursor) to read from and write to persistent knowledge graphs seamlessly.

Architecture & Integration Flow

The integration lives under cognee/api/mcp/ and bridges standard MCP client requests directly into Cognee's engine.

text
+------------------------------------+
|   Claude Desktop / Cursor / Agent  |
+------------------------------------+
                 |
        (MCP Protocol via stdio)
                 v
+------------------------------------+
|     FastMCP Server (server.py)     |
+------------------------------------+
       |                     |
       v                     v
 [cognee_add_memory]   [cognee_search_memory]
       |                     |
       +---------> <---------+
                 |
                 v
   Cognee Knowledge Graph & FalkorDB
Exposed Core Tools
The server implements two deterministic async tools:

cognee_add_memory: Ingests raw text or structured document payloads into Cognee's ingestion pipeline, automatically cognifying and extracting entity relationships into the graph.

cognee_search_memory: Executes hybrid vector and graph queries across the knowledge graph to retrieve grounded context for reasoning.

Implementation Snippet (server.py)
Python
from mcp.server.fastmcp import FastMCP
import cognee

mcp = FastMCP("Cognee-Context-Engine")

@mcp.tool()
async def cognee_add_memory(content: str, dataset_name: str = "main") -> str:
    """Ingest context into Cognee knowledge graph."""
    await cognee.add(content, dataset_name=dataset_name)
    await cognee.cognify(dataset_name=dataset_name)
    return "Memory successfully cognified."

@mcp.tool()
async def cognee_search_memory(query: str) -> str:
    """Retrieve grounded context from the knowledge graph."""
    results = await cognee.search(query_text=query)
    return str(results)

if __name__ == "__main__":
    mcp.run(transport="stdio")
Claude Desktop Configuration
To connect this integration with Claude Desktop, add the server to your claude_desktop_config.json:

JSON
{
  "mcpServers": {
    "cognee": {
      "command": "python",
      "args": ["-m", "cognee.api.mcp.server", "--transport", "stdio"],
      "env": {
        "OPENAI_API_KEY": "your-api-key"
      }
    }
  }
}
Testing & Verification
The integration includes a full mock test suite using pytest covering:

Correct tool schema dispatching.

Graceful handling of empty queries and malformed payloads.

Case insensitivity and dataset segregation.

Result: 10/10 unit tests passed (100% green) in 0.06 seconds.

Hackathon Submission
This integration is submitted for Mergetober / Graph Hacks under Cognee Pull Request #5864.