Synchronize Memory Across AI Tools

mcp-memory-service

Persistent Shared Memory for AI Agent Pipelines

Open-source memory backend for AI agents - REST API, MCP, OAuth, CLI, dashboard. One self-hosted service, every transport. Agents store decisions, share causal knowledge graphs, and retrieve context in 5ms - without cloud lock-in or API costs.

Works with LangGraph · CrewAI · AutoGen · any HTTP client · Claude Desktop · OpenCode

Why Agents Need This

Your AI assistant forgets everything when you start a new chat. You spend 10 minutes re-explaining your architecture. Again. MCP Memory Service captures project context, architecture decisions, and code patterns automatically - new sessions start with everything already known.

Key capabilities for agent pipelines:

• Framework-agnostic REST API - no MCP client library needed; the live endpoint list is at /api/docs
• Knowledge graph - agents share causal chains, not just facts
• X-Agent-ID header - auto-tag memories by agent identity for scoped retrieval
• conversation_id - bypass deduplication for incremental conversation storage
• SSE events - real-time notifications when any agent stores or deletes a memory
• Embeddings run locally via ONNX - memory never leaves your infrastructure

🚀 Get Started in 60 Seconds

Not sure which setup fits? The Setup Guide walks you to the right path in under a minute.

1. Install:

pip install mcp-memory-service

2. Configure your AI client:

Add to your config file:

• macOS: ~/Library/Application Support/Claude/claude_desktop_config.json
• Windows: %APPDATA%\Claude\claude_desktop_config.json
• Linux: ~/.config/Claude/claude_desktop_config.json

{
  "mcpServers": {
    "memory": {
      "command": "memory",
      "args": ["server"]
    }
  }
}

Restart Claude Desktop. Your AI now remembers everything across sessions.

claude mcp add memory -- memory server

Restart Claude Code. Memory tools will appear automatically.

MCP_ALLOW_ANONYMOUS_ACCESS=true memory server --http
# REST API running at http://localhost:8000

Store a memory with POST /api/memories, search with POST /api/search, retrieve by tag with POST /api/search/by-tag. Send X-Agent-ID: <id> on a store request and the server tags the memory agent:<id>, which a tag search then scopes retrieval by.

Worked examples per framework, the tag conventions and the async patterns: docs/agents/

MCP_ALLOW_ANONYMOUS_ACCESS=true memory server --http

The plugin ships as repository files for the local plugin directory, so clone the repository once even if you installed from PyPI. Install steps, the /memory slash command and the endpoint override: opencode/README.md

Remote MCP puts persistent memory in the browser on any device, no desktop app required: OAuth 2.0 over HTTPS, self-hosted or cloud-hosted. Both claude.ai and ChatGPT (Developer Mode) connect to the same endpoint.

Cloudflare Tunnel quick start, Let's Encrypt, nginx, Caddy and Docker production setups: Remote MCP Setup · 5-minute tutorial

git clone https://github.com/doobidoo/mcp-memory-service.git
cd mcp-memory-service
python scripts/installation/install.py

Choose from SQLite (local, fast, single-user), Cloudflare (cloud, multi-device sync), Hybrid (5ms local reads with background cloud sync - recommended for production) or Milvus (dedicated vector DB: Lite file, self-hosted, or Zilliz Cloud). For self-hosted team setups, the Hybrid backend can sync to another HTTP MCP Memory Service instead of Cloudflare using MCP_HYBRID_SECONDARY_BACKEND=http.

For long-lived services, prefer Docker Milvus or Zilliz Cloud over Milvus Lite - why.

⚡ Works With Your Favorite AI Tools

• Agent frameworks (REST): LangGraph · CrewAI · AutoGen · OpenClaw/Nanobot · any HTTP client
• CLI and terminal (MCP): Claude Code · Gemini CLI · OpenCode · Codex CLI · Goose · Aider · Amp
• Desktop and IDE (MCP): Claude Desktop · VS Code · Cursor · Windsurf · Raycast · JetBrains · Zed
• Chat (MCP): ChatGPT (Developer Mode) · claude.ai (Remote MCP over HTTPS)

Full list, plus clients without OAuth such as Home Assistant: docs/integrations.md

✨ Features

🧠 Persistent Memory - Context survives across sessions with semantic search 🔍 Smart Retrieval - Finds relevant context automatically using AI embeddings ⚡ 5ms Speed - Instant context injection, no latency ☁️ Cloud Sync - Optional Cloudflare backend for team collaboration 🔒 Privacy-First - Local-first, you control your data 📊 Web Dashboard - Visualize and manage memories at http://localhost:8000 🧬 Knowledge Graph - Interactive D3.js visualization of memory relationships 🏠 Homelab Quality Scoring - Point scoring at any OpenAI-compatible endpoint (Ollama, LiteLLM, vLLM) 🔗 Entity Extraction - Auto-links @mentions, #tags, URLs, and file paths to a queryable entity graph 💡 Insight Cards - Consolidation surfaces patterns, trends, and knowledge gaps as structured insights 🏷️ Tag Match Filtering - tag_match=AND/OR on memory_search for precise multi-tag queries

