图数据库内存服务

# Clone the repository
git clone https://github.com/jkear/kuzu-memory-graph-mcp.git
cd kuzu-memory-graph-mcp

# Install dependencies
uv sync

# Activate the virtual environment
source .venv/bin/activate  # On Windows: .venv\Scripts\activate

# Clone the repository
git clone https://github.com/jkear/kuzu-memory-graph-mcp.git
cd kuzu-memory-graph-mcp

# Create virtual environment
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

# Install dependencies
pip install -e .

# Run using uv in development mode (recommended)
uv run kuzu-memory-server

# Or run directly with Python after activating the venv
python -m kuzu_memory_server

# Or use uvx for testing (installs from PyPI - for published package only)
# uvx kuzu-memory-server

{
  "mcpServers": {
    "kuzu-memory": {
      "command": "uv",
      "args": [
        "--directory",
        "/path/to/kuzu-memory-graph-mcp",
        "run",
        "kuzu-memory-server"
      ],
      "env": {
        "KUZU_MEMORY_DB_PATH": "/path/to/DBMS/memory.kuzu",
        "KUZU_DATABASES_DIR": "/path/to/DBMS",
        "KUZU_WRITABLE_DATABASES": "memory,prompt_engineering,research_papers"
      }
    }
  }
}

{
  "mcpServers": {
    "kuzu-memory": {
      "command": "uvx",
      "args": ["kuzu-memory-server"],
      "env": {
        "KUZU_MEMORY_DB_PATH": "/path/to/DBMS/memory.kuzu",
        "KUZU_DATABASES_DIR": "/path/to/DBMS",
        "KUZU_WRITABLE_DATABASES": "memory,prompt_engineering,research_papers"
      }
    }
  }
}

/path/to/databases/
├── memory.kuzu           # Writable database
├── prompt_engineer.kuzu  # Writable database
└── research_papers.kuzu  # Writable database

export KUZU_WRITABLE_DATABASES="memory,prompt_engineer,research_papers"
export KUZU_DATABASES_DIR="/path/to/databases"

# Switch to prompt_engineer database
switch_primary_database(database="prompt_engineer")

# Now create entities in prompt_engineer
create_entity(
    database="prompt_engineer",
    name="Chain of Thought",
    entity_type="prompt_pattern",
    observations=["Improves reasoning", "Step-by-step thinking"]
)

# List all available databases
resource_result = await access_resource("kuzu://databases/list")
# Returns JSON with database metadata including writable status

# Create a person entity in the 'memory' database
create_entity(
    database="memory",
    name="Jordan Kearfott",
    entity_type="person",
    observations=["Vibe engineer for Sales & Marketing", "Studies LLM Memory techniques", "Lives in Gainesville", "Mid at Python"]
)

# Switch to prompt_engineer and create an entity
switch_primary_database(database="prompt_engineer")
create_entity(
    database="prompt_engineer",
    name="Chain of Thought",
    entity_type="prompt_pattern",
    observations=["Improves reasoning", "Step-by-step thinking"]
)

# Switch to research_papers and create an entity
switch_primary_database(database="research_papers")
create_entity(
    database="research_papers",
    name="Attention Is All You Need",
    entity_type="paper",
    observations=["Transformer architecture", "Self-attention mechanism"]
)

# Create relationship between entities in specific database
create_relationship(
    database="memory",
    from_entity="Sam Altman",
    to_entity="OpenAI",
    relationship_type="IS_CEO",
    confidence=0.9
)

# Search in specific database
search_entities(database="memory", query="software engineer", limit=10)

# Semantic search across research papers
semantic_search(database="research_papers", query="transformer architecture", limit=10, threshold=0.3)

# Get related entities in prompt database
get_related_entities(database="prompt_engineer", entity_name="Chain of Thought", max_depth=2)

# Get overview of all databases
summary = await get_graph_summary()  # No database param = all databases

# Get detailed summary of specific database
summary = await get_graph_summary(database="memory")

# Run basic functionality test
uv python test_server.py

