Graphiti MCP

#!/usr/bin/env python3 """ Web UI Server for Graphiti MCP A FastAPI-based web interface to demonstrate and interact with the Graphiti MCP Server. """ import os import json from typing import Dict, Any, List, Optional from fastapi import FastAPI, HTTPException from fastapi.responses import FileResponse from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel from dotenv import load_dotenv # Load environment variables load_dotenv() # Import GraphitiMCP class from graphiti_mcp_server import GraphitiMCP app = FastAPI(title="Graphiti MCP Web UI", version="1.0.0") # Enable CORS for frontend app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Initialize GraphitiMCP instance graphiti_server: Optional[GraphitiMCP] = None def get_server() -> GraphitiMCP: """Get or initialize the GraphitiMCP server instance.""" global graphiti_server if graphiti_server is None: neo4j_uri = os.getenv("NEO4J_URI") neo4j_user = os.getenv("NEO4J_USER", "neo4j") neo4j_password = os.getenv("NEO4J_PASSWORD") openrouter_api_key = os.getenv("OPENROUTER_API_KEY") or os.getenv("OPENAI_API_KEY") model_name = os.getenv("MODEL_NAME", "openai/gpt-4o-mini") use_openrouter = bool(os.getenv("OPENROUTER_API_KEY")) if not neo4j_uri or not neo4j_password or not openrouter_api_key: raise HTTPException( status_code=500, detail="Missing required environment variables. Please set NEO4J_URI, NEO4J_PASSWORD, and OPENROUTER_API_KEY (or OPENAI_API_KEY)" ) graphiti_server = GraphitiMCP( neo4j_uri=neo4j_uri, neo4j_user=neo4j_user, neo4j_password=neo4j_password, openrouter_api_key=openrouter_api_key, model_name=model_name, use_openrouter=use_openrouter ) return graphiti_server # Pydantic models for request/response class StoreMemoryRequest(BaseModel): content: str metadata: Optional[Dict[str, Any]] = {} tags: Optional[List[str]] = [] class RetrieveMemoriesRequest(BaseModel): query: str limit: Optional[int] = 10 class CreateRelationshipRequest(BaseModel): source_id: str target_id: str relationship_type: str properties: Optional[Dict[str, Any]] = {} class GetContextRequest(BaseModel): query: str max_memories: Optional[int] = 20 class SearchGraphRequest(BaseModel): cypher_query: str parameters: Optional[Dict[str, Any]] = {} # API Endpoints @app.get("/") async def root(): """Serve the main HTML page.""" return FileResponse("web_ui.html") @app.get("/api/health") async def health_check(): """Health check endpoint.""" try: server = get_server() return {"status": "ok", "message": "Graphiti MCP Server is running"} except Exception as e: return {"status": "error", "message": str(e)} @app.post("/api/store_memory") async def store_memory(request: StoreMemoryRequest): """Store a memory in the graph database.""" try: server = get_server() result = await server._store_memory({ "content": request.content, "metadata": request.metadata, "tags": request.tags }) return result except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.post("/api/retrieve_memories") async def retrieve_memories(request: RetrieveMemoriesRequest): """Retrieve relevant memories from the graph database.""" try: server = get_server() result = await server._retrieve_memories({ "query": request.query, "limit": request.limit }) return result except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.post("/api/create_relationship") async def create_relationship(request: CreateRelationshipRequest): """Create a relationship between two memories.""" try: server = get_server() result = await server._create_relationship({ "source_id": request.source_id, "target_id": request.target_id, "relationship_type": request.relationship_type, "properties": request.properties }) return result except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.post("/api/get_context") async def get_context(request: GetContextRequest): """Get contextual information by retrieving and synthesizing memories.""" try: server = get_server() result = await server._get_context({ "query": request.query, "max_memories": request.max_memories }) return result except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.post("/api/search_graph") async def search_graph(request: SearchGraphRequest): """Search the graph database using Cypher query.""" try: server = get_server() result = await server._search_graph({ "cypher_query": request.cypher_query, "parameters": request.parameters }) return result except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.get("/api/list_all_memories") async def list_all_memories(limit: int = 50): """List all memories in the database.""" try: server = get_server() result = await server._search_graph({ "cypher_query": "MATCH (m:Memory) RETURN m.id as id, m.content as content, m.tags as tags, m.created_at as created_at ORDER BY m.created_at DESC LIMIT $limit", "parameters": {"limit": limit} }) return result except Exception as e: raise HTTPException(status_code=500, detail=str(e)) if __name__ == "__main__": import uvicorn import sys import socket # Allow port to be specified via command line or environment variable port = int(os.getenv("WEB_UI_PORT", "8081")) if len(sys.argv) > 1: try: port = int(sys.argv[1]) except ValueError: print(f"Invalid port: {sys.argv[1]}. Using default port 8081.") # Check if port is available def is_port_available(port): with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: try: s.bind(('0.0.0.0', port)) return True except OSError: return False # Try to find an available port if the requested one is in use original_port = port max_attempts = 10 for attempt in range(max_attempts): if is_port_available(port): break if attempt == 0: print(f"Port {port} is already in use. Trying alternative ports...") port += 1 if not is_port_available(port): print(f"Error: Could not find an available port after {max_attempts} attempts.") print(f"Please free up a port or specify a different one: python web_ui_server.py <port>") sys.exit(1) if port != original_port: print(f"Note: Port {original_port} was in use. Using port {port} instead.") print("=" * 60) print(f"Graphiti MCP Web UI is starting...") print("=" * 60) print(f"✓ Server URL: http://localhost:{port}") print(f"✓ Open your browser and navigate to: http://localhost:{port}") print() print("⚠️ IMPORTANT: Use 'localhost' or '127.0.0.1', NOT '0.0.0.0'") print("=" * 60) print("Press Ctrl+C to stop the server") print() uvicorn.run(app, host="0.0.0.0", port=port)