The dashboard has eight tabs - Dashboard, Search, Browse, Documents, Manage, Analytics, Quality, API Docs. Two-minute walkthrough on YouTube · Web Dashboard Guide

How it compares to Mem0, Zep and the MCP-native alternatives, benchmark results, and deployments people run in production: mcpmemory.services

🛠️ Configuration Highlights

memory launch                  # Start HTTP server in background (127.0.0.1:8000)
memory launch --port 8192      # Custom port
memory info                    # Status and health
memory logs --lines 50         # Recent logs
memory stop                    # Stop server

These commands are optimized for fast startup and avoid loading heavy ML dependencies unless needed.

⚠️ Security note: the server binds to 127.0.0.1 (localhost only) by default. --host 0.0.0.0 / MCP_HTTP_HOST=0.0.0.0 exposes the API to your network - do that only in trusted environments with authentication and firewall rules, or behind TLS termination or a VPN overlay.

Backends, embedding models, quality scoring and every environment variable: Configuration Guide

📚 Documentation

• Setup Guide - Decision tree and step-by-step paths
• Agent Integration Guides - LangGraph, CrewAI, AutoGen, HTTP generic
• Remote MCP Setup - claude.ai and ChatGPT via HTTPS + OAuth
• Configuration Guide - Backend options and customization
• Architecture Overview - How it works under the hood
• Knowledge Graph Dashboard - Interactive graph visualization
• Benchmarks - LongMemEval, DevBench, LoCoMo, with run commands
• Migration Guide - Upgrading between major versions
• Troubleshooting - Common issues and solutions
• Wiki - Long-form guides and API reference
• Full documentation index - Everything else

Also listed on Glama and Spark.

📦 Releases

Every release, with upgrade notes: CHANGELOG.md · GitHub Releases · archived history

🤝 Contributing

We welcome contributions! See CONTRIBUTING.md for guidelines and SECURITY.md for reporting a vulnerability.

Who authors this project, who holds copyright, and what every change passes before it reaches main: AUTHORSHIP.md.

Quick Development Setup:

git clone https://github.com/doobidoo/mcp-memory-service.git
cd mcp-memory-service
pip install -e .  # Editable install
pytest tests/      # Run test suite

Supporting the Project

MCP Memory Service is maintained by one person. If it saves you or your company time, you can support its development via Ko-fi, Buy Me a Coffee or PayPal: SPONSORS.md.

scripts/installation/install.py

#!/usr/bin/env python3
"""
Installation script for MCP Memory Service with cross-platform compatibility.
This script guides users through the installation process with the appropriate
dependencies for their platform.
"""
import os
import sys
import platform
import subprocess
import argparse
import shutil
from pathlib import Path
from typing import Tuple, Dict, Any, Optional
import importlib.util
import json
import re

def is_python_version_at_least(major, minor):
    """Check if current Python version is at least the specified version.

    Args:
        major: Major version number
        minor: Minor version number

    Returns:
        bool: True if current Python version >= specified version
    """
    return sys.version_info >= (major, minor)

def get_python_version_string():
    """Get Python version as a string (e.g., '3.12').

    Returns:
        str: Python version string
    """
    return f"{sys.version_info.major}.{sys.version_info.minor}"

def get_package_version():
    """Get the current package version from pyproject.toml.

    Returns:
        str: The version string from pyproject.toml or fallback version
    """
    fallback_version = "7.2.0"

    try:
        # Get path to pyproject.toml relative to this script
        pyproject_path = Path(__file__).parent.parent.parent / "pyproject.toml"

        if not pyproject_path.exists():
            print_warning(f"pyproject.toml not found at {pyproject_path}, using fallback version {fallback_version}")
            return fallback_version

        with open(pyproject_path, "r", encoding="utf-8") as f:
            content = f.read()

        # Extract version using regex - matches standard pyproject.toml format
        version_pattern = r'^version\s*=\s*["\']([^"\'\n]+)["\']'
        version_match = re.search(version_pattern, content, re.MULTILINE)

        if version_match:
            version = version_match.group(1).strip()
            if version:  # Ensure non-empty version
                return version
            else:
                print_warning("Empty version found in pyproject.toml, using fallback")
                return fallback_version
        else:
            print_warning("Version not found in pyproject.toml, using fallback")
            return fallback_version

    except (OSError, IOError) as e:
        print_warning(f"Failed to read pyproject.toml: {e}, using fallback version {fallback_version}")
        return fallback_version
    except Exception as e:
        print_warning(f"Unexpected error parsing version: {e}, using fallback version {fallback_version}")
        return fallback_version

def print_header(text):
    """Print a formatted header."""
    print("\n" + "=" * 80)
    print(f" {text}")
    print("=" * 80)

def print_step(step, text):
    """Print a formatted step."""
    print(f"\n[{step}] {text}")

def print_info(text):
    """Print formatted info text."""
    print(f"  → {text}")