# Run architecture verification test
uv python test_architecture.py

# Run multi-primary database tests
uv python test_multi_primary.py

# Run with pytest (if installed)
pytest tests/

@asynccontextmanager
async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]:
    """Manage application lifecycle with database and model initialization."""
    # Parse writable databases configuration
    writable_dbs_str = os.getenv('KUZU_WRITABLE_DATABASES', '')
    writable_databases = [db.strip() for db in writable_dbs_str.split(',') if db.strip()]
    
    # Get databases directory
    databases_dir = os.getenv('KUZU_DATABASES_DIR', './DBMS')
    
    # Load embedding models
    embedding_model = load_embeddings()
    tokenizer = load_tokenizer()
    
    # Initialize database manager
    db_manager.initialize(writable_databases, databases_dir, embedding_model, tokenizer)
    
    # Switch to initial primary database
    initial_db = writable_databases[0] if writable_databases else get_primary_db_name(db_path)
    success, msg = db_manager.switch_to(initial_db)
    
    try:
        yield AppContext(
            db_manager=db_manager,
            embedding_model=embedding_model,
            tokenizer=tokenizer,
            databases_dir=databases_dir
        )
    finally:
        # Cleanup all connections
        db_manager.cleanup()

def generate_embedding(model: Any, tokenizer: Any, text: str) -> list[float]:
    """Generate 384-dimensional embedding using MLX or fallback."""
    if not text or not text.strip():
        return [0.0] * 384
    
    # Try MLX first
    try:
        import mlx.core as mx
        inputs = tokenizer.encode(text.strip(), return_tensors="mlx")
        outputs = model(inputs)
        return outputs.text_embeds.tolist()
    except:
        # Fallback to sentence transformers
        embedding = embedding_model.encode(text.strip(), convert_to_numpy=True)
        return embedding.tolist()

# Entity node table
conn.execute("""
    CREATE NODE TABLE IF NOT EXISTS Entity (
        name STRING PRIMARY KEY,
        type STRING,
        observations STRING[],
        embedding FLOAT[384],
        created_date DATE DEFAULT current_date(),
        updated_date DATE DEFAULT current_date()
    )
""")

# Relationship table
conn.execute("""
    CREATE REL TABLE IF NOT EXISTS RELATED_TO (
        FROM Entity TO Entity,
        relationship_type STRING,
        confidence FLOAT DEFAULT 1.0,
        created_date DATE DEFAULT current_date()
    )
""")

import pytest
from unittest.mock import Mock, AsyncMock
from src.kuzu_memory_server import create_entity

@pytest.mark.asyncio
async def test_create_entity():
    # Setup mock context
    mock_ctx = Mock()
    mock_ctx.request_context.lifespan_context = Mock()
    mock_ctx.request_context.lifespan_context.conn = AsyncMock()
    
    # Mock database response
    mock_ctx.request_context.lifespan_context.conn.execute.return_value = Mock()
    mock_ctx.request_context.lifespan_context.conn.execute.return_value.has_next.return_value = False
    
    # Test entity creation
    result = await create_entity(
        mock_ctx,
        name="Test Entity",
        entity_type="test",
        observations=["Test observation"]
    )
    
    # Assertions
    assert result["status"] == "created"
    assert result["name"] == "Test Entity"
    assert result["type"] == "test"

# In test setup
test_db_path = "./test_memory.kuzu"
os.environ["KUZU_MEMORY_DB_PATH"] = test_db_path

# Cleanup after tests
if os.path.exists(test_db_path):
    shutil.rmtree(test_db_path)

import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# In your code
logger.info(f"Creating entity: {name}")
logger.debug(f"Generated embedding dimension: {len(embedding)}")

import cProfile
import pstats

# Profile embedding generation
profiler = cProfile.Profile()
profiler.enable()

# Your code here
embedding = generate_embedding(model, tokenizer, text)

profiler.disable()
stats = pstats.Stats(profiler)
stats.sort_stats('cumulative')
stats.print_stats(10)