def print_error(text):
    """Print formatted error text."""
    print(f"  ❌ ERROR: {text}")

def print_success(text):
    """Print formatted success text."""
    print(f"  ✅ {text}")

def print_warning(text):
    """Print formatted warning text."""
    print(f"  ⚠️  {text}")

def run_command_safe(cmd, success_msg=None, error_msg=None, silent=False,
                     timeout=None, fallback_in_venv=False):
    """
    Run a subprocess command with standardized error handling.

    Args:
        cmd: Command to run (list of strings)
        success_msg: Message to print on success
        error_msg: Custom error message
        silent: If True, suppress stdout/stderr
        timeout: Command timeout in seconds
        fallback_in_venv: If True and command fails, warn instead of error when in virtual environment

    Returns:
        tuple: (success: bool, result: subprocess.CompletedProcess or None)
    """
    # Validate command input
    if not cmd or not isinstance(cmd, (list, tuple)):
        print_error("Invalid command: must be a non-empty list or tuple")
        return False, None

    if not all(isinstance(arg, (str, int, float)) for arg in cmd if arg is not None):
        print_error("Invalid command arguments: all arguments must be strings, numbers, or None")
        return False, None

    # Filter out None values and convert to strings
    cmd_clean = [str(arg) for arg in cmd if arg is not None]
    if not cmd_clean:
        print_error("Command is empty after filtering")
        return False, None

    try:
        kwargs = {'capture_output': False, 'text': True}
        if silent:
            kwargs.update({'stdout': subprocess.DEVNULL, 'stderr': subprocess.DEVNULL})
        if timeout:
            kwargs['timeout'] = timeout

        result = subprocess.run(cmd_clean, check=True, **kwargs)
        if success_msg:
            print_success(success_msg)
        return True, result
    except subprocess.TimeoutExpired as e:
        if error_msg:
            timeout_msg = error_msg
        elif hasattr(e, 'timeout') and e.timeout:
            timeout_msg = f"Command timed out after {e.timeout}s"
        else:
            timeout_msg = "Command timed out"
        print_error(timeout_msg)
        return False, None
    except subprocess.CalledProcessError as e:
        if fallback_in_venv:
            in_venv = sys.prefix != sys.base_prefix
            if in_venv:
                fallback_msg = error_msg or "Command failed, but you're in a virtual environment. If you're using an alternative package manager, this may be normal."
                print_warning(fallback_msg)
                print_warning("Note: Installation may not have succeeded. Please verify manually if needed.")
                return True, None  # Proceed anyway in venv with warning

        if error_msg:
            print_error(error_msg)
        else:
            # Safe command formatting for error messages
            cmd_str = ' '.join(f'"{arg}"' if ' ' in str(arg) else str(arg) for arg in cmd_clean)
            print_error(f"Command failed (exit code {e.returncode}): {cmd_str}")
        return False, None
    except FileNotFoundError:
        if error_msg:
            print_error(error_msg)
        else:
            print_error(f"Command not found: {cmd_clean[0]}")
        return False, None
    except PermissionError:
        permission_msg = error_msg or f"Permission denied executing: {cmd_clean[0]}"
        print_error(permission_msg)
        return False, None

def install_package_safe(package, success_msg=None, error_msg=None, fallback_in_venv=True):
    """
    Install a Python package with standardized error handling.

    Args:
        package: Package name or requirement string
        success_msg: Message to print on success
        error_msg: Custom error message
        fallback_in_venv: If True, warn instead of error when in virtual environment

    Returns:
        bool: True if installation succeeded OR if fallback was applied (see warning messages)
    """
    pip_module_available = importlib.util.find_spec("pip") is not None
    uv_path = shutil.which("uv")
    if pip_module_available:
        cmd = [sys.executable, '-m', 'pip', 'install', package]
    elif uv_path:
        # uv environments commonly omit pip; use uv to install into the current interpreter
        cmd = [uv_path, 'pip', 'install', '--python', sys.executable, package]
    else:
        print_error("Neither pip nor uv detected. Cannot install packages.")
        return False
    default_success = success_msg or f"{package} installed successfully"
    default_error = error_msg or f"Failed to install {package}"

    if fallback_in_venv:
        default_error += ". If you're using an alternative package manager like uv, please install manually."

    success, _ = run_command_safe(
        cmd,
        success_msg=default_success,
        error_msg=default_error,
        silent=True,
        fallback_in_venv=fallback_in_venv
    )
    return success

def detect_system():
    """Detect the system architecture and platform."""
    system = platform.system().lower()
    machine = platform.machine().lower()
    python_version = f"{sys.version_info.major}.{sys.version_info.minor}"
    
    is_windows = system == "windows"
    is_macos = system == "darwin"
    is_linux = system == "linux"
    is_arm = machine in ("arm64", "aarch64")
    is_x86 = machine in ("x86_64", "amd64", "x64")
    
    print_info(f"System: {platform.system()} {platform.release()}")
    print_info(f"Architecture: {machine}")
    print_info(f"Python: {python_version}")
    
    # Check for virtual environment
    in_venv = sys.prefix != sys.base_prefix
    if not in_venv:
        print_warning("Not running in a virtual environment. It's recommended to install in a virtual environment.")
    else:
        print_info(f"Virtual environment: {sys.prefix}")
    
    # Check for Homebrew PyTorch installation
    has_homebrew_pytorch = False
    homebrew_pytorch_version = None
    if is_macos:
        try:
            # Check if pytorch is installed via brew
            result = subprocess.run(
                ['brew', 'list', 'pytorch', '--version'],
                capture_output=True,
                text=True
            )
            if result.returncode == 0:
                has_homebrew_pytorch = True
                # Extract version from output
                version_line = result.stdout.strip()
                homebrew_pytorch_version = version_line.split()[1] if len(version_line.split()) > 1 else "Unknown"
                print_info(f"Detected Homebrew PyTorch installation: {homebrew_pytorch_version}")
        except (subprocess.SubprocessError, FileNotFoundError):
            pass
    
    return {
        "system": system,
        "machine": machine,
        "python_version": python_version,
        "is_windows": is_windows,
        "is_macos": is_macos,
        "is_linux": is_linux,
        "is_arm": is_arm,
        "is_x86": is_x86,
        "in_venv": in_venv,
        "has_homebrew_pytorch": has_homebrew_pytorch,
        "homebrew_pytorch_version": homebrew_pytorch_version
    }

def detect_gpu():
    """Detect GPU and acceleration capabilities.

    Wrapper function that uses the shared GPU detection module.
    """
    # Import the module directly by file path to avoid triggering package __init__.py
    # which has dependencies that aren't installed yet
    import importlib.util
    
    gpu_detection_path = Path(__file__).parent.parent.parent / "src" / "mcp_memory_service" / "utils" / "gpu_detection.py"
    
    try:
        # First try if package is already installed
        from mcp_memory_service.utils.gpu_detection import detect_gpu as shared_detect_gpu
    except ImportError:
        # Load module directly from file path without triggering package init
        spec = importlib.util.spec_from_file_location("gpu_detection", gpu_detection_path)
        if spec is None or spec.loader is None:
            raise ImportError(f"Could not load GPU detection module from {gpu_detection_path}")
        gpu_module = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(gpu_module)
        shared_detect_gpu = gpu_module.detect_gpu

    system_info = detect_system()

    # Use shared GPU detection module
    gpu_info = shared_detect_gpu(system_info)

    # Print GPU information (maintain installer output format)
    if gpu_info.get("has_cuda"):
        cuda_version = gpu_info.get("cuda_version")
        print_info(f"CUDA detected: {cuda_version or 'Unknown version'}")
    if gpu_info.get("has_rocm"):
        rocm_version = gpu_info.get("rocm_version")
        print_info(f"ROCm detected: {rocm_version or 'Unknown version'}")
    if gpu_info.get("has_mps"):
        print_info("Apple Metal Performance Shaders (MPS) detected")
    if gpu_info.get("has_directml"):
        directml_version = gpu_info.get("directml_version")
        if directml_version:
            print_info(f"DirectML detected: {directml_version}")
        else:
            print_info("DirectML detected")

    if not (gpu_info.get("has_cuda") or gpu_info.get("has_rocm") or
            gpu_info.get("has_mps") or gpu_info.get("has_directml")):
        print_info("No GPU acceleration detected, will use CPU-only mode")

    return gpu_info

def check_dependencies():
    """Check for required dependencies.
    
    Note on package managers:
    - Traditional virtual environments (venv, virtualenv) include pip by default
    - Alternative package managers like uv may not include pip or may manage packages differently
    - We attempt multiple detection methods for pip and only fail if:
      a) We're not in a virtual environment, or
      b) We can't detect pip AND can't install dependencies
    
    We proceed with installation even if pip isn't detected when in a virtual environment,
    assuming an alternative package manager (like uv) is handling dependencies.
    
    Returns:
        bool: True if all dependencies are met, False otherwise.
    """
    print_step("2", "Checking dependencies")
    
    # Check for pip
    pip_installed = False
    
    # Try subprocess check first
    success, _ = run_command_safe(
        [sys.executable, '-m', 'pip', '--version'],
        success_msg="pip is installed",
        silent=True,
        fallback_in_venv=True
    )

    if success:
        pip_installed = True
    else:
        # Fallback to import check
        try:
            import pip
            pip_installed = True
            print_info(f"pip is installed: {pip.__version__}")
        except ImportError:
            # Check if we're in a virtual environment
            in_venv = sys.prefix != sys.base_prefix
            if in_venv:
                print_warning("pip could not be detected, but you're in a virtual environment. "
                            "If you're using uv or another alternative package manager, this is normal. "
                            "Continuing installation (will use uv where needed)...")
                pip_installed = True  # Proceed anyway
            else:
                print_error("pip is not installed. Please install pip first.")
                return False
    
    # Check for setuptools
    try:
        import setuptools
        print_info(f"setuptools is installed: {setuptools.__version__}")
    except ImportError:
        print_warning("setuptools is not installed. Will attempt to install it.")
        # If pip is available, use it to install setuptools
        if pip_installed:
            success = install_package_safe("setuptools")
            if not success:
                return False
        else:
            # Should be unreachable since pip_installed would only be False if we returned earlier
            print_error("Cannot install setuptools without pip. Please install setuptools manually.")
            return False
    
    # Check for wheel
    try:
        import wheel
        print_info(f"wheel is installed: {wheel.__version__}")
    except ImportError:
        print_warning("wheel is not installed. Will attempt to install it.")
        # If pip is available, use it to install wheel
        if pip_installed:
            success = install_package_safe("wheel")
            if not success:
                return False
        else:
            # Should be unreachable since pip_installed would only be False if we returned earlier
            print_error("Cannot install wheel without pip. Please install wheel manually.")
            return False
    
    return True

def install_pytorch_platform_specific(system_info, gpu_info):
    """Install PyTorch with platform-specific configurations."""
    if system_info["is_windows"]:
        return install_pytorch_windows(gpu_info)
    elif system_info["is_macos"] and system_info["is_x86"]:
        return install_pytorch_macos_intel()
    else:
        # For other platforms, let the regular installer handle it
        return True

def install_pytorch_macos_intel():
    """Install PyTorch specifically for macOS with Intel CPUs."""
    print_step("3a", "Installing PyTorch for macOS Intel CPU")
    
    # Use the versions known to work well on macOS Intel and with Python 3.13+
    try:
        # For Python 3.13+, we need newer PyTorch versions
        python_version = sys.version_info
        
        if python_version >= (3, 13):
            # For Python 3.13+, try to install latest compatible version
            print_info(f"Installing PyTorch for macOS Intel (Python {python_version.major}.{python_version.minor})...")
            print_info("Attempting to install latest PyTorch compatible with Python 3.13...")
            
            # Try to install without version specifiers to get latest compatible version
            cmd = [
                sys.executable, '-m', 'pip', 'install',
                "torch", "torchvision", "torchaudio"
            ]
            success, _ = run_command_safe(
                cmd,
                success_msg="Latest PyTorch installed successfully",
                silent=False
            )

            if success:
                st_version = "3.0.0"  # Newer sentence-transformers for newer PyTorch
            else:
                print_warning("Failed to install latest PyTorch, trying fallback version...")
                # Fallback to a specific version
                torch_version = "2.1.0"
                torch_vision_version = "0.16.0"
                torch_audio_version = "2.1.0"
                st_version = "3.0.0"

                print_info(f"Trying fallback to PyTorch {torch_version}...")

                cmd = [
                    sys.executable, '-m', 'pip', 'install',
                    f"torch=={torch_version}",
                    f"torchvision=={torch_vision_version}",
                    f"torchaudio=={torch_audio_version}"
                ]
                success, _ = run_command_safe(
                    cmd,
                    success_msg=f"PyTorch {torch_version} installed successfully",
                    error_msg="Failed to install PyTorch fallback version",
                    silent=False
                )
                if not success:
                    return False
        else:
            # Use traditional versions for older Python
            torch_version = "1.13.1"
            torch_vision_version = "0.14.1"
            torch_audio_version = "0.13.1"
            st_version = "2.2.2"
            
            print_info(f"Installing PyTorch {torch_version} for macOS Intel (Python {python_version.major}.{python_version.minor})...")
            
            # Install PyTorch first with compatible version
            packages = [f"torch=={torch_version}", f"torchvision=={torch_vision_version}", f"torchaudio=={torch_audio_version}"]
            success, _ = run_command_safe(
                [sys.executable, '-m', 'pip', 'install'] + packages,
                success_msg=f"PyTorch {torch_version} installed successfully"
            )
            if not success:
                raise RuntimeError(f"Failed to install PyTorch {torch_version}")
        
        # Install a compatible version of sentence-transformers
        print_info(f"Installing sentence-transformers {st_version}...")
        success, _ = run_command_safe(
            [sys.executable, '-m', 'pip', 'install', f"sentence-transformers=={st_version}"],
            success_msg=f"sentence-transformers {st_version} installed successfully"
        )
        if not success:
            raise RuntimeError(f"Failed to install sentence-transformers {st_version}")
        
        print_success(f"PyTorch {torch_version} and sentence-transformers {st_version} installed successfully for macOS Intel")
        return True
    except RuntimeError as e:
        print_error(f"Failed to install PyTorch for macOS Intel: {e}")
        
        # Provide fallback instructions
        if python_version >= (3, 13):
            print_warning("You may need to manually install compatible versions for Python 3.13+ on Intel macOS:")
            print_info("pip install torch==2.3.0 torchvision==0.18.0 torchaudio==2.3.0")
            print_info("pip install sentence-transformers==3.0.0")
        else:
            print_warning("You may need to manually install compatible versions for Intel macOS:")
            print_info("pip install torch==1.13.1 torchvision==0.14.1 torchaudio==0.13.1")
            print_info("pip install sentence-transformers==2.2.2")
        
        return False

def install_pytorch_windows(gpu_info):
    """Install PyTorch on Windows using the appropriate index URL."""
    print_step("3a", "Installing PyTorch for Windows")
    
    # Determine the appropriate PyTorch index URL based on GPU
    if gpu_info["has_cuda"]:
        # Get CUDA version and determine appropriate index URL
        cuda_version = gpu_info.get("cuda_version", "")
        
        # Extract major version from CUDA version string
        cuda_major = None
        if cuda_version:
            # Try to extract the major version (e.g., "11.8" -> "11")
            try:
                cuda_major = cuda_version.split('.')[0]
            except (IndexError, AttributeError):
                pass
        
        # Default to cu118 if we couldn't determine the version or it's not a common one
        if cuda_major == "12":
            cuda_suffix = "cu121"  # CUDA 12.x
            print_info(f"Detected CUDA {cuda_version}, using cu121 channel")
        elif cuda_major == "11":
            cuda_suffix = "cu118"  # CUDA 11.x
            print_info(f"Detected CUDA {cuda_version}, using cu118 channel")
        elif cuda_major == "10":
            cuda_suffix = "cu102"  # CUDA 10.x
            print_info(f"Detected CUDA {cuda_version}, using cu102 channel")
        else:
            # Default to cu118 as a safe choice for newer NVIDIA GPUs
            cuda_suffix = "cu118"
            print_info(f"Using default cu118 channel for CUDA {cuda_version}")
            
        index_url = f"https://download.pytorch.org/whl/{cuda_suffix}"
    else:
        # CPU-only version
        index_url = "https://download.pytorch.org/whl/cpu"
        print_info("Using CPU-only PyTorch for Windows")
    
    # Install PyTorch with the appropriate index URL
    try:
        # Use a stable version that's known to have Windows wheels
        torch_version = "2.1.0"  # This version has Windows wheels available
        
        cmd = [
            sys.executable, '-m', 'pip', 'install',
            f"torch=={torch_version}",
            f"torchvision=={torch_version}",
            f"torchaudio=={torch_version}",
            f"--index-url={index_url}"
        ]
        
        success, _ = run_command_safe(
            cmd,
            success_msg="PyTorch installed successfully for Windows",
            error_msg="Failed to install PyTorch for Windows",
            silent=False
        )
        if not success:
            return False
        
        # Check if DirectML is needed
        if gpu_info["has_directml"]:
            success = install_package_safe(
                "torch-directml>=0.2.0",
                success_msg="torch-directml installed successfully for DirectML support",
                error_msg="Failed to install torch-directml"
            )
            if not success:
                print_warning("DirectML support may not be available")
            
        print_success("PyTorch installed successfully for Windows")
        return True
    except RuntimeError as e:
        print_error(f"Failed to install PyTorch for Windows: {e}")
        print_warning("You may need to manually install PyTorch using instructions from https://pytorch.org/get-started/locally/")
        return False

def detect_storage_backend_compatibility(system_info, gpu_info):
    """Detect which storage backends are compatible with the current environment."""
    print_step("3a", "Analyzing storage backend compatibility")
    
    compatibility = {
        "cloudflare": {"supported": True, "issues": [], "recommendation": "production"},
        "sqlite_vec": {"supported": True, "issues": [], "recommendation": "development"},
        "chromadb": {"supported": True, "issues": [], "recommendation": "team"},
        "hybrid": {"supported": True, "issues": [], "recommendation": "recommended"}
    }
    
    # Check ChromaDB compatibility issues
    chromadb_issues = []
    
    # macOS Intel compatibility issues
    if system_info["is_macos"] and system_info["is_x86"]:
        chromadb_issues.append("ChromaDB has known installation issues on older macOS Intel systems")
        chromadb_issues.append("May require specific dependency versions")
        compatibility["chromadb"]["recommendation"] = "problematic"
        compatibility["sqlite_vec"]["recommendation"] = "recommended"
    
    # Memory constraints
    total_memory_gb = 0
    try:
        import psutil
        total_memory_gb = psutil.virtual_memory().total / (1024**3)
    except ImportError:
        # Fallback memory detection
        try:
            with open('/proc/meminfo', 'r') as f:
                for line in f:
                    if line.startswith('MemTotal:'):
                        total_memory_gb = int(line.split()[1]) / (1024**2)
                        break
        except (FileNotFoundError, IOError):
            pass
    
    if total_memory_gb > 0 and total_memory_gb < 4:
        chromadb_issues.append(f"System has {total_memory_gb:.1f}GB RAM - ChromaDB may consume significant memory")
        compatibility["sqlite_vec"]["recommendation"] = "recommended"
    
    # Older Python versions
    python_version = f"{sys.version_info.major}.{sys.version_info.minor}"
    if sys.version_info < (3, 9):
        chromadb_issues.append(f"Python {python_version} may have ChromaDB compatibility issues")
    
    # ARM architecture considerations
    if system_info["is_arm"]:
        print_info("ARM architecture detected - both backends should work well")
    
    compatibility["chromadb"]["issues"] = chromadb_issues
    
    # Print compatibility analysis
    print_info("Storage Backend Compatibility Analysis:")
    
    for backend, info in compatibility.items():
        status = "✅" if info["supported"] else "❌"
        rec_text = {
            "production": "☁️ PRODUCTION (Cloud)",
            "development": "🪶 DEVELOPMENT (Local)",
            "team": "👥 TEAM (Multi-client)",
            "recommended": "🌟 RECOMMENDED",
            "default": "📦 Standard",
            "problematic": "⚠️  May have issues",
            "lightweight": "🪶 Lightweight"
        }.get(info["recommendation"], "")
        
        print_info(f"  {status} {backend.upper()}: {rec_text}")
        
        if info["issues"]:
            for issue in info["issues"]:
                print_info(f"    • {issue}")
    
    return compatibility

def choose_storage_backend(system_info, gpu_info, args):
    """Choose storage backend based on environment and user preferences."""
    compatibility = detect_storage_backend_compatibility(system_info, gpu_info)
    
    # Check if user specified a backend via environment
    env_backend = os.environ.get('MCP_MEMORY_STORAGE_BACKEND')
    if env_backend:
        print_info(f"Using storage backend from environment: {env_backend}")
        return env_backend
    
    # Check for command line argument (we'll add this)
    if hasattr(args, 'storage_backend') and args.storage_backend:
        print_info(f"Using storage backend from command line: {args.storage_backend}")
        return args.storage_backend
    
    # Auto-select based on compatibility
    recommended_backend = None
    for backend, info in compatibility.items():
        if info["recommendation"] == "recommended":
            recommended_backend = backend
            break
    
    if not recommended_backend:
        recommended_backend = "sqlite_vec"  # Default fallback for local development

    # Interactive backend selection
    print_step("3b", "Storage Backend Selection")
    print_info("Choose the storage backend that best fits your use case:")
    print_info("")
    print_info("Usage scenarios:")
    print_info("  1. Production/Shared (Cloudflare) - Cloud storage, multi-user access, requires credentials")
    print_info("  2. Development/Personal (SQLite-vec) - Local, lightweight, single-user")
    print_info("  3. Team/Multi-client (ChromaDB) - Local server, multiple clients")
    print_info("  4. Hybrid (Recommended) - Fast local SQLite + background Cloudflare sync")
    print_info("  5. Auto-detect - Try optimal backend based on your system")
    print_info("")

    # Show compatibility analysis
    for i, (backend, info) in enumerate(compatibility.items(), 1):
        if backend == "auto_detect":
            continue
        status = "✅" if info["supported"] else "❌"
        rec_text = {
            "production": "☁️ PRODUCTION (Cloud)",
            "development": "🪶 DEVELOPMENT (Local)",
            "team": "👥 TEAM (Multi-client)",
            "recommended": "🌟 RECOMMENDED",
            "problematic": "⚠️  May have issues"
        }.get(info["recommendation"], "")
        print_info(f"  {status} {i}. {backend.upper()}: {rec_text}")
        if info["issues"]:
            for issue in info["issues"]:
                print_info(f"     • {issue}")

    print_info("")
    default_choice = "2" if compatibility["chromadb"]["recommendation"] == "problematic" else "2"

    while True:
        try:
            choice = input(f"Choose storage backend [1-5] (default: 4 - hybrid): ").strip()
            if not choice:
                choice = "4"  # Default to hybrid

            if choice == "1":
                return "cloudflare"
            elif choice == "2":
                return "sqlite_vec"
            elif choice == "3":
                return "chromadb"
            elif choice == "4":
                return "hybrid"
            elif choice == "5":
                return "auto_detect"
            else:
                print_error("Please enter 1, 2, 3, 4, or 5")
        except (EOFError, KeyboardInterrupt):
            print_info(f"\nUsing recommended backend: hybrid")
            return "hybrid"

def setup_cloudflare_credentials():
    """Interactive setup of Cloudflare credentials."""
    print_step("3c", "Cloudflare Backend Setup")
    print_info("Cloudflare backend requires API credentials for D1 database and Vectorize index.")
    print_info("You'll need:")
    print_info("  • Cloudflare API Token (with D1 and Vectorize permissions)")
    print_info("  • Account ID")
    print_info("  • D1 Database ID")
    print_info("  • Vectorize Index name")
    print_info("")
    print_info("Visit https://dash.cloudflare.com to get these credentials.")
    print_info("")

    credentials = {}

    try:
        # Get API Token
        while True:
            token = input("Enter Cloudflare API Token: ").strip()
            if token:
                credentials['CLOUDFLARE_API_TOKEN'] = token
                break
            print_error("API token is required")

        # Get Account ID
        while True:
            account_id = input("Enter Cloudflare Account ID: ").strip()
            if account_id:
                credentials['CLOUDFLARE_ACCOUNT_ID'] = account_id
                break
            print_error("Account ID is required")

        # Get D1 Database ID
        while True:
            d1_id = input("Enter D1 Database ID: ").strip()
            if d1_id:
                credentials['CLOUDFLARE_D1_DATABASE_ID'] = d1_id
                break
            print_error("D1 Database ID is required")

        # Get Vectorize Index
        vectorize_index = input("Enter Vectorize Index name (default: mcp-memory-index): ").strip()
        if not vectorize_index:
            vectorize_index = "mcp-memory-index"
        credentials['CLOUDFLARE_VECTORIZE_INDEX'] = vectorize_index

        # Set storage backend
        credentials['MCP_MEMORY_STORAGE_BACKEND'] = 'cloudflare'

        return credentials

    except (EOFError, KeyboardInterrupt):
        print_info("\nCloudflare setup cancelled.")
        return None

def save_credentials_to_env(credentials):
    """Save credentials to .env file and current environment."""
    env_file = Path('.env')

    # Read existing .env content if it exists
    existing_lines = []
    if env_file.exists():
        with open(env_file, 'r') as f:
            existing_lines = f.readlines()

    # Filter out any existing Cloudflare variables
    filtered_lines = [
        line for line in existing_lines
        if not any(key in line for key in credentials.keys())
    ]

    # Add new credentials
    with open(env_file, 'w') as f:
        # Write existing non-Cloudflare lines
        f.writelines(filtered_lines)

        # Add separator if file wasn't empty
        if filtered_lines and not filtered_lines[-1].endswith('\n'):
            f.write('\n')
        if filtered_lines:
            f.write('\n# Cloudflare Backend Configuration\n')

        # Write Cloudflare credentials
        for key, value in credentials.items():
            f.write(f'{key}={value}\n')

    # Also set credentials in current environment for immediate use
    for key, value in credentials.items():
        os.environ[key] = value

    print_success(f"Credentials saved to .env file and current environment")

def test_cloudflare_connection(credentials):
    """Test Cloudflare API connection."""
    print_info("Testing Cloudflare API connection...")

    try:
        import requests

        headers = {
            'Authorization': f"Bearer {credentials['CLOUDFLARE_API_TOKEN']}",
            'Content-Type': 'application/json'
        }

        # Test API token validity using account-specific endpoint
        account_id = credentials.get('CLOUDFLARE_ACCOUNT_ID', '').strip()
        if account_id:
            endpoint = f"https://api.cloudflare.com/client/v4/accounts/{account_id}/tokens/verify"
        else:
            # Fallback to user-level endpoint if account_id is not provided
            endpoint = "https://api.cloudflare.com/client/v4/user/tokens/verify"

        response = requests.get(
            endpoint,
            headers=headers,
            timeout=10
        )

        if response.status_code == 200:
            data = response.json()
            if data.get("success"):
                print_success("API token is valid")
                return True
            else:
                print_error(f"API token validation failed: {data.get('errors')}")
                return False
        else:
            print_error(f"API connection failed with status {response.status_code}")
            return False

    except ImportError:
        print_warning("Could not test connection (requests not installed)")
        print_info("Connection will be tested when the service starts")
        return True
    except Exception as e:
        print_warning(f"Could not test connection: {e}")
        print_info("Connection will be tested when the service starts")
        return True

def install_storage_backend(backend, system_info):
    """Install the chosen storage backend."""
    print_step("3c", f"Installing {backend} storage backend")

    if backend == "cloudflare":
        print_info("Cloudflare backend uses cloud services - no local dependencies needed")

        # Setup credentials interactively
        credentials = setup_cloudflare_credentials()
        if not credentials:
            print_warning("Cloudflare setup cancelled. Falling back to SQLite-vec.")
            return install_storage_backend("sqlite_vec", system_info)

        # Save credentials to .env file
        save_credentials_to_env(credentials)

        # Test connection
        connection_ok = test_cloudflare_connection(credentials)
        if connection_ok:
            print_success("Cloudflare backend configured successfully")
            return "cloudflare"
        else:
            print_warning("Cloudflare connection test failed. You can continue and fix credentials later.")
            fallback = input("Continue with Cloudflare anyway? [y/N]: ").strip().lower()
            if fallback.startswith('y'):
                return "cloudflare"
            else:
                print_info("Falling back to SQLite-vec for local development.")
                return install_storage_backend("sqlite_vec", system_info)

    elif backend == "sqlite_vec":
        return install_package_safe(
            "sqlite-vec",
            success_msg="SQLite-vec installed successfully",
            error_msg="Failed to install SQLite-vec"
        )

    elif backend == "hybrid":
        print_info("Hybrid backend combines fast local SQLite with background Cloudflare sync")

        # First install SQLite-vec for local storage
        print_info("Installing SQLite-vec for local storage...")
        sqlite_success = install_package_safe(
            "sqlite-vec",
            success_msg="SQLite-vec installed successfully",
            error_msg="Failed to install SQLite-vec...