This is page 34 of 35. Use http://codebase.md/doobidoo/mcp-memory-service?page={x} to view the full context.
# Directory Structure
```
├── .claude
│ ├── agents
│ │ ├── amp-bridge.md
│ │ ├── amp-pr-automator.md
│ │ ├── code-quality-guard.md
│ │ ├── gemini-pr-automator.md
│ │ └── github-release-manager.md
│ ├── settings.local.json.backup
│ └── settings.local.json.local
├── .commit-message
├── .dockerignore
├── .env.example
├── .env.sqlite.backup
├── .envnn#
├── .gitattributes
├── .github
│ ├── FUNDING.yml
│ ├── ISSUE_TEMPLATE
│ │ ├── bug_report.yml
│ │ ├── config.yml
│ │ ├── feature_request.yml
│ │ └── performance_issue.yml
│ ├── pull_request_template.md
│ └── workflows
│ ├── bridge-tests.yml
│ ├── CACHE_FIX.md
│ ├── claude-code-review.yml
│ ├── claude.yml
│ ├── cleanup-images.yml.disabled
│ ├── dev-setup-validation.yml
│ ├── docker-publish.yml
│ ├── LATEST_FIXES.md
│ ├── main-optimized.yml.disabled
│ ├── main.yml
│ ├── publish-and-test.yml
│ ├── README_OPTIMIZATION.md
│ ├── release-tag.yml.disabled
│ ├── release.yml
│ ├── roadmap-review-reminder.yml
│ ├── SECRET_CONDITIONAL_FIX.md
│ └── WORKFLOW_FIXES.md
├── .gitignore
├── .mcp.json.backup
├── .mcp.json.template
├── .pyscn
│ ├── .gitignore
│ └── reports
│ └── analyze_20251123_214224.html
├── AGENTS.md
├── archive
│ ├── deployment
│ │ ├── deploy_fastmcp_fixed.sh
│ │ ├── deploy_http_with_mcp.sh
│ │ └── deploy_mcp_v4.sh
│ ├── deployment-configs
│ │ ├── empty_config.yml
│ │ └── smithery.yaml
│ ├── development
│ │ └── test_fastmcp.py
│ ├── docs-removed-2025-08-23
│ │ ├── authentication.md
│ │ ├── claude_integration.md
│ │ ├── claude-code-compatibility.md
│ │ ├── claude-code-integration.md
│ │ ├── claude-code-quickstart.md
│ │ ├── claude-desktop-setup.md
│ │ ├── complete-setup-guide.md
│ │ ├── database-synchronization.md
│ │ ├── development
│ │ │ ├── autonomous-memory-consolidation.md
│ │ │ ├── CLEANUP_PLAN.md
│ │ │ ├── CLEANUP_README.md
│ │ │ ├── CLEANUP_SUMMARY.md
│ │ │ ├── dream-inspired-memory-consolidation.md
│ │ │ ├── hybrid-slm-memory-consolidation.md
│ │ │ ├── mcp-milestone.md
│ │ │ ├── multi-client-architecture.md
│ │ │ ├── test-results.md
│ │ │ └── TIMESTAMP_FIX_SUMMARY.md
│ │ ├── distributed-sync.md
│ │ ├── invocation_guide.md
│ │ ├── macos-intel.md
│ │ ├── master-guide.md
│ │ ├── mcp-client-configuration.md
│ │ ├── multi-client-server.md
│ │ ├── service-installation.md
│ │ ├── sessions
│ │ │ └── MCP_ENHANCEMENT_SESSION_MEMORY_v4.1.0.md
│ │ ├── UBUNTU_SETUP.md
│ │ ├── ubuntu.md
│ │ ├── windows-setup.md
│ │ └── windows.md
│ ├── docs-root-cleanup-2025-08-23
│ │ ├── AWESOME_LIST_SUBMISSION.md
│ │ ├── CLOUDFLARE_IMPLEMENTATION.md
│ │ ├── DOCUMENTATION_ANALYSIS.md
│ │ ├── DOCUMENTATION_CLEANUP_PLAN.md
│ │ ├── DOCUMENTATION_CONSOLIDATION_COMPLETE.md
│ │ ├── LITESTREAM_SETUP_GUIDE.md
│ │ ├── lm_studio_system_prompt.md
│ │ ├── PYTORCH_DOWNLOAD_FIX.md
│ │ └── README-ORIGINAL-BACKUP.md
│ ├── investigations
│ │ └── MACOS_HOOKS_INVESTIGATION.md
│ ├── litestream-configs-v6.3.0
│ │ ├── install_service.sh
│ │ ├── litestream_master_config_fixed.yml
│ │ ├── litestream_master_config.yml
│ │ ├── litestream_replica_config_fixed.yml
│ │ ├── litestream_replica_config.yml
│ │ ├── litestream_replica_simple.yml
│ │ ├── litestream-http.service
│ │ ├── litestream.service
│ │ └── requirements-cloudflare.txt
│ ├── release-notes
│ │ └── release-notes-v7.1.4.md
│ └── setup-development
│ ├── README.md
│ ├── setup_consolidation_mdns.sh
│ ├── STARTUP_SETUP_GUIDE.md
│ └── test_service.sh
├── CHANGELOG-HISTORIC.md
├── CHANGELOG.md
├── claude_commands
│ ├── memory-context.md
│ ├── memory-health.md
│ ├── memory-ingest-dir.md
│ ├── memory-ingest.md
│ ├── memory-recall.md
│ ├── memory-search.md
│ ├── memory-store.md
│ ├── README.md
│ └── session-start.md
├── claude-hooks
│ ├── config.json
│ ├── config.template.json
│ ├── CONFIGURATION.md
│ ├── core
│ │ ├── memory-retrieval.js
│ │ ├── mid-conversation.js
│ │ ├── session-end.js
│ │ ├── session-start.js
│ │ └── topic-change.js
│ ├── debug-pattern-test.js
│ ├── install_claude_hooks_windows.ps1
│ ├── install_hooks.py
│ ├── memory-mode-controller.js
│ ├── MIGRATION.md
│ ├── README-NATURAL-TRIGGERS.md
│ ├── README-phase2.md
│ ├── README.md
│ ├── simple-test.js
│ ├── statusline.sh
│ ├── test-adaptive-weights.js
│ ├── test-dual-protocol-hook.js
│ ├── test-mcp-hook.js
│ ├── test-natural-triggers.js
│ ├── test-recency-scoring.js
│ ├── tests
│ │ ├── integration-test.js
│ │ ├── phase2-integration-test.js
│ │ ├── test-code-execution.js
│ │ ├── test-cross-session.json
│ │ ├── test-session-tracking.json
│ │ └── test-threading.json
│ ├── utilities
│ │ ├── adaptive-pattern-detector.js
│ │ ├── context-formatter.js
│ │ ├── context-shift-detector.js
│ │ ├── conversation-analyzer.js
│ │ ├── dynamic-context-updater.js
│ │ ├── git-analyzer.js
│ │ ├── mcp-client.js
│ │ ├── memory-client.js
│ │ ├── memory-scorer.js
│ │ ├── performance-manager.js
│ │ ├── project-detector.js
│ │ ├── session-tracker.js
│ │ ├── tiered-conversation-monitor.js
│ │ └── version-checker.js
│ └── WINDOWS-SESSIONSTART-BUG.md
├── CLAUDE.md
├── CODE_OF_CONDUCT.md
├── CONTRIBUTING.md
├── Development-Sprint-November-2025.md
├── docs
│ ├── amp-cli-bridge.md
│ ├── api
│ │ ├── code-execution-interface.md
│ │ ├── memory-metadata-api.md
│ │ ├── PHASE1_IMPLEMENTATION_SUMMARY.md
│ │ ├── PHASE2_IMPLEMENTATION_SUMMARY.md
│ │ ├── PHASE2_REPORT.md
│ │ └── tag-standardization.md
│ ├── architecture
│ │ ├── search-enhancement-spec.md
│ │ └── search-examples.md
│ ├── architecture.md
│ ├── archive
│ │ └── obsolete-workflows
│ │ ├── load_memory_context.md
│ │ └── README.md
│ ├── assets
│ │ └── images
│ │ ├── dashboard-v3.3.0-preview.png
│ │ ├── memory-awareness-hooks-example.png
│ │ ├── project-infographic.svg
│ │ └── README.md
│ ├── CLAUDE_CODE_QUICK_REFERENCE.md
│ ├── cloudflare-setup.md
│ ├── deployment
│ │ ├── docker.md
│ │ ├── dual-service.md
│ │ ├── production-guide.md
│ │ └── systemd-service.md
│ ├── development
│ │ ├── ai-agent-instructions.md
│ │ ├── code-quality
│ │ │ ├── phase-2a-completion.md
│ │ │ ├── phase-2a-handle-get-prompt.md
│ │ │ ├── phase-2a-index.md
│ │ │ ├── phase-2a-install-package.md
│ │ │ └── phase-2b-session-summary.md
│ │ ├── code-quality-workflow.md
│ │ ├── dashboard-workflow.md
│ │ ├── issue-management.md
│ │ ├── pr-review-guide.md
│ │ ├── refactoring-notes.md
│ │ ├── release-checklist.md
│ │ └── todo-tracker.md
│ ├── docker-optimized-build.md
│ ├── document-ingestion.md
│ ├── DOCUMENTATION_AUDIT.md
│ ├── enhancement-roadmap-issue-14.md
│ ├── examples
│ │ ├── analysis-scripts.js
│ │ ├── maintenance-session-example.md
│ │ ├── memory-distribution-chart.jsx
│ │ └── tag-schema.json
│ ├── first-time-setup.md
│ ├── glama-deployment.md
│ ├── guides
│ │ ├── advanced-command-examples.md
│ │ ├── chromadb-migration.md
│ │ ├── commands-vs-mcp-server.md
│ │ ├── mcp-enhancements.md
│ │ ├── mdns-service-discovery.md
│ │ ├── memory-consolidation-guide.md
│ │ ├── migration.md
│ │ ├── scripts.md
│ │ └── STORAGE_BACKENDS.md
│ ├── HOOK_IMPROVEMENTS.md
│ ├── hooks
│ │ └── phase2-code-execution-migration.md
│ ├── http-server-management.md
│ ├── ide-compatability.md
│ ├── IMAGE_RETENTION_POLICY.md
│ ├── images
│ │ └── dashboard-placeholder.md
│ ├── implementation
│ │ ├── health_checks.md
│ │ └── performance.md
│ ├── IMPLEMENTATION_PLAN_HTTP_SSE.md
│ ├── integration
│ │ ├── homebrew.md
│ │ └── multi-client.md
│ ├── integrations
│ │ ├── gemini.md
│ │ ├── groq-bridge.md
│ │ ├── groq-integration-summary.md
│ │ └── groq-model-comparison.md
│ ├── integrations.md
│ ├── legacy
│ │ └── dual-protocol-hooks.md
│ ├── LM_STUDIO_COMPATIBILITY.md
│ ├── maintenance
│ │ └── memory-maintenance.md
│ ├── mastery
│ │ ├── api-reference.md
│ │ ├── architecture-overview.md
│ │ ├── configuration-guide.md
│ │ ├── local-setup-and-run.md
│ │ ├── testing-guide.md
│ │ └── troubleshooting.md
│ ├── migration
│ │ └── code-execution-api-quick-start.md
│ ├── natural-memory-triggers
│ │ ├── cli-reference.md
│ │ ├── installation-guide.md
│ │ └── performance-optimization.md
│ ├── oauth-setup.md
│ ├── pr-graphql-integration.md
│ ├── quick-setup-cloudflare-dual-environment.md
│ ├── README.md
│ ├── remote-configuration-wiki-section.md
│ ├── research
│ │ ├── code-execution-interface-implementation.md
│ │ └── code-execution-interface-summary.md
│ ├── ROADMAP.md
│ ├── sqlite-vec-backend.md
│ ├── statistics
│ │ ├── charts
│ │ │ ├── activity_patterns.png
│ │ │ ├── contributors.png
│ │ │ ├── growth_trajectory.png
│ │ │ ├── monthly_activity.png
│ │ │ └── october_sprint.png
│ │ ├── data
│ │ │ ├── activity_by_day.csv
│ │ │ ├── activity_by_hour.csv
│ │ │ ├── contributors.csv
│ │ │ └── monthly_activity.csv
│ │ ├── generate_charts.py
│ │ └── REPOSITORY_STATISTICS.md
│ ├── technical
│ │ ├── development.md
│ │ ├── memory-migration.md
│ │ ├── migration-log.md
│ │ ├── sqlite-vec-embedding-fixes.md
│ │ └── tag-storage.md
│ ├── testing
│ │ └── regression-tests.md
│ ├── testing-cloudflare-backend.md
│ ├── troubleshooting
│ │ ├── cloudflare-api-token-setup.md
│ │ ├── cloudflare-authentication.md
│ │ ├── general.md
│ │ ├── hooks-quick-reference.md
│ │ ├── pr162-schema-caching-issue.md
│ │ ├── session-end-hooks.md
│ │ └── sync-issues.md
│ └── tutorials
│ ├── advanced-techniques.md
│ ├── data-analysis.md
│ └── demo-session-walkthrough.md
├── examples
│ ├── claude_desktop_config_template.json
│ ├── claude_desktop_config_windows.json
│ ├── claude-desktop-http-config.json
│ ├── config
│ │ └── claude_desktop_config.json
│ ├── http-mcp-bridge.js
│ ├── memory_export_template.json
│ ├── README.md
│ ├── setup
│ │ └── setup_multi_client_complete.py
│ └── start_https_example.sh
├── install_service.py
├── install.py
├── LICENSE
├── NOTICE
├── pyproject.toml
├── pytest.ini
├── README.md
├── run_server.py
├── scripts
│ ├── .claude
│ │ └── settings.local.json
│ ├── archive
│ │ └── check_missing_timestamps.py
│ ├── backup
│ │ ├── backup_memories.py
│ │ ├── backup_sqlite_vec.sh
│ │ ├── export_distributable_memories.sh
│ │ └── restore_memories.py
│ ├── benchmarks
│ │ ├── benchmark_code_execution_api.py
│ │ ├── benchmark_hybrid_sync.py
│ │ └── benchmark_server_caching.py
│ ├── database
│ │ ├── analyze_sqlite_vec_db.py
│ │ ├── check_sqlite_vec_status.py
│ │ ├── db_health_check.py
│ │ └── simple_timestamp_check.py
│ ├── development
│ │ ├── debug_server_initialization.py
│ │ ├── find_orphaned_files.py
│ │ ├── fix_mdns.sh
│ │ ├── fix_sitecustomize.py
│ │ ├── remote_ingest.sh
│ │ ├── setup-git-merge-drivers.sh
│ │ ├── uv-lock-merge.sh
│ │ └── verify_hybrid_sync.py
│ ├── hooks
│ │ └── pre-commit
│ ├── installation
│ │ ├── install_linux_service.py
│ │ ├── install_macos_service.py
│ │ ├── install_uv.py
│ │ ├── install_windows_service.py
│ │ ├── install.py
│ │ ├── setup_backup_cron.sh
│ │ ├── setup_claude_mcp.sh
│ │ └── setup_cloudflare_resources.py
│ ├── linux
│ │ ├── service_status.sh
│ │ ├── start_service.sh
│ │ ├── stop_service.sh
│ │ ├── uninstall_service.sh
│ │ └── view_logs.sh
│ ├── maintenance
│ │ ├── assign_memory_types.py
│ │ ├── check_memory_types.py
│ │ ├── cleanup_corrupted_encoding.py
│ │ ├── cleanup_memories.py
│ │ ├── cleanup_organize.py
│ │ ├── consolidate_memory_types.py
│ │ ├── consolidation_mappings.json
│ │ ├── delete_orphaned_vectors_fixed.py
│ │ ├── fast_cleanup_duplicates_with_tracking.sh
│ │ ├── find_all_duplicates.py
│ │ ├── find_cloudflare_duplicates.py
│ │ ├── find_duplicates.py
│ │ ├── memory-types.md
│ │ ├── README.md
│ │ ├── recover_timestamps_from_cloudflare.py
│ │ ├── regenerate_embeddings.py
│ │ ├── repair_malformed_tags.py
│ │ ├── repair_memories.py
│ │ ├── repair_sqlite_vec_embeddings.py
│ │ ├── repair_zero_embeddings.py
│ │ ├── restore_from_json_export.py
│ │ └── scan_todos.sh
│ ├── migration
│ │ ├── cleanup_mcp_timestamps.py
│ │ ├── legacy
│ │ │ └── migrate_chroma_to_sqlite.py
│ │ ├── mcp-migration.py
│ │ ├── migrate_sqlite_vec_embeddings.py
│ │ ├── migrate_storage.py
│ │ ├── migrate_tags.py
│ │ ├── migrate_timestamps.py
│ │ ├── migrate_to_cloudflare.py
│ │ ├── migrate_to_sqlite_vec.py
│ │ ├── migrate_v5_enhanced.py
│ │ ├── TIMESTAMP_CLEANUP_README.md
│ │ └── verify_mcp_timestamps.py
│ ├── pr
│ │ ├── amp_collect_results.sh
│ │ ├── amp_detect_breaking_changes.sh
│ │ ├── amp_generate_tests.sh
│ │ ├── amp_pr_review.sh
│ │ ├── amp_quality_gate.sh
│ │ ├── amp_suggest_fixes.sh
│ │ ├── auto_review.sh
│ │ ├── detect_breaking_changes.sh
│ │ ├── generate_tests.sh
│ │ ├── lib
│ │ │ └── graphql_helpers.sh
│ │ ├── quality_gate.sh
│ │ ├── resolve_threads.sh
│ │ ├── run_pyscn_analysis.sh
│ │ ├── run_quality_checks.sh
│ │ ├── thread_status.sh
│ │ └── watch_reviews.sh
│ ├── quality
│ │ ├── fix_dead_code_install.sh
│ │ ├── phase1_dead_code_analysis.md
│ │ ├── phase2_complexity_analysis.md
│ │ ├── README_PHASE1.md
│ │ ├── README_PHASE2.md
│ │ ├── track_pyscn_metrics.sh
│ │ └── weekly_quality_review.sh
│ ├── README.md
│ ├── run
│ │ ├── run_mcp_memory.sh
│ │ ├── run-with-uv.sh
│ │ └── start_sqlite_vec.sh
│ ├── run_memory_server.py
│ ├── server
│ │ ├── check_http_server.py
│ │ ├── check_server_health.py
│ │ ├── memory_offline.py
│ │ ├── preload_models.py
│ │ ├── run_http_server.py
│ │ ├── run_memory_server.py
│ │ ├── start_http_server.bat
│ │ └── start_http_server.sh
│ ├── service
│ │ ├── deploy_dual_services.sh
│ │ ├── install_http_service.sh
│ │ ├── mcp-memory-http.service
│ │ ├── mcp-memory.service
│ │ ├── memory_service_manager.sh
│ │ ├── service_control.sh
│ │ ├── service_utils.py
│ │ └── update_service.sh
│ ├── sync
│ │ ├── check_drift.py
│ │ ├── claude_sync_commands.py
│ │ ├── export_memories.py
│ │ ├── import_memories.py
│ │ ├── litestream
│ │ │ ├── apply_local_changes.sh
│ │ │ ├── enhanced_memory_store.sh
│ │ │ ├── init_staging_db.sh
│ │ │ ├── io.litestream.replication.plist
│ │ │ ├── manual_sync.sh
│ │ │ ├── memory_sync.sh
│ │ │ ├── pull_remote_changes.sh
│ │ │ ├── push_to_remote.sh
│ │ │ ├── README.md
│ │ │ ├── resolve_conflicts.sh
│ │ │ ├── setup_local_litestream.sh
│ │ │ ├── setup_remote_litestream.sh
│ │ │ ├── staging_db_init.sql
│ │ │ ├── stash_local_changes.sh
│ │ │ ├── sync_from_remote_noconfig.sh
│ │ │ └── sync_from_remote.sh
│ │ ├── README.md
│ │ ├── safe_cloudflare_update.sh
│ │ ├── sync_memory_backends.py
│ │ └── sync_now.py
│ ├── testing
│ │ ├── run_complete_test.py
│ │ ├── run_memory_test.sh
│ │ ├── simple_test.py
│ │ ├── test_cleanup_logic.py
│ │ ├── test_cloudflare_backend.py
│ │ ├── test_docker_functionality.py
│ │ ├── test_installation.py
│ │ ├── test_mdns.py
│ │ ├── test_memory_api.py
│ │ ├── test_memory_simple.py
│ │ ├── test_migration.py
│ │ ├── test_search_api.py
│ │ ├── test_sqlite_vec_embeddings.py
│ │ ├── test_sse_events.py
│ │ ├── test-connection.py
│ │ └── test-hook.js
│ ├── utils
│ │ ├── claude_commands_utils.py
│ │ ├── generate_personalized_claude_md.sh
│ │ ├── groq
│ │ ├── groq_agent_bridge.py
│ │ ├── list-collections.py
│ │ ├── memory_wrapper_uv.py
│ │ ├── query_memories.py
│ │ ├── smithery_wrapper.py
│ │ ├── test_groq_bridge.sh
│ │ └── uv_wrapper.py
│ └── validation
│ ├── check_dev_setup.py
│ ├── check_documentation_links.py
│ ├── diagnose_backend_config.py
│ ├── validate_configuration_complete.py
│ ├── validate_memories.py
│ ├── validate_migration.py
│ ├── validate_timestamp_integrity.py
│ ├── verify_environment.py
│ ├── verify_pytorch_windows.py
│ └── verify_torch.py
├── SECURITY.md
├── selective_timestamp_recovery.py
├── SPONSORS.md
├── src
│ └── mcp_memory_service
│ ├── __init__.py
│ ├── api
│ │ ├── __init__.py
│ │ ├── client.py
│ │ ├── operations.py
│ │ ├── sync_wrapper.py
│ │ └── types.py
│ ├── backup
│ │ ├── __init__.py
│ │ └── scheduler.py
│ ├── cli
│ │ ├── __init__.py
│ │ ├── ingestion.py
│ │ ├── main.py
│ │ └── utils.py
│ ├── config.py
│ ├── consolidation
│ │ ├── __init__.py
│ │ ├── associations.py
│ │ ├── base.py
│ │ ├── clustering.py
│ │ ├── compression.py
│ │ ├── consolidator.py
│ │ ├── decay.py
│ │ ├── forgetting.py
│ │ ├── health.py
│ │ └── scheduler.py
│ ├── dependency_check.py
│ ├── discovery
│ │ ├── __init__.py
│ │ ├── client.py
│ │ └── mdns_service.py
│ ├── embeddings
│ │ ├── __init__.py
│ │ └── onnx_embeddings.py
│ ├── ingestion
│ │ ├── __init__.py
│ │ ├── base.py
│ │ ├── chunker.py
│ │ ├── csv_loader.py
│ │ ├── json_loader.py
│ │ ├── pdf_loader.py
│ │ ├── registry.py
│ │ ├── semtools_loader.py
│ │ └── text_loader.py
│ ├── lm_studio_compat.py
│ ├── mcp_server.py
│ ├── models
│ │ ├── __init__.py
│ │ └── memory.py
│ ├── server.py
│ ├── services
│ │ ├── __init__.py
│ │ └── memory_service.py
│ ├── storage
│ │ ├── __init__.py
│ │ ├── base.py
│ │ ├── cloudflare.py
│ │ ├── factory.py
│ │ ├── http_client.py
│ │ ├── hybrid.py
│ │ └── sqlite_vec.py
│ ├── sync
│ │ ├── __init__.py
│ │ ├── exporter.py
│ │ ├── importer.py
│ │ └── litestream_config.py
│ ├── utils
│ │ ├── __init__.py
│ │ ├── cache_manager.py
│ │ ├── content_splitter.py
│ │ ├── db_utils.py
│ │ ├── debug.py
│ │ ├── document_processing.py
│ │ ├── gpu_detection.py
│ │ ├── hashing.py
│ │ ├── http_server_manager.py
│ │ ├── port_detection.py
│ │ ├── system_detection.py
│ │ └── time_parser.py
│ └── web
│ ├── __init__.py
│ ├── api
│ │ ├── __init__.py
│ │ ├── analytics.py
│ │ ├── backup.py
│ │ ├── consolidation.py
│ │ ├── documents.py
│ │ ├── events.py
│ │ ├── health.py
│ │ ├── manage.py
│ │ ├── mcp.py
│ │ ├── memories.py
│ │ ├── search.py
│ │ └── sync.py
│ ├── app.py
│ ├── dependencies.py
│ ├── oauth
│ │ ├── __init__.py
│ │ ├── authorization.py
│ │ ├── discovery.py
│ │ ├── middleware.py
│ │ ├── models.py
│ │ ├── registration.py
│ │ └── storage.py
│ ├── sse.py
│ └── static
│ ├── app.js
│ ├── index.html
│ ├── README.md
│ ├── sse_test.html
│ └── style.css
├── start_http_debug.bat
├── start_http_server.sh
├── test_document.txt
├── test_version_checker.js
├── tests
│ ├── __init__.py
│ ├── api
│ │ ├── __init__.py
│ │ ├── test_compact_types.py
│ │ └── test_operations.py
│ ├── bridge
│ │ ├── mock_responses.js
│ │ ├── package-lock.json
│ │ ├── package.json
│ │ └── test_http_mcp_bridge.js
│ ├── conftest.py
│ ├── consolidation
│ │ ├── __init__.py
│ │ ├── conftest.py
│ │ ├── test_associations.py
│ │ ├── test_clustering.py
│ │ ├── test_compression.py
│ │ ├── test_consolidator.py
│ │ ├── test_decay.py
│ │ └── test_forgetting.py
│ ├── contracts
│ │ └── api-specification.yml
│ ├── integration
│ │ ├── package-lock.json
│ │ ├── package.json
│ │ ├── test_api_key_fallback.py
│ │ ├── test_api_memories_chronological.py
│ │ ├── test_api_tag_time_search.py
│ │ ├── test_api_with_memory_service.py
│ │ ├── test_bridge_integration.js
│ │ ├── test_cli_interfaces.py
│ │ ├── test_cloudflare_connection.py
│ │ ├── test_concurrent_clients.py
│ │ ├── test_data_serialization_consistency.py
│ │ ├── test_http_server_startup.py
│ │ ├── test_mcp_memory.py
│ │ ├── test_mdns_integration.py
│ │ ├── test_oauth_basic_auth.py
│ │ ├── test_oauth_flow.py
│ │ ├── test_server_handlers.py
│ │ └── test_store_memory.py
│ ├── performance
│ │ ├── test_background_sync.py
│ │ └── test_hybrid_live.py
│ ├── README.md
│ ├── smithery
│ │ └── test_smithery.py
│ ├── sqlite
│ │ └── simple_sqlite_vec_test.py
│ ├── test_client.py
│ ├── test_content_splitting.py
│ ├── test_database.py
│ ├── test_hybrid_cloudflare_limits.py
│ ├── test_hybrid_storage.py
│ ├── test_memory_ops.py
│ ├── test_semantic_search.py
│ ├── test_sqlite_vec_storage.py
│ ├── test_time_parser.py
│ ├── test_timestamp_preservation.py
│ ├── timestamp
│ │ ├── test_hook_vs_manual_storage.py
│ │ ├── test_issue99_final_validation.py
│ │ ├── test_search_retrieval_inconsistency.py
│ │ ├── test_timestamp_issue.py
│ │ └── test_timestamp_simple.py
│ └── unit
│ ├── conftest.py
│ ├── test_cloudflare_storage.py
│ ├── test_csv_loader.py
│ ├── test_fastapi_dependencies.py
│ ├── test_import.py
│ ├── test_json_loader.py
│ ├── test_mdns_simple.py
│ ├── test_mdns.py
│ ├── test_memory_service.py
│ ├── test_memory.py
│ ├── test_semtools_loader.py
│ ├── test_storage_interface_compatibility.py
│ └── test_tag_time_filtering.py
├── tools
│ ├── docker
│ │ ├── DEPRECATED.md
│ │ ├── docker-compose.http.yml
│ │ ├── docker-compose.pythonpath.yml
│ │ ├── docker-compose.standalone.yml
│ │ ├── docker-compose.uv.yml
│ │ ├── docker-compose.yml
│ │ ├── docker-entrypoint-persistent.sh
│ │ ├── docker-entrypoint-unified.sh
│ │ ├── docker-entrypoint.sh
│ │ ├── Dockerfile
│ │ ├── Dockerfile.glama
│ │ ├── Dockerfile.slim
│ │ ├── README.md
│ │ └── test-docker-modes.sh
│ └── README.md
└── uv.lock
```
# Files
--------------------------------------------------------------------------------
/src/mcp_memory_service/server.py:
--------------------------------------------------------------------------------
```python
# Copyright 2024 Heinrich Krupp
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""
MCP Memory Service
Copyright (c) 2024 Heinrich Krupp
Licensed under the MIT License. See LICENSE file in the project root for full license text.
"""
import sys
import os
import socket
import time
import logging
import psutil
# Client detection for environment-aware behavior
def detect_mcp_client():
"""Detect which MCP client is running this server."""
try:
# Get the parent process (the MCP client)
current_process = psutil.Process()
parent = current_process.parent()
if parent:
parent_name = parent.name().lower()
parent_exe = parent.exe() if hasattr(parent, 'exe') else ""
# Check for Claude Desktop
if 'claude' in parent_name or 'claude' in parent_exe.lower():
return 'claude_desktop'
# Check for LM Studio
if 'lmstudio' in parent_name or 'lm-studio' in parent_name or 'lmstudio' in parent_exe.lower():
return 'lm_studio'
# Check command line for additional clues
try:
cmdline = parent.cmdline()
cmdline_str = ' '.join(cmdline).lower()
if 'claude' in cmdline_str:
return 'claude_desktop'
if 'lmstudio' in cmdline_str or 'lm-studio' in cmdline_str:
return 'lm_studio'
except (OSError, IndexError, AttributeError) as e:
logger.debug(f"Could not detect client from process: {e}")
pass
# Fallback: check environment variables
if os.getenv('CLAUDE_DESKTOP'):
return 'claude_desktop'
if os.getenv('LM_STUDIO'):
return 'lm_studio'
# Default to Claude Desktop for strict JSON compliance
return 'claude_desktop'
except Exception:
# If detection fails, default to Claude Desktop (strict mode)
return 'claude_desktop'
# Detect the current MCP client
MCP_CLIENT = detect_mcp_client()
# Custom logging handler that routes INFO/DEBUG to stdout, WARNING/ERROR to stderr
class DualStreamHandler(logging.Handler):
"""Client-aware handler that adjusts logging behavior based on MCP client."""
def __init__(self, client_type='claude_desktop'):
super().__init__()
self.client_type = client_type
self.stdout_handler = logging.StreamHandler(sys.stdout)
self.stderr_handler = logging.StreamHandler(sys.stderr)
# Set the same formatter for both handlers
formatter = logging.Formatter('%(levelname)s:%(name)s:%(message)s')
self.stdout_handler.setFormatter(formatter)
self.stderr_handler.setFormatter(formatter)
def emit(self, record):
"""Route log records based on client type and level."""
# For Claude Desktop: strict JSON mode - suppress most output, route everything to stderr
if self.client_type == 'claude_desktop':
# Only emit WARNING and above to stderr to maintain JSON protocol
if record.levelno >= logging.WARNING:
self.stderr_handler.emit(record)
# Suppress INFO/DEBUG for Claude Desktop to prevent JSON parsing errors
return
# For LM Studio: enhanced mode with dual-stream
if record.levelno >= logging.WARNING: # WARNING, ERROR, CRITICAL
self.stderr_handler.emit(record)
else: # DEBUG, INFO
self.stdout_handler.emit(record)
# Configure logging with client-aware handler BEFORE any imports that use logging
log_level = os.getenv('LOG_LEVEL', 'WARNING').upper() # Default to WARNING for performance
root_logger = logging.getLogger()
root_logger.setLevel(getattr(logging, log_level, logging.WARNING))
# Remove any existing handlers to avoid duplicates
for handler in root_logger.handlers[:]:
root_logger.removeHandler(handler)
# Add our custom client-aware handler
client_aware_handler = DualStreamHandler(client_type=MCP_CLIENT)
root_logger.addHandler(client_aware_handler)
logger = logging.getLogger(__name__)
# Enhanced path detection for Claude Desktop compatibility
def setup_python_paths():
"""Setup Python paths for dependency access."""
current_dir = os.path.dirname(os.path.dirname(os.path.dirname(__file__)))
# Check for virtual environment
potential_venv_paths = [
os.path.join(current_dir, 'venv', 'Lib', 'site-packages'), # Windows venv
os.path.join(current_dir, 'venv', 'lib', 'python3.11', 'site-packages'), # Linux/Mac venv
os.path.join(current_dir, '.venv', 'Lib', 'site-packages'), # Windows .venv
os.path.join(current_dir, '.venv', 'lib', 'python3.11', 'site-packages'), # Linux/Mac .venv
]
for venv_path in potential_venv_paths:
if os.path.exists(venv_path):
sys.path.insert(0, venv_path)
logger.debug(f"Added venv path: {venv_path}")
break
# For Claude Desktop: also check if we can access global site-packages
try:
import site
global_paths = site.getsitepackages()
user_path = site.getusersitepackages()
# Add user site-packages if not blocked by PYTHONNOUSERSITE
if not os.environ.get('PYTHONNOUSERSITE') and user_path not in sys.path:
sys.path.append(user_path)
logger.debug(f"Added user site-packages: {user_path}")
# Add global site-packages if available
for path in global_paths:
if path not in sys.path:
sys.path.append(path)
logger.debug(f"Added global site-packages: {path}")
except Exception as e:
logger.warning(f"Could not access site-packages: {e}")
# Setup paths before other imports
setup_python_paths()
import asyncio
import traceback
import json
import platform
from collections import deque
from typing import List, Dict, Any, Optional, Tuple
from datetime import datetime, timedelta
from mcp.server.models import InitializationOptions
import mcp.types as types
from mcp.server import NotificationOptions, Server
import mcp.server.stdio
from mcp.types import Resource, Prompt
from . import __version__
from .lm_studio_compat import patch_mcp_for_lm_studio, add_windows_timeout_handling
from .dependency_check import run_dependency_check, get_recommended_timeout
from .config import (
BACKUPS_PATH,
SERVER_NAME,
SERVER_VERSION,
STORAGE_BACKEND,
SQLITE_VEC_PATH,
CONSOLIDATION_ENABLED,
CONSOLIDATION_CONFIG,
CONSOLIDATION_SCHEDULE,
INCLUDE_HOSTNAME,
# Cloudflare configuration
CLOUDFLARE_API_TOKEN,
CLOUDFLARE_ACCOUNT_ID,
CLOUDFLARE_VECTORIZE_INDEX,
CLOUDFLARE_D1_DATABASE_ID,
CLOUDFLARE_R2_BUCKET,
CLOUDFLARE_EMBEDDING_MODEL,
CLOUDFLARE_LARGE_CONTENT_THRESHOLD,
CLOUDFLARE_MAX_RETRIES,
CLOUDFLARE_BASE_DELAY,
# Hybrid backend configuration
HYBRID_SYNC_INTERVAL,
HYBRID_BATCH_SIZE,
HYBRID_SYNC_ON_STARTUP
)
# Storage imports will be done conditionally in the server class
from .models.memory import Memory
from .utils.hashing import generate_content_hash
from .utils.document_processing import _process_and_store_chunk
from .utils.system_detection import (
get_system_info,
print_system_diagnostics,
AcceleratorType
)
from .services.memory_service import MemoryService
from .utils.time_parser import extract_time_expression, parse_time_expression
# Consolidation system imports (conditional)
if CONSOLIDATION_ENABLED:
from .consolidation.base import ConsolidationConfig
from .consolidation.consolidator import DreamInspiredConsolidator
from .consolidation.scheduler import ConsolidationScheduler
# Note: Logging is already configured at the top of the file with dual-stream handler
# Configure performance-critical module logging
if not os.getenv('DEBUG_MODE'):
# Set higher log levels for performance-critical modules
for module_name in ['sentence_transformers', 'transformers', 'torch', 'numpy']:
logging.getLogger(module_name).setLevel(logging.WARNING)
# Check if UV is being used
def check_uv_environment():
"""Check if UV is being used and provide recommendations if not."""
running_with_uv = 'UV_ACTIVE' in os.environ or any('uv' in arg.lower() for arg in sys.argv)
if not running_with_uv:
logger.info("Memory server is running without UV. For better performance and dependency management, consider using UV:")
logger.info(" pip install uv")
logger.info(" uv run memory")
else:
logger.info("Memory server is running with UV")
def check_version_consistency():
"""
Check if installed package version matches source code version.
Warns if version mismatch detected (common "stale venv" issue).
This helps catch the scenario where source code is updated but
the package wasn't reinstalled with 'pip install -e .'.
"""
try:
# Get source code version (from __init__.py)
source_version = __version__
# Get installed package version (from package metadata)
try:
import pkg_resources
installed_version = pkg_resources.get_distribution("mcp-memory-service").version
except:
# If pkg_resources fails, try importlib.metadata (Python 3.8+)
try:
from importlib import metadata
installed_version = metadata.version("mcp-memory-service")
except:
# Can't determine installed version - skip check
return
# Compare versions
if installed_version != source_version:
logger.warning("=" * 70)
logger.warning("⚠️ VERSION MISMATCH DETECTED!")
logger.warning(f" Source code: v{source_version}")
logger.warning(f" Installed: v{installed_version}")
logger.warning("")
logger.warning(" This usually means you need to run:")
logger.warning(" pip install -e . --force-reinstall")
logger.warning("")
logger.warning(" Then restart the MCP server:")
logger.warning(" - In Claude Code: Run /mcp")
logger.warning(" - In Claude Desktop: Restart the application")
logger.warning("=" * 70)
else:
logger.debug(f"Version check OK: v{source_version}")
except Exception as e:
# Don't fail server startup on version check errors
logger.debug(f"Version check failed (non-critical): {e}")
# Configure environment variables based on detected system
def configure_environment():
"""Configure environment variables based on detected system."""
system_info = get_system_info()
# Log system information
logger.info(f"Detected system: {system_info.os_name} {system_info.architecture}")
logger.info(f"Memory: {system_info.memory_gb:.2f} GB")
logger.info(f"Accelerator: {system_info.accelerator}")
# Set environment variables for better cross-platform compatibility
os.environ["PYTORCH_ENABLE_MPS_FALLBACK"] = "1"
# For Apple Silicon, ensure we use MPS when available
if system_info.architecture == "arm64" and system_info.os_name == "darwin":
logger.info("Configuring for Apple Silicon")
os.environ["PYTORCH_MPS_HIGH_WATERMARK_RATIO"] = "0.0"
# For Windows with limited GPU memory, use smaller chunks
if system_info.os_name == "windows" and system_info.accelerator == AcceleratorType.CUDA:
logger.info("Configuring for Windows with CUDA")
os.environ["PYTORCH_CUDA_ALLOC_CONF"] = "max_split_size_mb:128"
# For Linux with ROCm, ensure we use the right backend
if system_info.os_name == "linux" and system_info.accelerator == AcceleratorType.ROCm:
logger.info("Configuring for Linux with ROCm")
os.environ["HSA_OVERRIDE_GFX_VERSION"] = "10.3.0"
# For systems with limited memory, reduce cache sizes
if system_info.memory_gb < 8:
logger.info("Configuring for low-memory system")
# Use BACKUPS_PATH parent directory for model caches
cache_base = os.path.dirname(BACKUPS_PATH)
os.environ["TRANSFORMERS_CACHE"] = os.path.join(cache_base, "model_cache")
os.environ["HF_HOME"] = os.path.join(cache_base, "hf_cache")
os.environ["SENTENCE_TRANSFORMERS_HOME"] = os.path.join(cache_base, "st_cache")
# Configure environment before any imports that might use it
configure_environment()
# Performance optimization environment variables
def configure_performance_environment():
"""Configure environment variables for optimal performance."""
# PyTorch optimizations
os.environ["PYTORCH_ENABLE_MPS_FALLBACK"] = "1"
os.environ["PYTORCH_CUDA_ALLOC_CONF"] = "max_split_size_mb:128,garbage_collection_threshold:0.6"
# CPU optimizations
os.environ["OMP_NUM_THREADS"] = str(min(8, os.cpu_count() or 1))
os.environ["MKL_NUM_THREADS"] = str(min(8, os.cpu_count() or 1))
# Disable unnecessary features for performance
os.environ["TOKENIZERS_PARALLELISM"] = "false"
os.environ["TRANSFORMERS_NO_ADVISORY_WARNINGS"] = "1"
# Async CUDA operations
os.environ["CUDA_LAUNCH_BLOCKING"] = "0"
# Apply performance optimizations
configure_performance_environment()
# =============================================================================
# GLOBAL CACHING FOR MCP SERVER PERFORMANCE OPTIMIZATION
# =============================================================================
# Module-level caches to persist storage/service instances across MCP tool calls.
# This reduces initialization overhead from ~1,810ms to <400ms on cache hits.
#
# Cache Keys:
# - Storage: "{backend_type}:{db_path}" (e.g., "sqlite_vec:/path/to/db")
# - MemoryService: storage instance ID (id(storage))
#
# Thread Safety:
# - Uses asyncio.Lock to prevent race conditions during concurrent access
#
# Lifecycle:
# - Cached instances persist for the lifetime of the Python process
# - NOT cleared between MCP tool calls (intentional for performance)
# - Cleaned up on process shutdown
_STORAGE_CACHE: Dict[str, Any] = {} # Storage instances keyed by "{backend}:{path}"
_MEMORY_SERVICE_CACHE: Dict[int, Any] = {} # MemoryService instances keyed by storage ID
_CACHE_LOCK: Optional[asyncio.Lock] = None # Initialized on first use to avoid event loop issues
_CACHE_STATS = {
"storage_hits": 0,
"storage_misses": 0,
"service_hits": 0,
"service_misses": 0,
"total_calls": 0,
"initialization_times": [] # Track initialization durations for cache misses
}
def _get_cache_lock() -> asyncio.Lock:
"""Get or create the global cache lock (lazy initialization to avoid event loop issues)."""
global _CACHE_LOCK
if _CACHE_LOCK is None:
_CACHE_LOCK = asyncio.Lock()
return _CACHE_LOCK
def _get_or_create_memory_service(storage: Any) -> Any:
"""
Get cached MemoryService or create new one.
Args:
storage: Storage instance to use as cache key
Returns:
MemoryService instance (cached or newly created)
"""
from .services.memory import MemoryService
storage_id = id(storage)
if storage_id in _MEMORY_SERVICE_CACHE:
memory_service = _MEMORY_SERVICE_CACHE[storage_id]
_CACHE_STATS["service_hits"] += 1
logger.info(f"✅ MemoryService Cache HIT - Reusing service instance (storage_id: {storage_id})")
else:
_CACHE_STATS["service_misses"] += 1
logger.info(f"❌ MemoryService Cache MISS - Creating new service instance...")
# Initialize memory service with shared business logic
memory_service = MemoryService(storage)
# Cache the memory service instance
_MEMORY_SERVICE_CACHE[storage_id] = memory_service
logger.info(f"💾 Cached MemoryService instance (storage_id: {storage_id})")
return memory_service
def _log_cache_performance(start_time: float) -> None:
"""
Log comprehensive cache performance statistics.
Args:
start_time: Timer start time to calculate total elapsed time
"""
total_time = (time.time() - start_time) * 1000
cache_hit_rate = (
(_CACHE_STATS["storage_hits"] + _CACHE_STATS["service_hits"]) /
(_CACHE_STATS["total_calls"] * 2) # 2 caches per call
) * 100
logger.info(
f"📊 Cache Stats - "
f"Hit Rate: {cache_hit_rate:.1f}% | "
f"Storage: {_CACHE_STATS['storage_hits']}H/{_CACHE_STATS['storage_misses']}M | "
f"Service: {_CACHE_STATS['service_hits']}H/{_CACHE_STATS['service_misses']}M | "
f"Total Time: {total_time:.1f}ms"
)
class MemoryServer:
def __init__(self):
"""Initialize the server with hardware-aware configuration."""
self.server = Server(SERVER_NAME)
self.system_info = get_system_info()
# Initialize query time tracking
self.query_times = deque(maxlen=50) # Keep last 50 query times for averaging
# Initialize progress tracking
self.current_progress = {} # Track ongoing operations
# Initialize consolidation system (if enabled)
self.consolidator = None
self.consolidation_scheduler = None
if CONSOLIDATION_ENABLED:
try:
config = ConsolidationConfig(**CONSOLIDATION_CONFIG)
self.consolidator = None # Will be initialized after storage
self.consolidation_scheduler = None # Will be initialized after consolidator
logger.info("Consolidation system will be initialized after storage")
except Exception as e:
logger.error(f"Failed to initialize consolidation config: {e}")
self.consolidator = None
self.consolidation_scheduler = None
try:
# Initialize paths
logger.info(f"Creating directories if they don't exist...")
os.makedirs(BACKUPS_PATH, exist_ok=True)
# Log system diagnostics
logger.info(f"Initializing on {platform.system()} {platform.machine()} with Python {platform.python_version()}")
logger.info(f"Using accelerator: {self.system_info.accelerator}")
# DEFER STORAGE INITIALIZATION - Initialize storage lazily when needed
# This prevents hanging during server startup due to embedding model loading
logger.info(f"Deferring {STORAGE_BACKEND} storage initialization to prevent hanging")
if MCP_CLIENT == 'lm_studio':
print(f"Deferring {STORAGE_BACKEND} storage initialization to prevent startup hanging", file=sys.stdout, flush=True)
self.storage = None
self.memory_service = None
self._storage_initialized = False
except Exception as e:
logger.error(f"Initialization error: {str(e)}")
logger.error(traceback.format_exc())
# Set storage to None to prevent any hanging
self.storage = None
self.memory_service = None
self._storage_initialized = False
# Register handlers
self.register_handlers()
logger.info("Server initialization complete")
# Test handler registration with proper arguments
try:
logger.info("Testing handler registration...")
capabilities = self.server.get_capabilities(
notification_options=NotificationOptions(),
experimental_capabilities={}
)
logger.info(f"Server capabilities: {capabilities}")
if MCP_CLIENT == 'lm_studio':
print(f"Server capabilities registered successfully!", file=sys.stdout, flush=True)
except Exception as e:
logger.error(f"Handler registration test failed: {str(e)}")
print(f"Handler registration issue: {str(e)}", file=sys.stderr, flush=True)
def record_query_time(self, query_time_ms: float):
"""Record a query time for averaging."""
self.query_times.append(query_time_ms)
logger.debug(f"Recorded query time: {query_time_ms:.2f}ms")
def get_average_query_time(self) -> float:
"""Get the average query time from recent operations."""
if not self.query_times:
return 0.0
avg = sum(self.query_times) / len(self.query_times)
logger.debug(f"Average query time: {avg:.2f}ms (from {len(self.query_times)} samples)")
return round(avg, 2)
async def send_progress_notification(self, operation_id: str, progress: float, message: str = None):
"""Send a progress notification for a long-running operation."""
try:
# Store progress for potential querying
self.current_progress[operation_id] = {
"progress": progress,
"message": message or f"Operation {operation_id}: {progress:.0f}% complete",
"timestamp": datetime.now().isoformat()
}
# Send notification if server supports it
if hasattr(self.server, 'send_progress_notification'):
await self.server.send_progress_notification(
progress=progress,
progress_token=operation_id,
message=message
)
logger.debug(f"Progress {operation_id}: {progress:.0f}% - {message}")
# Clean up completed operations
if progress >= 100:
self.current_progress.pop(operation_id, None)
except Exception as e:
logger.debug(f"Could not send progress notification: {e}")
def get_operation_progress(self, operation_id: str) -> Optional[Dict[str, Any]]:
"""Get the current progress of an operation."""
return self.current_progress.get(operation_id)
async def _initialize_storage_with_timeout(self):
"""Initialize storage with timeout and caching optimization."""
global _STORAGE_CACHE, _MEMORY_SERVICE_CACHE, _CACHE_STATS
# Track call statistics
_CACHE_STATS["total_calls"] += 1
start_time = time.time()
logger.info(f"🚀 EAGER INIT Call #{_CACHE_STATS['total_calls']}: Checking global cache...")
# Acquire lock for thread-safe cache access
cache_lock = _get_cache_lock()
async with cache_lock:
# Generate cache key for storage backend
cache_key = f"{STORAGE_BACKEND}:{SQLITE_VEC_PATH}"
# Check storage cache
if cache_key in _STORAGE_CACHE:
self.storage = _STORAGE_CACHE[cache_key]
_CACHE_STATS["storage_hits"] += 1
logger.info(f"✅ Storage Cache HIT - Reusing {STORAGE_BACKEND} instance (key: {cache_key})")
self._storage_initialized = True
# Check memory service cache and log performance
self.memory_service = _get_or_create_memory_service(self.storage)
_log_cache_performance(start_time)
return True # Cached initialization succeeded
# Cache miss - proceed with initialization
_CACHE_STATS["storage_misses"] += 1
logger.info(f"❌ Storage Cache MISS - Initializing {STORAGE_BACKEND} instance...")
try:
logger.info(f"🚀 EAGER INIT: Starting {STORAGE_BACKEND} storage initialization...")
logger.info(f"🔧 EAGER INIT: Environment check - STORAGE_BACKEND={STORAGE_BACKEND}")
# Log all Cloudflare config values for debugging
if STORAGE_BACKEND == 'cloudflare':
logger.info(f"🔧 EAGER INIT: Cloudflare config validation:")
logger.info(f" API_TOKEN: {'SET' if CLOUDFLARE_API_TOKEN else 'NOT SET'}")
logger.info(f" ACCOUNT_ID: {CLOUDFLARE_ACCOUNT_ID}")
logger.info(f" VECTORIZE_INDEX: {CLOUDFLARE_VECTORIZE_INDEX}")
logger.info(f" D1_DATABASE_ID: {CLOUDFLARE_D1_DATABASE_ID}")
logger.info(f" R2_BUCKET: {CLOUDFLARE_R2_BUCKET}")
logger.info(f" EMBEDDING_MODEL: {CLOUDFLARE_EMBEDDING_MODEL}")
if STORAGE_BACKEND == 'sqlite_vec':
# Check for multi-client coordination mode
from .utils.port_detection import ServerCoordinator
coordinator = ServerCoordinator()
coordination_mode = await coordinator.detect_mode()
logger.info(f"🔧 EAGER INIT: SQLite-vec - detected coordination mode: {coordination_mode}")
if coordination_mode == "http_client":
# Use HTTP client to connect to existing server
from .storage.http_client import HTTPClientStorage
self.storage = HTTPClientStorage()
logger.info(f"✅ EAGER INIT: Using HTTP client storage")
elif coordination_mode == "http_server":
# Try to auto-start HTTP server for coordination
from .utils.http_server_manager import auto_start_http_server_if_needed
server_started = await auto_start_http_server_if_needed()
if server_started:
# Wait a moment for the server to be ready, then use HTTP client
await asyncio.sleep(2)
from .storage.http_client import HTTPClientStorage
self.storage = HTTPClientStorage()
logger.info(f"✅ EAGER INIT: Started HTTP server and using HTTP client storage")
else:
# Fall back to direct SQLite-vec storage
from . import storage
import importlib
storage_module = importlib.import_module('mcp_memory_service.storage.sqlite_vec')
SqliteVecMemoryStorage = storage_module.SqliteVecMemoryStorage
self.storage = SqliteVecMemoryStorage(SQLITE_VEC_PATH)
logger.info(f"✅ EAGER INIT: HTTP server auto-start failed, using direct SQLite-vec storage")
else:
# Import sqlite-vec storage module (supports dynamic class replacement)
from . import storage
import importlib
storage_module = importlib.import_module('mcp_memory_service.storage.sqlite_vec')
SqliteVecMemoryStorage = storage_module.SqliteVecMemoryStorage
self.storage = SqliteVecMemoryStorage(SQLITE_VEC_PATH)
logger.info(f"✅ EAGER INIT: Using direct SQLite-vec storage at {SQLITE_VEC_PATH}")
elif STORAGE_BACKEND == 'cloudflare':
# Initialize Cloudflare storage
logger.info(f"☁️ EAGER INIT: Importing CloudflareStorage...")
from .storage.cloudflare import CloudflareStorage
logger.info(f"☁️ EAGER INIT: Creating CloudflareStorage instance...")
self.storage = CloudflareStorage(
api_token=CLOUDFLARE_API_TOKEN,
account_id=CLOUDFLARE_ACCOUNT_ID,
vectorize_index=CLOUDFLARE_VECTORIZE_INDEX,
d1_database_id=CLOUDFLARE_D1_DATABASE_ID,
r2_bucket=CLOUDFLARE_R2_BUCKET,
embedding_model=CLOUDFLARE_EMBEDDING_MODEL,
large_content_threshold=CLOUDFLARE_LARGE_CONTENT_THRESHOLD,
max_retries=CLOUDFLARE_MAX_RETRIES,
base_delay=CLOUDFLARE_BASE_DELAY
)
logger.info(f"✅ EAGER INIT: CloudflareStorage instance created with index: {CLOUDFLARE_VECTORIZE_INDEX}")
elif STORAGE_BACKEND == 'hybrid':
# Initialize Hybrid storage (SQLite-vec + Cloudflare)
logger.info(f"🔄 EAGER INIT: Using Hybrid storage...")
from .storage.hybrid import HybridMemoryStorage
# Prepare Cloudflare configuration dict
cloudflare_config = None
if all([CLOUDFLARE_API_TOKEN, CLOUDFLARE_ACCOUNT_ID, CLOUDFLARE_VECTORIZE_INDEX, CLOUDFLARE_D1_DATABASE_ID]):
cloudflare_config = {
'api_token': CLOUDFLARE_API_TOKEN,
'account_id': CLOUDFLARE_ACCOUNT_ID,
'vectorize_index': CLOUDFLARE_VECTORIZE_INDEX,
'd1_database_id': CLOUDFLARE_D1_DATABASE_ID,
'r2_bucket': CLOUDFLARE_R2_BUCKET,
'embedding_model': CLOUDFLARE_EMBEDDING_MODEL,
'large_content_threshold': CLOUDFLARE_LARGE_CONTENT_THRESHOLD,
'max_retries': CLOUDFLARE_MAX_RETRIES,
'base_delay': CLOUDFLARE_BASE_DELAY
}
logger.info(f"🔄 EAGER INIT: Cloudflare config prepared for hybrid storage")
else:
logger.warning("🔄 EAGER INIT: Incomplete Cloudflare config, hybrid will run in SQLite-only mode")
self.storage = HybridMemoryStorage(
sqlite_db_path=SQLITE_VEC_PATH,
embedding_model="all-MiniLM-L6-v2",
cloudflare_config=cloudflare_config,
sync_interval=HYBRID_SYNC_INTERVAL or 300,
batch_size=HYBRID_BATCH_SIZE or 50
)
logger.info(f"✅ EAGER INIT: HybridMemoryStorage instance created")
else:
# Unknown backend - should not reach here due to factory validation
logger.error(f"❌ EAGER INIT: Unknown storage backend: {STORAGE_BACKEND}")
raise ValueError(f"Unsupported storage backend: {STORAGE_BACKEND}")
# Initialize the storage backend
logger.info(f"🔧 EAGER INIT: Calling storage.initialize()...")
await self.storage.initialize()
logger.info(f"✅ EAGER INIT: storage.initialize() completed successfully")
self._storage_initialized = True
logger.info(f"🎉 EAGER INIT: {STORAGE_BACKEND} storage initialization successful")
# Cache the newly initialized storage instance
async with cache_lock:
_STORAGE_CACHE[cache_key] = self.storage
init_time = (time.time() - start_time) * 1000
_CACHE_STATS["initialization_times"].append(init_time)
logger.info(f"💾 Cached storage instance (key: {cache_key}, init_time: {init_time:.1f}ms)")
# Initialize and cache MemoryService
_CACHE_STATS["service_misses"] += 1
self.memory_service = MemoryService(self.storage)
storage_id = id(self.storage)
_MEMORY_SERVICE_CACHE[storage_id] = self.memory_service
logger.info(f"💾 Cached MemoryService instance (storage_id: {storage_id})")
# Verify storage type
storage_type = self.storage.__class__.__name__
logger.info(f"🔍 EAGER INIT: Final storage type verification: {storage_type}")
# Initialize consolidation system after storage is ready
await self._initialize_consolidation()
return True
except Exception as e:
logger.error(f"❌ EAGER INIT: Storage initialization failed: {str(e)}")
logger.error(f"📋 EAGER INIT: Full traceback:")
logger.error(traceback.format_exc())
return False
async def _ensure_storage_initialized(self):
"""Lazily initialize storage backend when needed with global caching."""
if not self._storage_initialized:
global _STORAGE_CACHE, _MEMORY_SERVICE_CACHE, _CACHE_STATS
# Track call statistics
_CACHE_STATS["total_calls"] += 1
start_time = time.time()
logger.info(f"🔄 LAZY INIT Call #{_CACHE_STATS['total_calls']}: Checking global cache...")
# Acquire lock for thread-safe cache access
cache_lock = _get_cache_lock()
async with cache_lock:
# Generate cache key for storage backend
cache_key = f"{STORAGE_BACKEND}:{SQLITE_VEC_PATH}"
# Check storage cache
if cache_key in _STORAGE_CACHE:
self.storage = _STORAGE_CACHE[cache_key]
_CACHE_STATS["storage_hits"] += 1
logger.info(f"✅ Storage Cache HIT - Reusing {STORAGE_BACKEND} instance (key: {cache_key})")
self._storage_initialized = True
# Check memory service cache and log performance
self.memory_service = _get_or_create_memory_service(self.storage)
_log_cache_performance(start_time)
return self.storage
# Cache miss - proceed with initialization
_CACHE_STATS["storage_misses"] += 1
logger.info(f"❌ Storage Cache MISS - Initializing {STORAGE_BACKEND} instance...")
try:
logger.info(f"🔄 LAZY INIT: Starting {STORAGE_BACKEND} storage initialization...")
logger.info(f"🔧 LAZY INIT: Environment check - STORAGE_BACKEND={STORAGE_BACKEND}")
# Log all Cloudflare config values for debugging
if STORAGE_BACKEND == 'cloudflare':
logger.info(f"🔧 LAZY INIT: Cloudflare config validation:")
logger.info(f" API_TOKEN: {'SET' if CLOUDFLARE_API_TOKEN else 'NOT SET'}")
logger.info(f" ACCOUNT_ID: {CLOUDFLARE_ACCOUNT_ID}")
logger.info(f" VECTORIZE_INDEX: {CLOUDFLARE_VECTORIZE_INDEX}")
logger.info(f" D1_DATABASE_ID: {CLOUDFLARE_D1_DATABASE_ID}")
logger.info(f" R2_BUCKET: {CLOUDFLARE_R2_BUCKET}")
logger.info(f" EMBEDDING_MODEL: {CLOUDFLARE_EMBEDDING_MODEL}")
if STORAGE_BACKEND == 'sqlite_vec':
# Check for multi-client coordination mode
from .utils.port_detection import ServerCoordinator
coordinator = ServerCoordinator()
coordination_mode = await coordinator.detect_mode()
logger.info(f"🔧 LAZY INIT: SQLite-vec - detected coordination mode: {coordination_mode}")
if coordination_mode == "http_client":
# Use HTTP client to connect to existing server
from .storage.http_client import HTTPClientStorage
self.storage = HTTPClientStorage()
logger.info(f"✅ LAZY INIT: Using HTTP client storage")
elif coordination_mode == "http_server":
# Try to auto-start HTTP server for coordination
from .utils.http_server_manager import auto_start_http_server_if_needed
server_started = await auto_start_http_server_if_needed()
if server_started:
# Wait a moment for the server to be ready, then use HTTP client
await asyncio.sleep(2)
from .storage.http_client import HTTPClientStorage
self.storage = HTTPClientStorage()
logger.info(f"✅ LAZY INIT: Started HTTP server and using HTTP client storage")
else:
# Fall back to direct SQLite-vec storage
import importlib
storage_module = importlib.import_module('mcp_memory_service.storage.sqlite_vec')
SqliteVecMemoryStorage = storage_module.SqliteVecMemoryStorage
self.storage = SqliteVecMemoryStorage(SQLITE_VEC_PATH)
logger.info(f"✅ LAZY INIT: HTTP server auto-start failed, using direct SQLite-vec storage at: {SQLITE_VEC_PATH}")
else:
# Use direct SQLite-vec storage (with WAL mode for concurrent access)
import importlib
storage_module = importlib.import_module('mcp_memory_service.storage.sqlite_vec')
SqliteVecMemoryStorage = storage_module.SqliteVecMemoryStorage
self.storage = SqliteVecMemoryStorage(SQLITE_VEC_PATH)
logger.info(f"✅ LAZY INIT: Created SQLite-vec storage at: {SQLITE_VEC_PATH}")
elif STORAGE_BACKEND == 'cloudflare':
# Cloudflare backend using Vectorize, D1, and R2
logger.info(f"☁️ LAZY INIT: Importing CloudflareStorage...")
from .storage.cloudflare import CloudflareStorage
logger.info(f"☁️ LAZY INIT: Creating CloudflareStorage instance...")
self.storage = CloudflareStorage(
api_token=CLOUDFLARE_API_TOKEN,
account_id=CLOUDFLARE_ACCOUNT_ID,
vectorize_index=CLOUDFLARE_VECTORIZE_INDEX,
d1_database_id=CLOUDFLARE_D1_DATABASE_ID,
r2_bucket=CLOUDFLARE_R2_BUCKET,
embedding_model=CLOUDFLARE_EMBEDDING_MODEL,
large_content_threshold=CLOUDFLARE_LARGE_CONTENT_THRESHOLD,
max_retries=CLOUDFLARE_MAX_RETRIES,
base_delay=CLOUDFLARE_BASE_DELAY
)
logger.info(f"✅ LAZY INIT: Created Cloudflare storage with Vectorize index: {CLOUDFLARE_VECTORIZE_INDEX}")
elif STORAGE_BACKEND == 'hybrid':
# Hybrid backend using SQLite-vec as primary and Cloudflare as secondary
logger.info(f"🔄 LAZY INIT: Importing HybridMemoryStorage...")
from .storage.hybrid import HybridMemoryStorage
# Prepare Cloudflare configuration dict
cloudflare_config = None
if all([CLOUDFLARE_API_TOKEN, CLOUDFLARE_ACCOUNT_ID, CLOUDFLARE_VECTORIZE_INDEX, CLOUDFLARE_D1_DATABASE_ID]):
cloudflare_config = {
'api_token': CLOUDFLARE_API_TOKEN,
'account_id': CLOUDFLARE_ACCOUNT_ID,
'vectorize_index': CLOUDFLARE_VECTORIZE_INDEX,
'd1_database_id': CLOUDFLARE_D1_DATABASE_ID,
'r2_bucket': CLOUDFLARE_R2_BUCKET,
'embedding_model': CLOUDFLARE_EMBEDDING_MODEL,
'large_content_threshold': CLOUDFLARE_LARGE_CONTENT_THRESHOLD,
'max_retries': CLOUDFLARE_MAX_RETRIES,
'base_delay': CLOUDFLARE_BASE_DELAY
}
logger.info(f"🔄 LAZY INIT: Cloudflare config prepared for hybrid storage")
else:
logger.warning("🔄 LAZY INIT: Incomplete Cloudflare config, hybrid will run in SQLite-only mode")
logger.info(f"🔄 LAZY INIT: Creating HybridMemoryStorage instance...")
self.storage = HybridMemoryStorage(
sqlite_db_path=SQLITE_VEC_PATH,
embedding_model="all-MiniLM-L6-v2",
cloudflare_config=cloudflare_config,
sync_interval=HYBRID_SYNC_INTERVAL or 300,
batch_size=HYBRID_BATCH_SIZE or 50
)
logger.info(f"✅ LAZY INIT: Created Hybrid storage at: {SQLITE_VEC_PATH} with Cloudflare sync")
else:
# Unknown/unsupported backend
logger.error("=" * 70)
logger.error(f"❌ LAZY INIT: Unsupported storage backend: {STORAGE_BACKEND}")
logger.error("")
logger.error("Supported backends:")
logger.error(" - sqlite_vec (recommended for single-device use)")
logger.error(" - cloudflare (cloud storage)")
logger.error(" - hybrid (recommended for multi-device use)")
logger.error("=" * 70)
raise ValueError(
f"Unsupported storage backend: {STORAGE_BACKEND}. "
"Use 'sqlite_vec', 'cloudflare', or 'hybrid'."
)
# Initialize the storage backend
logger.info(f"🔧 LAZY INIT: Calling storage.initialize()...")
await self.storage.initialize()
logger.info(f"✅ LAZY INIT: storage.initialize() completed successfully")
# Verify the storage is properly initialized
if hasattr(self.storage, 'is_initialized') and not self.storage.is_initialized():
# Get detailed status for debugging
if hasattr(self.storage, 'get_initialization_status'):
status = self.storage.get_initialization_status()
logger.error(f"❌ LAZY INIT: Storage initialization incomplete: {status}")
raise RuntimeError("Storage initialization incomplete")
self._storage_initialized = True
storage_type = self.storage.__class__.__name__
logger.info(f"🎉 LAZY INIT: Storage backend ({STORAGE_BACKEND}) initialization successful")
logger.info(f"🔍 LAZY INIT: Final storage type verification: {storage_type}")
# Cache the newly initialized storage instance
async with cache_lock:
_STORAGE_CACHE[cache_key] = self.storage
init_time = (time.time() - start_time) * 1000
_CACHE_STATS["initialization_times"].append(init_time)
logger.info(f"💾 Cached storage instance (key: {cache_key}, init_time: {init_time:.1f}ms)")
# Initialize and cache MemoryService
_CACHE_STATS["service_misses"] += 1
self.memory_service = MemoryService(self.storage)
storage_id = id(self.storage)
_MEMORY_SERVICE_CACHE[storage_id] = self.memory_service
logger.info(f"💾 Cached MemoryService instance (storage_id: {storage_id})")
# Initialize consolidation system after storage is ready
await self._initialize_consolidation()
except Exception as e:
logger.error(f"❌ LAZY INIT: Failed to initialize {STORAGE_BACKEND} storage: {str(e)}")
logger.error(f"📋 LAZY INIT: Full traceback:")
logger.error(traceback.format_exc())
# Set storage to None to indicate failure
self.storage = None
self._storage_initialized = False
raise
return self.storage
async def initialize(self):
"""Async initialization method with eager storage initialization and timeout."""
try:
# Run any async initialization tasks here
logger.info("🚀 SERVER INIT: Starting async initialization...")
# Print system diagnostics only for LM Studio (avoid JSON parsing errors in Claude Desktop)
if MCP_CLIENT == 'lm_studio':
print("\n=== System Diagnostics ===", file=sys.stdout, flush=True)
print(f"OS: {self.system_info.os_name} {self.system_info.os_version}", file=sys.stdout, flush=True)
print(f"Architecture: {self.system_info.architecture}", file=sys.stdout, flush=True)
print(f"Memory: {self.system_info.memory_gb:.2f} GB", file=sys.stdout, flush=True)
print(f"Accelerator: {self.system_info.accelerator}", file=sys.stdout, flush=True)
print(f"Python: {platform.python_version()}", file=sys.stdout, flush=True)
# Log environment info
logger.info(f"🔧 SERVER INIT: Environment - STORAGE_BACKEND={STORAGE_BACKEND}")
# Attempt eager storage initialization with timeout
# Get dynamic timeout based on system and dependency status
timeout_seconds = get_recommended_timeout()
logger.info(f"⏱️ SERVER INIT: Attempting eager storage initialization (timeout: {timeout_seconds}s)...")
if MCP_CLIENT == 'lm_studio':
print(f"Attempting eager storage initialization (timeout: {timeout_seconds}s)...", file=sys.stdout, flush=True)
try:
init_task = asyncio.create_task(self._initialize_storage_with_timeout())
success = await asyncio.wait_for(init_task, timeout=timeout_seconds)
if success:
if MCP_CLIENT == 'lm_studio':
print("[OK] Eager storage initialization successful", file=sys.stdout, flush=True)
logger.info("✅ SERVER INIT: Eager storage initialization completed successfully")
# Verify storage type after successful eager init
if hasattr(self, 'storage') and self.storage:
storage_type = self.storage.__class__.__name__
logger.info(f"🔍 SERVER INIT: Eager init resulted in storage type: {storage_type}")
else:
if MCP_CLIENT == 'lm_studio':
print("[WARNING] Eager storage initialization failed, will use lazy loading", file=sys.stdout, flush=True)
logger.warning("⚠️ SERVER INIT: Eager initialization failed, falling back to lazy loading")
# Reset state for lazy loading
self.storage = None
self._storage_initialized = False
except asyncio.TimeoutError:
if MCP_CLIENT == 'lm_studio':
print("[TIMEOUT] Eager storage initialization timed out, will use lazy loading", file=sys.stdout, flush=True)
logger.warning(f"⏱️ SERVER INIT: Storage initialization timed out after {timeout_seconds}s, falling back to lazy loading")
# Reset state for lazy loading
self.storage = None
self._storage_initialized = False
except Exception as e:
if MCP_CLIENT == 'lm_studio':
print(f"[WARNING] Eager initialization error: {str(e)}, will use lazy loading", file=sys.stdout, flush=True)
logger.warning(f"⚠️ SERVER INIT: Eager initialization error: {str(e)}, falling back to lazy loading")
logger.warning(f"📋 SERVER INIT: Eager init error traceback:")
logger.warning(traceback.format_exc())
# Reset state for lazy loading
self.storage = None
self._storage_initialized = False
# Add explicit console output for Smithery to see (only for LM Studio)
if MCP_CLIENT == 'lm_studio':
print("MCP Memory Service initialization completed", file=sys.stdout, flush=True)
logger.info("🎉 SERVER INIT: Async initialization completed")
return True
except Exception as e:
logger.error(f"❌ SERVER INIT: Async initialization error: {str(e)}")
logger.error(f"📋 SERVER INIT: Full traceback:")
logger.error(traceback.format_exc())
# Add explicit console error output for Smithery to see
print(f"Initialization error: {str(e)}", file=sys.stderr, flush=True)
# Don't raise the exception, just return False
return False
async def validate_database_health(self):
"""Validate database health during initialization."""
from .utils.db_utils import validate_database, repair_database
try:
# Check database health
is_valid, message = await validate_database(self.storage)
if not is_valid:
logger.warning(f"Database validation failed: {message}")
# Attempt repair
logger.info("Attempting database repair...")
repair_success, repair_message = await repair_database(self.storage)
if not repair_success:
logger.error(f"Database repair failed: {repair_message}")
return False
else:
logger.info(f"Database repair successful: {repair_message}")
return True
else:
logger.info(f"Database validation successful: {message}")
return True
except Exception as e:
logger.error(f"Database validation error: {str(e)}")
return False
async def _initialize_consolidation(self):
"""Initialize the consolidation system after storage is ready."""
if not CONSOLIDATION_ENABLED or not self._storage_initialized:
return
try:
if self.consolidator is None:
# Create consolidation config
config = ConsolidationConfig(**CONSOLIDATION_CONFIG)
# Initialize the consolidator with storage
self.consolidator = DreamInspiredConsolidator(self.storage, config)
logger.info("Dream-inspired consolidator initialized")
# Initialize the scheduler if not disabled
if any(schedule != 'disabled' for schedule in CONSOLIDATION_SCHEDULE.values()):
self.consolidation_scheduler = ConsolidationScheduler(
self.consolidator,
CONSOLIDATION_SCHEDULE,
enabled=True
)
# Start the scheduler
if await self.consolidation_scheduler.start():
logger.info("Consolidation scheduler started successfully")
else:
logger.warning("Failed to start consolidation scheduler")
self.consolidation_scheduler = None
else:
logger.info("Consolidation scheduler disabled (all schedules set to 'disabled')")
except Exception as e:
logger.error(f"Failed to initialize consolidation system: {e}")
logger.error(traceback.format_exc())
self.consolidator = None
self.consolidation_scheduler = None
def handle_method_not_found(self, method: str) -> None:
"""Custom handler for unsupported methods.
This logs the unsupported method request but doesn't raise an exception,
allowing the MCP server to handle it with a standard JSON-RPC error response.
"""
logger.warning(f"Unsupported method requested: {method}")
# The MCP server will automatically respond with a Method not found error
# We don't need to do anything else here
def register_handlers(self):
# Enhanced Resources implementation
@self.server.list_resources()
async def handle_list_resources() -> List[Resource]:
"""List available memory resources."""
await self._ensure_storage_initialized()
resources = [
types.Resource(
uri="memory://stats",
name="Memory Statistics",
description="Current memory database statistics",
mimeType="application/json"
),
types.Resource(
uri="memory://tags",
name="Available Tags",
description="List of all tags used in memories",
mimeType="application/json"
),
types.Resource(
uri="memory://recent/10",
name="Recent Memories",
description="10 most recent memories",
mimeType="application/json"
)
]
# Add tag-specific resources for existing tags
try:
all_tags = await self.storage.get_all_tags()
for tag in all_tags[:5]: # Limit to first 5 tags for resources
resources.append(types.Resource(
uri=f"memory://tag/{tag}",
name=f"Memories tagged '{tag}'",
description=f"All memories with tag '{tag}'",
mimeType="application/json"
))
except AttributeError:
# get_all_tags method not available on this storage backend
pass
except Exception as e:
logger.warning(f"Failed to load tag resources: {e}")
pass
return resources
@self.server.read_resource()
async def handle_read_resource(uri: str) -> str:
"""Read a specific memory resource."""
await self._ensure_storage_initialized()
import json
from urllib.parse import unquote
try:
if uri == "memory://stats":
# Get memory statistics
stats = await self.storage.get_stats()
return json.dumps(stats, indent=2)
elif uri == "memory://tags":
# Get all available tags
tags = await self.storage.get_all_tags()
return json.dumps({"tags": tags, "count": len(tags)}, indent=2)
elif uri.startswith("memory://recent/"):
# Get recent memories
n = int(uri.split("/")[-1])
memories = await self.storage.get_recent_memories(n)
return json.dumps({
"memories": [m.to_dict() for m in memories],
"count": len(memories)
}, indent=2, default=str)
elif uri.startswith("memory://tag/"):
# Get memories by tag
tag = unquote(uri.split("/", 3)[-1])
memories = await self.storage.search_by_tag([tag])
return json.dumps({
"tag": tag,
"memories": [m.to_dict() for m in memories],
"count": len(memories)
}, indent=2, default=str)
elif uri.startswith("memory://search/"):
# Dynamic search
query = unquote(uri.split("/", 3)[-1])
results = await self.storage.search(query, n_results=10)
return json.dumps({
"query": query,
"results": [r.to_dict() for r in results],
"count": len(results)
}, indent=2, default=str)
else:
return json.dumps({"error": f"Resource not found: {uri}"}, indent=2)
except Exception as e:
logger.error(f"Error reading resource {uri}: {e}")
return json.dumps({"error": str(e)}, indent=2)
@self.server.list_resource_templates()
async def handle_list_resource_templates() -> List[types.ResourceTemplate]:
"""List resource templates for dynamic queries."""
return [
types.ResourceTemplate(
uriTemplate="memory://recent/{n}",
name="Recent Memories",
description="Get N most recent memories",
mimeType="application/json"
),
types.ResourceTemplate(
uriTemplate="memory://tag/{tag}",
name="Memories by Tag",
description="Get all memories with a specific tag",
mimeType="application/json"
),
types.ResourceTemplate(
uriTemplate="memory://search/{query}",
name="Search Memories",
description="Search memories by query",
mimeType="application/json"
)
]
@self.server.list_prompts()
async def handle_list_prompts() -> List[types.Prompt]:
"""List available guided prompts for memory operations."""
return [
types.Prompt(
name="memory_review",
description="Review and organize memories from a specific time period",
arguments=[
types.PromptArgument(
name="time_period",
description="Time period to review (e.g., 'last week', 'yesterday', '2 days ago')",
required=True
),
types.PromptArgument(
name="focus_area",
description="Optional area to focus on (e.g., 'work', 'personal', 'learning')",
required=False
)
]
),
types.Prompt(
name="memory_analysis",
description="Analyze patterns and themes in stored memories",
arguments=[
types.PromptArgument(
name="tags",
description="Tags to analyze (comma-separated)",
required=False
),
types.PromptArgument(
name="time_range",
description="Time range to analyze (e.g., 'last month', 'all time')",
required=False
)
]
),
types.Prompt(
name="knowledge_export",
description="Export memories in a specific format",
arguments=[
types.PromptArgument(
name="format",
description="Export format (json, markdown, text)",
required=True
),
types.PromptArgument(
name="filter",
description="Filter criteria (tags or search query)",
required=False
)
]
),
types.Prompt(
name="memory_cleanup",
description="Identify and remove duplicate or outdated memories",
arguments=[
types.PromptArgument(
name="older_than",
description="Remove memories older than (e.g., '6 months', '1 year')",
required=False
),
types.PromptArgument(
name="similarity_threshold",
description="Similarity threshold for duplicates (0.0-1.0)",
required=False
)
]
),
types.Prompt(
name="learning_session",
description="Store structured learning notes from a study session",
arguments=[
types.PromptArgument(
name="topic",
description="Learning topic or subject",
required=True
),
types.PromptArgument(
name="key_points",
description="Key points learned (comma-separated)",
required=True
),
types.PromptArgument(
name="questions",
description="Questions or areas for further study",
required=False
)
]
)
]
@self.server.get_prompt()
async def handle_get_prompt(name: str, arguments: dict) -> types.GetPromptResult:
"""Handle prompt execution with provided arguments."""
await self._ensure_storage_initialized()
# Dispatch to specific prompt handler
if name == "memory_review":
messages = await self._prompt_memory_review(arguments)
elif name == "memory_analysis":
messages = await self._prompt_memory_analysis(arguments)
elif name == "knowledge_export":
messages = await self._prompt_knowledge_export(arguments)
elif name == "memory_cleanup":
messages = await self._prompt_memory_cleanup(arguments)
elif name == "learning_session":
messages = await self._prompt_learning_session(arguments)
else:
messages = [
types.PromptMessage(
role="user",
content=types.TextContent(
type="text",
text=f"Unknown prompt: {name}"
)
)
]
return types.GetPromptResult(
description=f"Result of {name} prompt",
messages=messages
)
# Helper methods for specific prompts
async def _prompt_memory_review(self, arguments: dict) -> list:
"""Generate memory review prompt."""
time_period = arguments.get("time_period", "last week")
focus_area = arguments.get("focus_area", "")
# Retrieve memories from the specified time period
memories = await self.storage.recall_memory(time_period, n_results=20)
prompt_text = f"Review of memories from {time_period}"
if focus_area:
prompt_text += f" (focusing on {focus_area})"
prompt_text += ":\n\n"
if memories:
for mem in memories:
prompt_text += f"- {mem.content}\n"
if mem.metadata.tags:
prompt_text += f" Tags: {', '.join(mem.metadata.tags)}\n"
else:
prompt_text += "No memories found for this time period."
return [
types.PromptMessage(
role="user",
content=types.TextContent(type="text", text=prompt_text)
)
]
async def _prompt_memory_analysis(self, arguments: dict) -> list:
"""Generate memory analysis prompt."""
tags = arguments.get("tags", "").split(",") if arguments.get("tags") else []
time_range = arguments.get("time_range", "all time")
analysis_text = "Memory Analysis"
if tags:
analysis_text += f" for tags: {', '.join(tags)}"
if time_range != "all time":
analysis_text += f" from {time_range}"
analysis_text += "\n\n"
# Get relevant memories
if tags:
memories = await self.storage.search_by_tag(tags)
else:
memories = await self.storage.get_recent_memories(100)
# Analyze patterns
tag_counts = {}
type_counts = {}
for mem in memories:
for tag in mem.metadata.tags:
tag_counts[tag] = tag_counts.get(tag, 0) + 1
mem_type = mem.metadata.memory_type
type_counts[mem_type] = type_counts.get(mem_type, 0) + 1
analysis_text += f"Total memories analyzed: {len(memories)}\n\n"
analysis_text += "Top tags:\n"
for tag, count in sorted(tag_counts.items(), key=lambda x: x[1], reverse=True)[:10]:
analysis_text += f" - {tag}: {count} occurrences\n"
analysis_text += "\nMemory types:\n"
for mem_type, count in type_counts.items():
analysis_text += f" - {mem_type}: {count} memories\n"
return [
types.PromptMessage(
role="user",
content=types.TextContent(type="text", text=analysis_text)
)
]
async def _prompt_knowledge_export(self, arguments: dict) -> list:
"""Generate knowledge export prompt."""
format_type = arguments.get("format", "json")
filter_criteria = arguments.get("filter", "")
# Get memories based on filter
if filter_criteria:
if "," in filter_criteria:
# Assume tags
memories = await self.storage.search_by_tag(filter_criteria.split(","))
else:
# Assume search query
memories = await self.storage.search(filter_criteria, n_results=100)
else:
memories = await self.storage.get_recent_memories(100)
export_text = f"Exported {len(memories)} memories in {format_type} format:\n\n"
if format_type == "markdown":
for mem in memories:
export_text += f"## {mem.metadata.created_at_iso}\n"
export_text += f"{mem.content}\n"
if mem.metadata.tags:
export_text += f"*Tags: {', '.join(mem.metadata.tags)}*\n"
export_text += "\n"
elif format_type == "text":
for mem in memories:
export_text += f"[{mem.metadata.created_at_iso}] {mem.content}\n"
else: # json
import json
export_data = [m.to_dict() for m in memories]
export_text += json.dumps(export_data, indent=2, default=str)
return [
types.PromptMessage(
role="user",
content=types.TextContent(type="text", text=export_text)
)
]
async def _prompt_memory_cleanup(self, arguments: dict) -> list:
"""Generate memory cleanup prompt."""
older_than = arguments.get("older_than", "")
similarity_threshold = float(arguments.get("similarity_threshold", "0.95"))
cleanup_text = "Memory Cleanup Report:\n\n"
# Find duplicates
all_memories = await self.storage.get_recent_memories(1000)
duplicates = []
for i, mem1 in enumerate(all_memories):
for mem2 in all_memories[i+1:]:
# Simple similarity check based on content length
if abs(len(mem1.content) - len(mem2.content)) < 10:
if mem1.content[:50] == mem2.content[:50]:
duplicates.append((mem1, mem2))
cleanup_text += f"Found {len(duplicates)} potential duplicate pairs\n"
if older_than:
cleanup_text += f"\nMemories older than {older_than} can be archived\n"
return [
types.PromptMessage(
role="user",
content=types.TextContent(type="text", text=cleanup_text)
)
]
async def _prompt_learning_session(self, arguments: dict) -> list:
"""Generate learning session prompt."""
topic = arguments.get("topic", "General")
key_points = arguments.get("key_points", "").split(",")
questions = arguments.get("questions", "").split(",") if arguments.get("questions") else []
# Create structured learning note
learning_note = f"# Learning Session: {topic}\n\n"
learning_note += f"Date: {datetime.now().isoformat()}\n\n"
learning_note += "## Key Points:\n"
for point in key_points:
learning_note += f"- {point.strip()}\n"
if questions:
learning_note += "\n## Questions for Further Study:\n"
for question in questions:
learning_note += f"- {question.strip()}\n"
# Store the learning note
memory = Memory(
content=learning_note,
tags=["learning", topic.lower().replace(" ", "_")],
memory_type="learning_note"
)
success, message = await self.storage.store(memory)
response_text = f"Learning session stored successfully!\n\n{learning_note}"
if not success:
response_text = f"Failed to store learning session: {message}"
return [
types.PromptMessage(
role="user",
content=types.TextContent(type="text", text=response_text)
)
]
# Add a custom error handler for unsupported methods
self.server.on_method_not_found = self.handle_method_not_found
@self.server.list_tools()
async def handle_list_tools() -> List[types.Tool]:
logger.info("=== HANDLING LIST_TOOLS REQUEST ===")
try:
tools = [
types.Tool(
name="store_memory",
description="""Store new information with optional tags.
Accepts two tag formats in metadata:
- Array: ["tag1", "tag2"]
- String: "tag1,tag2"
Examples:
# Using array format:
{
"content": "Memory content",
"metadata": {
"tags": ["important", "reference"],
"type": "note"
}
}
# Using string format(preferred):
{
"content": "Memory content",
"metadata": {
"tags": "important,reference",
"type": "note"
}
}""",
inputSchema={
"type": "object",
"properties": {
"content": {
"type": "string",
"description": "The memory content to store, such as a fact, note, or piece of information."
},
"metadata": {
"type": "object",
"description": "Optional metadata about the memory, including tags and type.",
"properties": {
"tags": {
"oneOf": [
{
"type": "array",
"items": {"type": "string"},
"description": "Tags as an array of strings"
},
{
"type": "string",
"description": "Tags as comma-separated string"
}
],
"description": "Tags to categorize the memory. Accepts either an array of strings or a comma-separated string.",
"examples": [
"tag1,tag2,tag3",
["tag1", "tag2", "tag3"]
]
},
"type": {
"type": "string",
"description": "Optional type or category label for the memory, e.g., 'note', 'fact', 'reminder'."
}
}
}
},
"required": ["content"]
}
),
types.Tool(
name="recall_memory",
description="""Retrieve memories using natural language time expressions and optional semantic search.
Supports various time-related expressions such as:
- "yesterday", "last week", "2 days ago"
- "last summer", "this month", "last January"
- "spring", "winter", "Christmas", "Thanksgiving"
- "morning", "evening", "yesterday afternoon"
Examples:
{
"query": "recall what I stored last week"
}
{
"query": "find information about databases from two months ago",
"n_results": 5
}
""",
inputSchema={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Natural language query specifying the time frame or content to recall, e.g., 'last week', 'yesterday afternoon', or a topic."
},
"n_results": {
"type": "number",
"default": 5,
"description": "Maximum number of results to return."
}
},
"required": ["query"]
}
),
types.Tool(
name="retrieve_memory",
description="""Find relevant memories based on query.
Example:
{
"query": "find this memory",
"n_results": 5
}""",
inputSchema={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Search query to find relevant memories based on content."
},
"n_results": {
"type": "number",
"default": 5,
"description": "Maximum number of results to return."
}
},
"required": ["query"]
}
),
types.Tool(
name="search_by_tag",
description="""Search memories by tags. Must use array format.
Returns memories matching ANY of the specified tags.
Example:
{
"tags": ["important", "reference"]
}""",
inputSchema={
"type": "object",
"properties": {
"tags": {
"oneOf": [
{
"type": "array",
"items": {"type": "string"},
"description": "Tags as an array of strings"
},
{
"type": "string",
"description": "Tags as comma-separated string"
}
],
"description": "List of tags to search for. Returns memories matching ANY of these tags. Accepts either an array of strings or a comma-separated string."
}
},
"required": ["tags"]
}
),
types.Tool(
name="delete_memory",
description="""Delete a specific memory by its hash.
Example:
{
"content_hash": "a1b2c3d4..."
}""",
inputSchema={
"type": "object",
"properties": {
"content_hash": {
"type": "string",
"description": "Hash of the memory content to delete. Obtainable from memory metadata."
}
},
"required": ["content_hash"]
}
),
types.Tool(
name="delete_by_tag",
description="""Delete all memories with specific tags.
WARNING: Deletes ALL memories containing any of the specified tags.
Example:
{"tags": ["temporary", "outdated"]}""",
inputSchema={
"type": "object",
"properties": {
"tags": {
"oneOf": [
{
"type": "array",
"items": {"type": "string"},
"description": "Tags as an array of strings"
},
{
"type": "string",
"description": "Tags as comma-separated string"
}
],
"description": "Array of tag labels. Memories containing any of these tags will be deleted. Accepts either an array of strings or a comma-separated string."
}
},
"required": ["tags"]
}
),
types.Tool(
name="delete_by_tags",
description="""Delete all memories containing any of the specified tags.
This is the explicit multi-tag version for API clarity.
WARNING: Deletes ALL memories containing any of the specified tags.
Example:
{
"tags": ["temporary", "outdated", "test"]
}""",
inputSchema={
"type": "object",
"properties": {
"tags": {
"oneOf": [
{
"type": "array",
"items": {"type": "string"},
"description": "Tags as an array of strings"
},
{
"type": "string",
"description": "Tags as comma-separated string"
}
],
"description": "List of tag labels. Memories containing any of these tags will be deleted. Accepts either an array of strings or a comma-separated string."
}
},
"required": ["tags"]
}
),
types.Tool(
name="delete_by_all_tags",
description="""Delete memories that contain ALL of the specified tags.
WARNING: Only deletes memories that have every one of the specified tags.
Example:
{
"tags": ["important", "urgent"]
}""",
inputSchema={
"type": "object",
"properties": {
"tags": {
"oneOf": [
{
"type": "array",
"items": {"type": "string"},
"description": "Tags as an array of strings"
},
{
"type": "string",
"description": "Tags as comma-separated string"
}
],
"description": "List of tag labels. Only memories containing ALL of these tags will be deleted. Accepts either an array of strings or a comma-separated string."
}
},
"required": ["tags"]
}
),
types.Tool(
name="cleanup_duplicates",
description="Find and remove duplicate entries",
inputSchema={
"type": "object",
"properties": {}
}
),
types.Tool(
name="debug_retrieve",
description="""Retrieve memories with debug information.
Example:
{
"query": "debug this",
"n_results": 5,
"similarity_threshold": 0.0
}""",
inputSchema={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Search query for debugging retrieval, e.g., a phrase or keyword."
},
"n_results": {
"type": "number",
"default": 5,
"description": "Maximum number of results to return."
},
"similarity_threshold": {
"type": "number",
"default": 0.0,
"description": "Minimum similarity score threshold for results (0.0 to 1.0)."
}
},
"required": ["query"]
}
),
types.Tool(
name="exact_match_retrieve",
description="""Retrieve memories using exact content match.
Example:
{
"content": "find exactly this"
}""",
inputSchema={
"type": "object",
"properties": {
"content": {
"type": "string",
"description": "Exact content string to match against stored memories."
}
},
"required": ["content"]
}
),
types.Tool(
name="get_raw_embedding",
description="""Get raw embedding vector for debugging purposes.
Example:
{
"content": "text to embed"
}""",
inputSchema={
"type": "object",
"properties": {
"content": {
"type": "string",
"description": "Content to generate embedding for."
}
},
"required": ["content"]
}
),
types.Tool(
name="check_database_health",
description="Check database health and get statistics",
inputSchema={
"type": "object",
"properties": {}
}
),
types.Tool(
name="get_cache_stats",
description="""Get MCP server global cache statistics for performance monitoring.
Returns detailed metrics about storage and memory service caching,
including hit rates, initialization times, and cache sizes.
This tool is useful for:
- Monitoring cache effectiveness
- Debugging performance issues
- Verifying cache persistence across MCP tool calls
Returns cache statistics including total calls, hit rate percentage,
storage/service cache metrics, performance metrics, and backend info.""",
inputSchema={
"type": "object",
"properties": {}
}
),
types.Tool(
name="recall_by_timeframe",
description="""Retrieve memories within a specific timeframe.
Example:
{
"start_date": "2024-01-01",
"end_date": "2024-01-31",
"n_results": 5
}""",
inputSchema={
"type": "object",
"properties": {
"start_date": {
"type": "string",
"format": "date",
"description": "Start date (inclusive) in YYYY-MM-DD format."
},
"end_date": {
"type": "string",
"format": "date",
"description": "End date (inclusive) in YYYY-MM-DD format."
},
"n_results": {
"type": "number",
"default": 5,
"description": "Maximum number of results to return."
}
},
"required": ["start_date"]
}
),
types.Tool(
name="delete_by_timeframe",
description="""Delete memories within a specific timeframe.
Optional tag parameter to filter deletions.
Example:
{
"start_date": "2024-01-01",
"end_date": "2024-01-31",
"tag": "temporary"
}""",
inputSchema={
"type": "object",
"properties": {
"start_date": {
"type": "string",
"format": "date",
"description": "Start date (inclusive) in YYYY-MM-DD format."
},
"end_date": {
"type": "string",
"format": "date",
"description": "End date (inclusive) in YYYY-MM-DD format."
},
"tag": {
"type": "string",
"description": "Optional tag to filter deletions. Only memories with this tag will be deleted."
}
},
"required": ["start_date"]
}
),
types.Tool(
name="delete_before_date",
description="""Delete memories before a specific date.
Optional tag parameter to filter deletions.
Example:
{
"before_date": "2024-01-01",
"tag": "temporary"
}""",
inputSchema={
"type": "object",
"properties": {
"before_date": {"type": "string", "format": "date"},
"tag": {"type": "string"}
},
"required": ["before_date"]
}
),
types.Tool(
name="update_memory_metadata",
description="""Update memory metadata without recreating the entire memory entry.
This provides efficient metadata updates while preserving the original
memory content, embeddings, and optionally timestamps.
Examples:
# Add tags to a memory
{
"content_hash": "abc123...",
"updates": {
"tags": ["important", "reference", "new-tag"]
}
}
# Update memory type and custom metadata
{
"content_hash": "abc123...",
"updates": {
"memory_type": "reminder",
"metadata": {
"priority": "high",
"due_date": "2024-01-15"
}
}
}
# Update custom fields directly
{
"content_hash": "abc123...",
"updates": {
"priority": "urgent",
"status": "active"
}
}""",
inputSchema={
"type": "object",
"properties": {
"content_hash": {
"type": "string",
"description": "The content hash of the memory to update."
},
"updates": {
"type": "object",
"description": "Dictionary of metadata fields to update.",
"properties": {
"tags": {
"oneOf": [
{
"type": "array",
"items": {"type": "string"},
"description": "Tags as an array of strings"
},
{
"type": "string",
"description": "Tags as comma-separated string"
}
],
"description": "Replace existing tags with this list. Accepts either an array of strings or a comma-separated string."
},
"memory_type": {
"type": "string",
"description": "Update the memory type (e.g., 'note', 'reminder', 'fact')."
},
"metadata": {
"type": "object",
"description": "Custom metadata fields to merge with existing metadata."
}
}
},
"preserve_timestamps": {
"type": "boolean",
"default": True,
"description": "Whether to preserve the original created_at timestamp (default: true)."
}
},
"required": ["content_hash", "updates"]
}
)
]
# Add consolidation tools if enabled
if CONSOLIDATION_ENABLED and self.consolidator:
consolidation_tools = [
types.Tool(
name="consolidate_memories",
description="""Run memory consolidation for a specific time horizon.
Performs dream-inspired memory consolidation including:
- Exponential decay scoring
- Creative association discovery
- Semantic clustering and compression
- Controlled forgetting with archival
Example:
{
"time_horizon": "weekly"
}""",
inputSchema={
"type": "object",
"properties": {
"time_horizon": {
"type": "string",
"enum": ["daily", "weekly", "monthly", "quarterly", "yearly"],
"description": "Time horizon for consolidation operations."
}
},
"required": ["time_horizon"]
}
),
types.Tool(
name="consolidation_status",
description="Get status and health information about the consolidation system.",
inputSchema={"type": "object", "properties": {}}
),
types.Tool(
name="consolidation_recommendations",
description="""Get recommendations for consolidation based on current memory state.
Example:
{
"time_horizon": "monthly"
}""",
inputSchema={
"type": "object",
"properties": {
"time_horizon": {
"type": "string",
"enum": ["daily", "weekly", "monthly", "quarterly", "yearly"],
"description": "Time horizon to analyze for consolidation recommendations."
}
},
"required": ["time_horizon"]
}
),
types.Tool(
name="scheduler_status",
description="Get consolidation scheduler status and job information.",
inputSchema={"type": "object", "properties": {}}
),
types.Tool(
name="trigger_consolidation",
description="""Manually trigger a consolidation job.
Example:
{
"time_horizon": "weekly",
"immediate": true
}""",
inputSchema={
"type": "object",
"properties": {
"time_horizon": {
"type": "string",
"enum": ["daily", "weekly", "monthly", "quarterly", "yearly"],
"description": "Time horizon for the consolidation job."
},
"immediate": {
"type": "boolean",
"default": True,
"description": "Whether to run immediately or schedule for later."
}
},
"required": ["time_horizon"]
}
),
types.Tool(
name="pause_consolidation",
description="""Pause consolidation jobs.
Example:
{
"time_horizon": "weekly"
}""",
inputSchema={
"type": "object",
"properties": {
"time_horizon": {
"type": "string",
"enum": ["daily", "weekly", "monthly", "quarterly", "yearly"],
"description": "Specific time horizon to pause, or omit to pause all jobs."
}
}
}
),
types.Tool(
name="resume_consolidation",
description="""Resume consolidation jobs.
Example:
{
"time_horizon": "weekly"
}""",
inputSchema={
"type": "object",
"properties": {
"time_horizon": {
"type": "string",
"enum": ["daily", "weekly", "monthly", "quarterly", "yearly"],
"description": "Specific time horizon to resume, or omit to resume all jobs."
}
}
}
)
]
tools.extend(consolidation_tools)
logger.info(f"Added {len(consolidation_tools)} consolidation tools")
# Add document ingestion tools
ingestion_tools = [
types.Tool(
name="ingest_document",
description="""Ingest a single document file into the memory database.
Supports multiple formats:
- PDF files (.pdf)
- Text files (.txt, .md, .markdown, .rst)
- JSON files (.json)
The document will be parsed, chunked intelligently, and stored
as multiple memories with appropriate metadata.
Example:
{
"file_path": "/path/to/document.pdf",
"tags": ["documentation", "manual"],
"chunk_size": 1000
}""",
inputSchema={
"type": "object",
"properties": {
"file_path": {
"type": "string",
"description": "Path to the document file to ingest."
},
"tags": {
"oneOf": [
{
"type": "array",
"items": {"type": "string"},
"description": "Tags as an array of strings"
},
{
"type": "string",
"description": "Tags as comma-separated string"
}
],
"description": "Optional tags to apply to all memories created from this document. Accepts either an array of strings or a comma-separated string.",
"default": []
},
"chunk_size": {
"type": "number",
"description": "Target size for text chunks in characters (default: 1000).",
"default": 1000
},
"chunk_overlap": {
"type": "number",
"description": "Characters to overlap between chunks (default: 200).",
"default": 200
},
"memory_type": {
"type": "string",
"description": "Type label for created memories (default: 'document').",
"default": "document"
}
},
"required": ["file_path"]
}
),
types.Tool(
name="ingest_directory",
description="""Batch ingest all supported documents from a directory.
Recursively processes all supported file types in the directory,
creating memories with consistent tagging and metadata.
Supported formats: PDF, TXT, MD, JSON
Example:
{
"directory_path": "/path/to/documents",
"tags": ["knowledge-base"],
"recursive": true,
"file_extensions": ["pdf", "md", "txt"]
}""",
inputSchema={
"type": "object",
"properties": {
"directory_path": {
"type": "string",
"description": "Path to the directory containing documents to ingest."
},
"tags": {
"oneOf": [
{
"type": "array",
"items": {"type": "string"},
"description": "Tags as an array of strings"
},
{
"type": "string",
"description": "Tags as comma-separated string"
}
],
"description": "Optional tags to apply to all memories created. Accepts either an array of strings or a comma-separated string.",
"default": []
},
"recursive": {
"type": "boolean",
"description": "Whether to process subdirectories recursively (default: true).",
"default": True
},
"file_extensions": {
"type": "array",
"items": {"type": "string"},
"description": "File extensions to process (default: all supported).",
"default": ["pdf", "txt", "md", "json"]
},
"chunk_size": {
"type": "number",
"description": "Target size for text chunks in characters (default: 1000).",
"default": 1000
},
"max_files": {
"type": "number",
"description": "Maximum number of files to process (default: 100).",
"default": 100
}
},
"required": ["directory_path"]
}
)
]
tools.extend(ingestion_tools)
logger.info(f"Added {len(ingestion_tools)} ingestion tools")
logger.info(f"Returning {len(tools)} tools")
return tools
except Exception as e:
logger.error(f"Error in handle_list_tools: {str(e)}")
logger.error(traceback.format_exc())
raise
@self.server.call_tool()
async def handle_call_tool(name: str, arguments: dict | None) -> List[types.TextContent]:
# Add immediate debugging to catch any protocol issues
if MCP_CLIENT == 'lm_studio':
print(f"TOOL CALL INTERCEPTED: {name}", file=sys.stdout, flush=True)
logger.info(f"=== HANDLING TOOL CALL: {name} ===")
logger.info(f"Arguments: {arguments}")
try:
if arguments is None:
arguments = {}
logger.info(f"Processing tool: {name}")
if MCP_CLIENT == 'lm_studio':
print(f"Processing tool: {name}", file=sys.stdout, flush=True)
if name == "store_memory":
return await self.handle_store_memory(arguments)
elif name == "retrieve_memory":
return await self.handle_retrieve_memory(arguments)
elif name == "recall_memory":
return await self.handle_recall_memory(arguments)
elif name == "search_by_tag":
return await self.handle_search_by_tag(arguments)
elif name == "delete_memory":
return await self.handle_delete_memory(arguments)
elif name == "delete_by_tag":
return await self.handle_delete_by_tag(arguments)
elif name == "delete_by_tags":
return await self.handle_delete_by_tags(arguments)
elif name == "delete_by_all_tags":
return await self.handle_delete_by_all_tags(arguments)
elif name == "cleanup_duplicates":
return await self.handle_cleanup_duplicates(arguments)
elif name == "debug_retrieve":
return await self.handle_debug_retrieve(arguments)
elif name == "exact_match_retrieve":
return await self.handle_exact_match_retrieve(arguments)
elif name == "get_raw_embedding":
return await self.handle_get_raw_embedding(arguments)
elif name == "check_database_health":
logger.info("Calling handle_check_database_health")
return await self.handle_check_database_health(arguments)
elif name == "get_cache_stats":
logger.info("Calling handle_get_cache_stats")
return await self.handle_get_cache_stats(arguments)
elif name == "recall_by_timeframe":
return await self.handle_recall_by_timeframe(arguments)
elif name == "delete_by_timeframe":
return await self.handle_delete_by_timeframe(arguments)
elif name == "delete_before_date":
return await self.handle_delete_before_date(arguments)
elif name == "update_memory_metadata":
logger.info("Calling handle_update_memory_metadata")
return await self.handle_update_memory_metadata(arguments)
# Consolidation tool handlers
elif name == "consolidate_memories":
logger.info("Calling handle_consolidate_memories")
return await self.handle_consolidate_memories(arguments)
elif name == "consolidation_status":
logger.info("Calling handle_consolidation_status")
return await self.handle_consolidation_status(arguments)
elif name == "consolidation_recommendations":
logger.info("Calling handle_consolidation_recommendations")
return await self.handle_consolidation_recommendations(arguments)
elif name == "scheduler_status":
logger.info("Calling handle_scheduler_status")
return await self.handle_scheduler_status(arguments)
elif name == "trigger_consolidation":
logger.info("Calling handle_trigger_consolidation")
return await self.handle_trigger_consolidation(arguments)
elif name == "pause_consolidation":
logger.info("Calling handle_pause_consolidation")
return await self.handle_pause_consolidation(arguments)
elif name == "resume_consolidation":
logger.info("Calling handle_resume_consolidation")
return await self.handle_resume_consolidation(arguments)
elif name == "ingest_document":
logger.info("Calling handle_ingest_document")
return await self.handle_ingest_document(arguments)
elif name == "ingest_directory":
logger.info("Calling handle_ingest_directory")
return await self.handle_ingest_directory(arguments)
else:
logger.warning(f"Unknown tool requested: {name}")
raise ValueError(f"Unknown tool: {name}")
except Exception as e:
error_msg = f"Error in {name}: {str(e)}\n{traceback.format_exc()}"
logger.error(error_msg)
print(f"ERROR in tool execution: {error_msg}", file=sys.stderr, flush=True)
return [types.TextContent(type="text", text=f"Error: {str(e)}")]
async def handle_store_memory(self, arguments: dict) -> List[types.TextContent]:
content = arguments.get("content")
metadata = arguments.get("metadata", {})
if not content:
return [types.TextContent(type="text", text="Error: Content is required")]
try:
# Initialize storage lazily when needed (also initializes memory_service)
await self._ensure_storage_initialized()
# Extract parameters for MemoryService call
tags = metadata.get("tags", "")
memory_type = metadata.get("type", "note") # HTTP server uses metadata.type
client_hostname = arguments.get("client_hostname")
# Call shared MemoryService business logic
result = await self.memory_service.store_memory(
content=content,
tags=tags,
memory_type=memory_type,
metadata=metadata,
client_hostname=client_hostname
)
# Convert MemoryService result to MCP response format
if not result.get("success"):
error_msg = result.get("error", "Unknown error")
return [types.TextContent(type="text", text=f"Error storing memory: {error_msg}")]
if "memories" in result:
# Chunked response - multiple memories created
num_chunks = len(result["memories"])
original_hash = result.get("original_hash", "unknown")
message = f"Successfully stored {num_chunks} memory chunks (original hash: {original_hash[:8]}...)"
else:
# Single memory response
memory_hash = result["memory"]["content_hash"]
message = f"Memory stored successfully (hash: {memory_hash[:8]}...)"
return [types.TextContent(type="text", text=message)]
except Exception as e:
logger.error(f"Error storing memory: {str(e)}\n{traceback.format_exc()}")
return [types.TextContent(type="text", text=f"Error storing memory: {str(e)}")]
async def handle_retrieve_memory(self, arguments: dict) -> List[types.TextContent]:
query = arguments.get("query")
n_results = arguments.get("n_results", 5)
if not query:
return [types.TextContent(type="text", text="Error: Query is required")]
try:
# Initialize storage lazily when needed (also initializes memory_service)
await self._ensure_storage_initialized()
# Track performance
start_time = time.time()
# Call shared MemoryService business logic
result = await self.memory_service.retrieve_memories(
query=query,
n_results=n_results
)
query_time_ms = (time.time() - start_time) * 1000
# Record query time for performance monitoring
self.record_query_time(query_time_ms)
if result.get("error"):
return [types.TextContent(type="text", text=f"Error retrieving memories: {result['error']}")]
memories = result.get("memories", [])
if not memories:
return [types.TextContent(type="text", text="No matching memories found")]
# Format results in HTTP server style (different from MCP server)
formatted_results = []
for i, memory in enumerate(memories):
memory_info = [f"Memory {i+1}:"]
# HTTP server uses created_at instead of timestamp
created_at = memory.get("created_at")
if created_at:
# Parse ISO string and format
try:
# Handle both float (timestamp) and string (ISO format) types
if isinstance(created_at, (int, float)):
dt = datetime.fromtimestamp(created_at)
else:
dt = datetime.fromisoformat(created_at.replace('Z', '+00:00'))
memory_info.append(f"Timestamp: {dt.strftime('%Y-%m-%d %H:%M:%S')}")
except (ValueError, TypeError):
memory_info.append(f"Timestamp: {created_at}")
memory_info.extend([
f"Content: {memory['content']}",
f"Hash: {memory['content_hash']}",
f"Relevance Score: {memory['similarity_score']:.2f}"
])
tags = memory.get("tags", [])
if tags:
memory_info.append(f"Tags: {', '.join(tags)}")
memory_info.append("---")
formatted_results.append("\n".join(memory_info))
return [types.TextContent(
type="text",
text="Found the following memories:\n\n" + "\n".join(formatted_results)
)]
except Exception as e:
logger.error(f"Error retrieving memories: {str(e)}\n{traceback.format_exc()}")
return [types.TextContent(type="text", text=f"Error retrieving memories: {str(e)}")]
async def handle_search_by_tag(self, arguments: dict) -> List[types.TextContent]:
from .services.memory_service import normalize_tags
tags = normalize_tags(arguments.get("tags", []))
if not tags:
return [types.TextContent(type="text", text="Error: Tags are required")]
try:
# Initialize storage lazily when needed (also initializes memory_service)
await self._ensure_storage_initialized()
# Call shared MemoryService business logic
result = await self.memory_service.search_by_tag(tags=tags)
if result.get("error"):
return [types.TextContent(type="text", text=f"Error searching by tags: {result['error']}")]
memories = result.get("memories", [])
if not memories:
return [types.TextContent(
type="text",
text=f"No memories found with tags: {', '.join(tags)}"
)]
formatted_results = []
for i, memory in enumerate(memories):
memory_info = [f"Memory {i+1}:"]
created_at = memory.get("created_at")
if created_at:
try:
# Handle both float (timestamp) and string (ISO format) types
if isinstance(created_at, (int, float)):
dt = datetime.fromtimestamp(created_at)
else:
dt = datetime.fromisoformat(created_at.replace('Z', '+00:00'))
memory_info.append(f"Timestamp: {dt.strftime('%Y-%m-%d %H:%M:%S')}")
except (ValueError, TypeError) as e:
memory_info.append(f"Timestamp: {created_at}")
memory_info.extend([
f"Content: {memory['content']}",
f"Hash: {memory['content_hash']}",
f"Tags: {', '.join(memory.get('tags', []))}"
])
memory_type = memory.get("memory_type")
if memory_type:
memory_info.append(f"Type: {memory_type}")
memory_info.append("---")
formatted_results.append("\n".join(memory_info))
return [types.TextContent(
type="text",
text="Found the following memories:\n\n" + "\n".join(formatted_results)
)]
except Exception as e:
logger.error(f"Error searching by tags: {str(e)}\n{traceback.format_exc()}")
return [types.TextContent(type="text", text=f"Error searching by tags: {str(e)}")]
async def handle_delete_memory(self, arguments: dict) -> List[types.TextContent]:
content_hash = arguments.get("content_hash")
try:
# Initialize storage lazily when needed (also initializes memory_service)
await self._ensure_storage_initialized()
# Call shared MemoryService business logic
result = await self.memory_service.delete_memory(content_hash)
return [types.TextContent(type="text", text=result["message"])]
except Exception as e:
logger.error(f"Error deleting memory: {str(e)}\n{traceback.format_exc()}")
return [types.TextContent(type="text", text=f"Error deleting memory: {str(e)}")]
async def handle_delete_by_tag(self, arguments: dict) -> List[types.TextContent]:
"""Handler for deleting memories by tags."""
from .services.memory_service import normalize_tags
tags = arguments.get("tags", [])
if not tags:
return [types.TextContent(type="text", text="Error: Tags array is required")]
# Normalize tags (handles comma-separated strings and arrays)
tags = normalize_tags(tags)
try:
# Initialize storage lazily when needed
storage = await self._ensure_storage_initialized()
count, message = await storage.delete_by_tag(tags)
return [types.TextContent(type="text", text=message)]
except Exception as e:
logger.error(f"Error deleting by tag: {str(e)}\n{traceback.format_exc()}")
return [types.TextContent(type="text", text=f"Error deleting by tag: {str(e)}")]
async def handle_delete_by_tags(self, arguments: dict) -> List[types.TextContent]:
"""Handler for explicit multiple tag deletion with progress tracking."""
from .services.memory_service import normalize_tags
tags = normalize_tags(arguments.get("tags", []))
if not tags:
return [types.TextContent(type="text", text="Error: Tags array is required")]
try:
# Initialize storage lazily when needed
storage = await self._ensure_storage_initialized()
# Generate operation ID for progress tracking
import uuid
operation_id = f"delete_by_tags_{uuid.uuid4().hex[:8]}"
# Send initial progress notification
await self.send_progress_notification(operation_id, 0, f"Starting deletion of memories with tags: {', '.join(tags)}")
# Execute deletion with progress updates
await self.send_progress_notification(operation_id, 25, "Searching for memories to delete...")
# If storage supports progress callbacks, use them
if hasattr(storage, 'delete_by_tags_with_progress'):
count, message = await storage.delete_by_tags_with_progress(
tags,
progress_callback=lambda p, msg: asyncio.create_task(
self.send_progress_notification(operation_id, 25 + (p * 0.7), msg)
)
)
else:
await self.send_progress_notification(operation_id, 50, "Deleting memories...")
count, message = await storage.delete_by_tags(tags)
await self.send_progress_notification(operation_id, 90, f"Deleted {count} memories")
# Complete the operation
await self.send_progress_notification(operation_id, 100, f"Deletion completed: {message}")
return [types.TextContent(type="text", text=f"{message} (Operation ID: {operation_id})")]
except Exception as e:
logger.error(f"Error deleting by tags: {str(e)}\n{traceback.format_exc()}")
return [types.TextContent(type="text", text=f"Error deleting by tags: {str(e)}")]
async def handle_delete_by_all_tags(self, arguments: dict) -> List[types.TextContent]:
"""Handler for deleting memories that contain ALL specified tags."""
from .services.memory_service import normalize_tags
tags = normalize_tags(arguments.get("tags", []))
if not tags:
return [types.TextContent(type="text", text="Error: Tags array is required")]
try:
# Initialize storage lazily when needed
storage = await self._ensure_storage_initialized()
count, message = await storage.delete_by_all_tags(tags)
return [types.TextContent(type="text", text=message)]
except Exception as e:
logger.error(f"Error deleting by all tags: {str(e)}\n{traceback.format_exc()}")
return [types.TextContent(type="text", text=f"Error deleting by all tags: {str(e)}")]
async def handle_cleanup_duplicates(self, arguments: dict) -> List[types.TextContent]:
try:
# Initialize storage lazily when needed
storage = await self._ensure_storage_initialized()
count, message = await storage.cleanup_duplicates()
return [types.TextContent(type="text", text=message)]
except Exception as e:
logger.error(f"Error cleaning up duplicates: {str(e)}\n{traceback.format_exc()}")
return [types.TextContent(type="text", text=f"Error cleaning up duplicates: {str(e)}")]
async def handle_update_memory_metadata(self, arguments: dict) -> List[types.TextContent]:
"""Handle memory metadata update requests."""
try:
from .services.memory_service import normalize_tags
content_hash = arguments.get("content_hash")
updates = arguments.get("updates")
preserve_timestamps = arguments.get("preserve_timestamps", True)
if not content_hash:
return [types.TextContent(type="text", text="Error: content_hash is required")]
if not updates:
return [types.TextContent(type="text", text="Error: updates dictionary is required")]
if not isinstance(updates, dict):
return [types.TextContent(type="text", text="Error: updates must be a dictionary")]
# Normalize tags if present in updates
if "tags" in updates:
updates["tags"] = normalize_tags(updates["tags"])
# Initialize storage lazily when needed
storage = await self._ensure_storage_initialized()
# Call the storage method
success, message = await storage.update_memory_metadata(
content_hash=content_hash,
updates=updates,
preserve_timestamps=preserve_timestamps
)
if success:
logger.info(f"Successfully updated metadata for memory {content_hash}")
return [types.TextContent(
type="text",
text=f"Successfully updated memory metadata. {message}"
)]
else:
logger.warning(f"Failed to update metadata for memory {content_hash}: {message}")
return [types.TextContent(type="text", text=f"Failed to update memory metadata: {message}")]
except Exception as e:
error_msg = f"Error updating memory metadata: {str(e)}"
logger.error(f"{error_msg}\n{traceback.format_exc()}")
return [types.TextContent(type="text", text=error_msg)]
# Consolidation tool handlers
async def handle_consolidate_memories(self, arguments: dict) -> List[types.TextContent]:
"""Handle memory consolidation requests."""
if not CONSOLIDATION_ENABLED or not self.consolidator:
return [types.TextContent(type="text", text="Error: Consolidation system not available")]
try:
time_horizon = arguments.get("time_horizon")
if not time_horizon:
return [types.TextContent(type="text", text="Error: time_horizon is required")]
if time_horizon not in ["daily", "weekly", "monthly", "quarterly", "yearly"]:
return [types.TextContent(type="text", text="Error: Invalid time_horizon. Must be one of: daily, weekly, monthly, quarterly, yearly")]
logger.info(f"Starting {time_horizon} consolidation")
# Run consolidation
report = await self.consolidator.consolidate(time_horizon)
# Format response
result = f"""Consolidation completed successfully!
Time Horizon: {report.time_horizon}
Duration: {(report.end_time - report.start_time).total_seconds():.2f} seconds
Memories Processed: {report.memories_processed}
Associations Discovered: {report.associations_discovered}
Clusters Created: {report.clusters_created}
Memories Compressed: {report.memories_compressed}
Memories Archived: {report.memories_archived}"""
if report.errors:
result += f"\n\nWarnings/Errors:\n" + "\n".join(f"- {error}" for error in report.errors)
return [types.TextContent(type="text", text=result)]
except Exception as e:
error_msg = f"Error during consolidation: {str(e)}"
logger.error(f"{error_msg}\n{traceback.format_exc()}")
return [types.TextContent(type="text", text=error_msg)]
async def handle_consolidation_status(self, arguments: dict) -> List[types.TextContent]:
"""Handle consolidation status requests."""
if not CONSOLIDATION_ENABLED or not self.consolidator:
return [types.TextContent(type="text", text="Consolidation system: DISABLED")]
try:
# Get health check from consolidator
health = await self.consolidator.health_check()
# Format status report
status_lines = [
f"Consolidation System Status: {health['status'].upper()}",
f"Last Updated: {health['timestamp']}",
"",
"Component Health:"
]
for component, component_health in health['components'].items():
status = component_health['status']
status_lines.append(f" {component}: {status.upper()}")
if status == 'unhealthy' and 'error' in component_health:
status_lines.append(f" Error: {component_health['error']}")
status_lines.extend([
"",
"Statistics:",
f" Total consolidation runs: {health['statistics']['total_runs']}",
f" Successful runs: {health['statistics']['successful_runs']}",
f" Total memories processed: {health['statistics']['total_memories_processed']}",
f" Total associations created: {health['statistics']['total_associations_created']}",
f" Total clusters created: {health['statistics']['total_clusters_created']}",
f" Total memories compressed: {health['statistics']['total_memories_compressed']}",
f" Total memories archived: {health['statistics']['total_memories_archived']}"
])
if health['last_consolidation_times']:
status_lines.extend([
"",
"Last Consolidation Times:"
])
for horizon, timestamp in health['last_consolidation_times'].items():
status_lines.append(f" {horizon}: {timestamp}")
return [types.TextContent(type="text", text="\n".join(status_lines))]
except Exception as e:
error_msg = f"Error getting consolidation status: {str(e)}"
logger.error(f"{error_msg}\n{traceback.format_exc()}")
return [types.TextContent(type="text", text=error_msg)]
async def handle_consolidation_recommendations(self, arguments: dict) -> List[types.TextContent]:
"""Handle consolidation recommendation requests."""
if not CONSOLIDATION_ENABLED or not self.consolidator:
return [types.TextContent(type="text", text="Error: Consolidation system not available")]
try:
time_horizon = arguments.get("time_horizon")
if not time_horizon:
return [types.TextContent(type="text", text="Error: time_horizon is required")]
if time_horizon not in ["daily", "weekly", "monthly", "quarterly", "yearly"]:
return [types.TextContent(type="text", text="Error: Invalid time_horizon")]
# Get recommendations
recommendations = await self.consolidator.get_consolidation_recommendations(time_horizon)
# Format response
lines = [
f"Consolidation Recommendations for {time_horizon} horizon:",
"",
f"Recommendation: {recommendations['recommendation'].upper()}",
f"Memory Count: {recommendations['memory_count']}",
]
if 'reasons' in recommendations:
lines.extend([
"",
"Reasons:"
])
for reason in recommendations['reasons']:
lines.append(f" • {reason}")
if 'memory_types' in recommendations:
lines.extend([
"",
"Memory Types:"
])
for mem_type, count in recommendations['memory_types'].items():
lines.append(f" {mem_type}: {count}")
if 'total_size_bytes' in recommendations:
size_mb = recommendations['total_size_bytes'] / (1024 * 1024)
lines.append(f"\nTotal Size: {size_mb:.2f} MB")
if 'old_memory_percentage' in recommendations:
lines.append(f"Old Memory Percentage: {recommendations['old_memory_percentage']:.1f}%")
if 'estimated_duration_seconds' in recommendations:
lines.append(f"Estimated Duration: {recommendations['estimated_duration_seconds']:.1f} seconds")
return [types.TextContent(type="text", text="\n".join(lines))]
except Exception as e:
error_msg = f"Error getting consolidation recommendations: {str(e)}"
logger.error(f"{error_msg}\n{traceback.format_exc()}")
return [types.TextContent(type="text", text=error_msg)]
async def handle_scheduler_status(self, arguments: dict) -> List[types.TextContent]:
"""Handle scheduler status requests."""
if not CONSOLIDATION_ENABLED or not self.consolidation_scheduler:
return [types.TextContent(type="text", text="Consolidation scheduler: DISABLED")]
try:
# Get scheduler status
status = await self.consolidation_scheduler.get_scheduler_status()
if not status['enabled']:
return [types.TextContent(type="text", text=f"Scheduler: DISABLED\nReason: {status.get('reason', 'Unknown')}")]
# Format status report
lines = [
f"Consolidation Scheduler Status: {'RUNNING' if status['running'] else 'STOPPED'}",
"",
"Scheduled Jobs:"
]
for job in status['jobs']:
next_run = job['next_run_time'] or 'Not scheduled'
lines.append(f" {job['name']}: {next_run}")
lines.extend([
"",
"Execution Statistics:",
f" Total jobs executed: {status['execution_stats']['total_jobs']}",
f" Successful jobs: {status['execution_stats']['successful_jobs']}",
f" Failed jobs: {status['execution_stats']['failed_jobs']}"
])
if status['last_execution_times']:
lines.extend([
"",
"Last Execution Times:"
])
for horizon, timestamp in status['last_execution_times'].items():
lines.append(f" {horizon}: {timestamp}")
if status['recent_jobs']:
lines.extend([
"",
"Recent Jobs:"
])
for job in status['recent_jobs'][-5:]: # Show last 5 jobs
duration = (job['end_time'] - job['start_time']).total_seconds()
lines.append(f" {job['time_horizon']} ({job['status']}): {duration:.2f}s")
return [types.TextContent(type="text", text="\n".join(lines))]
except Exception as e:
error_msg = f"Error getting scheduler status: {str(e)}"
logger.error(f"{error_msg}\n{traceback.format_exc()}")
return [types.TextContent(type="text", text=error_msg)]
async def handle_trigger_consolidation(self, arguments: dict) -> List[types.TextContent]:
"""Handle manual consolidation trigger requests."""
if not CONSOLIDATION_ENABLED or not self.consolidation_scheduler:
return [types.TextContent(type="text", text="Error: Consolidation scheduler not available")]
try:
time_horizon = arguments.get("time_horizon")
immediate = arguments.get("immediate", True)
if not time_horizon:
return [types.TextContent(type="text", text="Error: time_horizon is required")]
if time_horizon not in ["daily", "weekly", "monthly", "quarterly", "yearly"]:
return [types.TextContent(type="text", text="Error: Invalid time_horizon")]
# Trigger consolidation
success = await self.consolidation_scheduler.trigger_consolidation(time_horizon, immediate)
if success:
action = "triggered immediately" if immediate else "scheduled for later"
return [types.TextContent(type="text", text=f"Successfully {action} {time_horizon} consolidation")]
else:
return [types.TextContent(type="text", text=f"Failed to trigger {time_horizon} consolidation")]
except Exception as e:
error_msg = f"Error triggering consolidation: {str(e)}"
logger.error(f"{error_msg}\n{traceback.format_exc()}")
return [types.TextContent(type="text", text=error_msg)]
async def handle_pause_consolidation(self, arguments: dict) -> List[types.TextContent]:
"""Handle consolidation pause requests."""
if not CONSOLIDATION_ENABLED or not self.consolidation_scheduler:
return [types.TextContent(type="text", text="Error: Consolidation scheduler not available")]
try:
time_horizon = arguments.get("time_horizon")
# Pause consolidation
success = await self.consolidation_scheduler.pause_consolidation(time_horizon)
if success:
target = time_horizon or "all"
return [types.TextContent(type="text", text=f"Successfully paused {target} consolidation jobs")]
else:
return [types.TextContent(type="text", text="Failed to pause consolidation jobs")]
except Exception as e:
error_msg = f"Error pausing consolidation: {str(e)}"
logger.error(f"{error_msg}\n{traceback.format_exc()}")
return [types.TextContent(type="text", text=error_msg)]
async def handle_resume_consolidation(self, arguments: dict) -> List[types.TextContent]:
"""Handle consolidation resume requests."""
if not CONSOLIDATION_ENABLED or not self.consolidation_scheduler:
return [types.TextContent(type="text", text="Error: Consolidation scheduler not available")]
try:
time_horizon = arguments.get("time_horizon")
# Resume consolidation
success = await self.consolidation_scheduler.resume_consolidation(time_horizon)
if success:
target = time_horizon or "all"
return [types.TextContent(type="text", text=f"Successfully resumed {target} consolidation jobs")]
else:
return [types.TextContent(type="text", text="Failed to resume consolidation jobs")]
except Exception as e:
error_msg = f"Error resuming consolidation: {str(e)}"
logger.error(f"{error_msg}\n{traceback.format_exc()}")
return [types.TextContent(type="text", text=error_msg)]
async def handle_debug_retrieve(self, arguments: dict) -> List[types.TextContent]:
query = arguments.get("query")
n_results = arguments.get("n_results", 5)
similarity_threshold = arguments.get("similarity_threshold", 0.0)
if not query:
return [types.TextContent(type="text", text="Error: Query is required")]
try:
# Initialize storage lazily when needed
storage = await self._ensure_storage_initialized()
from .utils.debug import debug_retrieve_memory
results = await debug_retrieve_memory(
storage,
query,
n_results,
similarity_threshold
)
if not results:
return [types.TextContent(type="text", text="No matching memories found")]
formatted_results = []
for i, result in enumerate(results):
memory_info = [
f"Memory {i+1}:",
f"Content: {result.memory.content}",
f"Hash: {result.memory.content_hash}",
f"Similarity Score: {result.relevance_score:.4f}"
]
# Add debug info if available
if result.debug_info:
if 'raw_distance' in result.debug_info:
memory_info.append(f"Raw Distance: {result.debug_info['raw_distance']:.4f}")
if 'backend' in result.debug_info:
memory_info.append(f"Backend: {result.debug_info['backend']}")
if 'query' in result.debug_info:
memory_info.append(f"Query: {result.debug_info['query']}")
if 'similarity_threshold' in result.debug_info:
memory_info.append(f"Threshold: {result.debug_info['similarity_threshold']:.2f}")
if result.memory.tags:
memory_info.append(f"Tags: {', '.join(result.memory.tags)}")
memory_info.append("---")
formatted_results.append("\n".join(memory_info))
return [types.TextContent(
type="text",
text="Found the following memories:\n\n" + "\n".join(formatted_results)
)]
except Exception as e:
return [types.TextContent(type="text", text=f"Error in debug retrieve: {str(e)}")]
async def handle_exact_match_retrieve(self, arguments: dict) -> List[types.TextContent]:
content = arguments.get("content")
if not content:
return [types.TextContent(type="text", text="Error: Content is required")]
try:
# Initialize storage lazily when needed
storage = await self._ensure_storage_initialized()
from .utils.debug import exact_match_retrieve
memories = await exact_match_retrieve(storage, content)
if not memories:
return [types.TextContent(type="text", text="No exact matches found")]
formatted_results = []
for i, memory in enumerate(memories):
memory_info = [
f"Memory {i+1}:",
f"Content: {memory.content}",
f"Hash: {memory.content_hash}"
]
if memory.tags:
memory_info.append(f"Tags: {', '.join(memory.tags)}")
memory_info.append("---")
formatted_results.append("\n".join(memory_info))
return [types.TextContent(
type="text",
text="Found the following exact matches:\n\n" + "\n".join(formatted_results)
)]
except Exception as e:
return [types.TextContent(type="text", text=f"Error in exact match retrieve: {str(e)}")]
async def handle_get_raw_embedding(self, arguments: dict) -> List[types.TextContent]:
content = arguments.get("content")
if not content:
return [types.TextContent(type="text", text="Error: Content is required")]
try:
# Initialize storage lazily when needed
storage = await self._ensure_storage_initialized()
from .utils.debug import get_raw_embedding
result = await asyncio.to_thread(get_raw_embedding, storage, content)
if result["status"] == "success":
embedding = result["embedding"]
dimension = result["dimension"]
# Show first 10 and last 10 values for readability
if len(embedding) > 20:
embedding_str = f"[{', '.join(f'{x:.6f}' for x in embedding[:10])}, ..., {', '.join(f'{x:.6f}' for x in embedding[-10:])}]"
else:
embedding_str = f"[{', '.join(f'{x:.6f}' for x in embedding)}]"
return [types.TextContent(
type="text",
text=f"Embedding generated successfully:\n"
f"Dimension: {dimension}\n"
f"Vector: {embedding_str}"
)]
else:
return [types.TextContent(
type="text",
text=f"Failed to generate embedding: {result['error']}"
)]
except Exception as e:
return [types.TextContent(type="text", text=f"Error getting raw embedding: {str(e)}")]
async def handle_recall_memory(self, arguments: dict) -> List[types.TextContent]:
"""
Handle memory recall requests with natural language time expressions.
This handler parses natural language time expressions from the query,
extracts time ranges, and combines them with optional semantic search.
"""
query = arguments.get("query", "")
n_results = arguments.get("n_results", 5)
if not query:
return [types.TextContent(type="text", text="Error: Query is required")]
try:
# Initialize storage lazily when needed
storage = await self._ensure_storage_initialized()
# Parse natural language time expressions
cleaned_query, (start_timestamp, end_timestamp) = extract_time_expression(query)
# Log the parsed timestamps and clean query
logger.info(f"Original query: {query}")
logger.info(f"Cleaned query for semantic search: {cleaned_query}")
logger.info(f"Parsed time range: {start_timestamp} to {end_timestamp}")
# Log more detailed timestamp information for debugging
if start_timestamp is not None:
start_dt = datetime.fromtimestamp(start_timestamp)
logger.info(f"Start timestamp: {start_timestamp} ({start_dt.strftime('%Y-%m-%d %H:%M:%S')})")
if end_timestamp is not None:
end_dt = datetime.fromtimestamp(end_timestamp)
logger.info(f"End timestamp: {end_timestamp} ({end_dt.strftime('%Y-%m-%d %H:%M:%S')})")
if start_timestamp is None and end_timestamp is None:
# No time expression found, try direct parsing
logger.info("No time expression found in query, trying direct parsing")
start_timestamp, end_timestamp = parse_time_expression(query)
logger.info(f"Direct parse result: {start_timestamp} to {end_timestamp}")
# Format human-readable time range for response
time_range_str = ""
if start_timestamp is not None and end_timestamp is not None:
start_dt = datetime.fromtimestamp(start_timestamp)
end_dt = datetime.fromtimestamp(end_timestamp)
time_range_str = f" from {start_dt.strftime('%Y-%m-%d %H:%M')} to {end_dt.strftime('%Y-%m-%d %H:%M')}"
# Retrieve memories with timestamp filter and optional semantic search
# If cleaned_query is empty or just whitespace after removing time expressions,
# we should perform time-based retrieval only
semantic_query = cleaned_query.strip() if cleaned_query.strip() else None
# Use the enhanced recall method that combines semantic search with time filtering,
# or just time filtering if no semantic query
results = await storage.recall(
query=semantic_query,
n_results=n_results,
start_timestamp=start_timestamp,
end_timestamp=end_timestamp
)
if not results:
no_results_msg = f"No memories found{time_range_str}"
return [types.TextContent(type="text", text=no_results_msg)]
# Format results
formatted_results = []
for i, result in enumerate(results):
memory_dt = result.memory.timestamp
memory_info = [
f"Memory {i+1}:",
]
# Add timestamp if available
if memory_dt:
memory_info.append(f"Timestamp: {memory_dt.strftime('%Y-%m-%d %H:%M:%S')}")
# Add other memory information
memory_info.extend([
f"Content: {result.memory.content}",
f"Hash: {result.memory.content_hash}"
])
# Add relevance score if available (may not be for time-only queries)
if hasattr(result, 'relevance_score') and result.relevance_score is not None:
memory_info.append(f"Relevance Score: {result.relevance_score:.2f}")
# Add tags if available
if result.memory.tags:
memory_info.append(f"Tags: {', '.join(result.memory.tags)}")
memory_info.append("---")
formatted_results.append("\n".join(memory_info))
# Include time range in response if available
found_msg = f"Found {len(results)} memories{time_range_str}:"
return [types.TextContent(
type="text",
text=f"{found_msg}\n\n" + "\n".join(formatted_results)
)]
except Exception as e:
logger.error(f"Error in recall_memory: {str(e)}\n{traceback.format_exc()}")
return [types.TextContent(type="text", text=f"Error recalling memories: {str(e)}")]
async def handle_check_database_health(self, arguments: dict) -> List[types.TextContent]:
"""Handle database health check requests with performance metrics."""
logger.info("=== EXECUTING CHECK_DATABASE_HEALTH ===")
try:
# Initialize storage lazily when needed
try:
storage = await self._ensure_storage_initialized()
except Exception as init_error:
# Storage initialization failed
result = {
"validation": {
"status": "unhealthy",
"message": f"Storage initialization failed: {str(init_error)}"
},
"statistics": {
"status": "error",
"error": "Cannot get statistics - storage not initialized"
},
"performance": {
"storage": {},
"server": {
"average_query_time_ms": self.get_average_query_time(),
"total_queries": len(self.query_times)
}
}
}
logger.error(f"Storage initialization failed during health check: {str(init_error)}")
return [types.TextContent(
type="text",
text=f"Database Health Check Results:\n{json.dumps(result, indent=2)}"
)]
# Skip db_utils completely for health check - implement directly here
# Get storage type for backend-specific handling
storage_type = storage.__class__.__name__
# Direct health check implementation based on storage type
is_valid = False
message = ""
stats = {}
if storage_type == "SqliteVecMemoryStorage":
# Direct SQLite-vec validation
if not hasattr(storage, 'conn') or storage.conn is None:
is_valid = False
message = "SQLite database connection is not initialized"
else:
try:
# Check for required tables
cursor = storage.conn.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='memories'")
if not cursor.fetchone():
is_valid = False
message = "SQLite database is missing required tables"
else:
# Count memories
cursor = storage.conn.execute('SELECT COUNT(*) FROM memories')
memory_count = cursor.fetchone()[0]
# Check if embedding tables exist
cursor = storage.conn.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='memory_embeddings'")
has_embeddings = cursor.fetchone() is not None
# Check embedding model
has_model = hasattr(storage, 'embedding_model') and storage.embedding_model is not None
# Collect stats
stats = {
"status": "healthy",
"backend": "sqlite-vec",
"total_memories": memory_count,
"has_embedding_tables": has_embeddings,
"has_embedding_model": has_model,
"embedding_model": storage.embedding_model_name if hasattr(storage, 'embedding_model_name') else "none"
}
# Get database file size
db_path = storage.db_path if hasattr(storage, 'db_path') else None
if db_path and os.path.exists(db_path):
file_size = os.path.getsize(db_path)
stats["database_size_bytes"] = file_size
stats["database_size_mb"] = round(file_size / (1024 * 1024), 2)
is_valid = True
message = "SQLite-vec database validation successful"
except Exception as e:
is_valid = False
message = f"SQLite database validation error: {str(e)}"
stats = {
"status": "error",
"error": str(e),
"backend": "sqlite-vec"
}
elif storage_type == "CloudflareStorage":
# Cloudflare storage validation
try:
# Check if storage is properly initialized
if not hasattr(storage, 'client') or storage.client is None:
is_valid = False
message = "Cloudflare storage client is not initialized"
stats = {
"status": "error",
"error": "Cloudflare storage client is not initialized",
"backend": "cloudflare"
}
else:
# Get storage stats
storage_stats = await storage.get_stats()
# Collect basic health info
stats = {
"status": "healthy",
"backend": "cloudflare",
"total_memories": storage_stats.get("total_memories", 0),
"vectorize_index": storage.vectorize_index,
"d1_database_id": storage.d1_database_id,
"r2_bucket": storage.r2_bucket,
"embedding_model": storage.embedding_model
}
# Add additional stats if available
stats.update(storage_stats)
is_valid = True
message = "Cloudflare storage validation successful"
except Exception as e:
is_valid = False
message = f"Cloudflare storage validation error: {str(e)}"
stats = {
"status": "error",
"error": str(e),
"backend": "cloudflare"
}
elif storage_type == "HybridMemoryStorage":
# Hybrid storage validation (SQLite-vec primary + Cloudflare secondary)
try:
if not hasattr(storage, 'primary') or storage.primary is None:
is_valid = False
message = "Hybrid storage primary backend is not initialized"
stats = {
"status": "error",
"error": "Hybrid storage primary backend is not initialized",
"backend": "hybrid"
}
else:
primary_storage = storage.primary
# Validate primary storage (SQLite-vec)
if not hasattr(primary_storage, 'conn') or primary_storage.conn is None:
is_valid = False
message = "Hybrid storage: SQLite connection is not initialized"
stats = {
"status": "error",
"error": "SQLite connection is not initialized",
"backend": "hybrid"
}
else:
# Check for required tables
cursor = primary_storage.conn.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='memories'")
if not cursor.fetchone():
is_valid = False
message = "Hybrid storage: SQLite database is missing required tables"
stats = {
"status": "error",
"error": "SQLite database is missing required tables",
"backend": "hybrid"
}
else:
# Count memories
cursor = primary_storage.conn.execute('SELECT COUNT(*) FROM memories')
memory_count = cursor.fetchone()[0]
# Check if embedding tables exist
cursor = primary_storage.conn.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='memory_embeddings'")
has_embeddings = cursor.fetchone() is not None
# Check secondary (Cloudflare) status
cloudflare_status = "not_configured"
if hasattr(storage, 'secondary') and storage.secondary:
sync_service = getattr(storage, 'sync_service', None)
if sync_service and getattr(sync_service, 'is_running', False):
cloudflare_status = "syncing"
else:
cloudflare_status = "configured"
# Collect stats
stats = {
"status": "healthy",
"backend": "hybrid",
"total_memories": memory_count,
"has_embeddings": has_embeddings,
"database_path": getattr(primary_storage, 'db_path', SQLITE_VEC_PATH),
"cloudflare_sync": cloudflare_status
}
is_valid = True
message = f"Hybrid storage validation successful ({memory_count} memories, Cloudflare: {cloudflare_status})"
except Exception as e:
is_valid = False
message = f"Hybrid storage validation error: {str(e)}"
stats = {
"status": "error",
"error": str(e),
"backend": "hybrid"
}
else:
is_valid = False
message = f"Unknown storage type: {storage_type}"
stats = {
"status": "error",
"error": f"Unknown storage type: {storage_type}",
"backend": "unknown"
}
# Get performance stats from optimized storage
performance_stats = {}
if hasattr(storage, 'get_performance_stats') and callable(storage.get_performance_stats):
try:
performance_stats = storage.get_performance_stats()
except Exception as perf_error:
logger.warning(f"Could not get performance stats: {str(perf_error)}")
performance_stats = {"error": str(perf_error)}
# Get server-level performance stats
server_stats = {
"average_query_time_ms": self.get_average_query_time(),
"total_queries": len(self.query_times)
}
# Add storage type for debugging
server_stats["storage_type"] = storage_type
# Add storage initialization status for debugging
if hasattr(storage, 'get_initialization_status') and callable(storage.get_initialization_status):
try:
server_stats["storage_initialization"] = storage.get_initialization_status()
except Exception:
pass
# Combine results with performance data
result = {
"version": __version__,
"validation": {
"status": "healthy" if is_valid else "unhealthy",
"message": message
},
"statistics": stats,
"performance": {
"storage": performance_stats,
"server": server_stats
}
}
logger.info(f"Database health result with performance data: {result}")
return [types.TextContent(
type="text",
text=f"Database Health Check Results:\n{json.dumps(result, indent=2)}"
)]
except Exception as e:
logger.error(f"Error in check_database_health: {str(e)}")
logger.error(traceback.format_exc())
return [types.TextContent(
type="text",
text=f"Error checking database health: {str(e)}"
)]
async def handle_get_cache_stats(self, arguments: dict) -> List[types.TextContent]:
"""
Get MCP server global cache statistics for performance monitoring.
Returns detailed metrics about storage and memory service caching,
including hit rates, initialization times, and cache sizes.
"""
global _CACHE_STATS, _STORAGE_CACHE, _MEMORY_SERVICE_CACHE
try:
# Import shared stats calculation utility
from .utils.cache_manager import CacheStats, calculate_cache_stats_dict
# Convert global dict to CacheStats dataclass
stats = CacheStats(
total_calls=_CACHE_STATS["total_calls"],
storage_hits=_CACHE_STATS["storage_hits"],
storage_misses=_CACHE_STATS["storage_misses"],
service_hits=_CACHE_STATS["service_hits"],
service_misses=_CACHE_STATS["service_misses"],
initialization_times=_CACHE_STATS["initialization_times"]
)
# Calculate statistics using shared utility
cache_sizes = (len(_STORAGE_CACHE), len(_MEMORY_SERVICE_CACHE))
result = calculate_cache_stats_dict(stats, cache_sizes)
# Add server-specific details
result["storage_cache"]["keys"] = list(_STORAGE_CACHE.keys())
result["backend_info"] = {
"storage_backend": STORAGE_BACKEND,
"sqlite_path": SQLITE_VEC_PATH
}
logger.info(f"Cache stats retrieved: {result['message']}")
# Return JSON string for easy parsing by clients
return [types.TextContent(
type="text",
text=json.dumps(result, indent=2)
)]
except Exception as e:
logger.error(f"Error in get_cache_stats: {str(e)}")
logger.error(traceback.format_exc())
return [types.TextContent(
type="text",
text=f"Error getting cache stats: {str(e)}"
)]
async def handle_recall_by_timeframe(self, arguments: dict) -> List[types.TextContent]:
"""Handle recall by timeframe requests."""
from datetime import datetime
try:
# Initialize storage lazily when needed
storage = await self._ensure_storage_initialized()
start_date = datetime.fromisoformat(arguments["start_date"]).date()
end_date = datetime.fromisoformat(arguments.get("end_date", arguments["start_date"])).date()
n_results = arguments.get("n_results", 5)
# Get timestamp range
start_timestamp = datetime(start_date.year, start_date.month, start_date.day).timestamp()
end_timestamp = datetime(end_date.year, end_date.month, end_date.day, 23, 59, 59).timestamp()
# Log the timestamp values for debugging
logger.info(f"Recall by timeframe: {start_date} to {end_date}")
logger.info(f"Start timestamp: {start_timestamp} ({datetime.fromtimestamp(start_timestamp).strftime('%Y-%m-%d %H:%M:%S')})")
logger.info(f"End timestamp: {end_timestamp} ({datetime.fromtimestamp(end_timestamp).strftime('%Y-%m-%d %H:%M:%S')})")
# Retrieve memories with proper parameters - query is None because this is pure time-based filtering
results = await storage.recall(
query=None,
n_results=n_results,
start_timestamp=start_timestamp,
end_timestamp=end_timestamp
)
if not results:
return [types.TextContent(type="text", text=f"No memories found from {start_date} to {end_date}")]
formatted_results = []
for i, result in enumerate(results):
memory_timestamp = result.memory.timestamp
memory_info = [
f"Memory {i+1}:",
]
# Add timestamp if available
if memory_timestamp:
memory_info.append(f"Timestamp: {memory_timestamp.strftime('%Y-%m-%d %H:%M:%S')}")
memory_info.extend([
f"Content: {result.memory.content}",
f"Hash: {result.memory.content_hash}"
])
if result.memory.tags:
memory_info.append(f"Tags: {', '.join(result.memory.tags)}")
memory_info.append("---")
formatted_results.append("\n".join(memory_info))
return [types.TextContent(
type="text",
text=f"Found {len(results)} memories from {start_date} to {end_date}:\n\n" + "\n".join(formatted_results)
)]
except Exception as e:
logger.error(f"Error in recall_by_timeframe: {str(e)}\n{traceback.format_exc()}")
return [types.TextContent(
type="text",
text=f"Error recalling memories: {str(e)}"
)]
async def handle_delete_by_timeframe(self, arguments: dict) -> List[types.TextContent]:
"""Handle delete by timeframe requests."""
from datetime import datetime
try:
# Initialize storage lazily when needed
storage = await self._ensure_storage_initialized()
start_date = datetime.fromisoformat(arguments["start_date"]).date()
end_date = datetime.fromisoformat(arguments.get("end_date", arguments["start_date"])).date()
tag = arguments.get("tag")
count, message = await storage.delete_by_timeframe(start_date, end_date, tag)
return [types.TextContent(
type="text",
text=f"Deleted {count} memories: {message}"
)]
except Exception as e:
return [types.TextContent(
type="text",
text=f"Error deleting memories: {str(e)}"
)]
async def handle_delete_before_date(self, arguments: dict) -> List[types.TextContent]:
"""Handle delete before date requests."""
from datetime import datetime
try:
# Initialize storage lazily when needed
storage = await self._ensure_storage_initialized()
before_date = datetime.fromisoformat(arguments["before_date"]).date()
tag = arguments.get("tag")
count, message = await storage.delete_before_date(before_date, tag)
return [types.TextContent(
type="text",
text=f"Deleted {count} memories: {message}"
)]
except Exception as e:
return [types.TextContent(
type="text",
text=f"Error deleting memories: {str(e)}"
)]
async def handle_ingest_document(self, arguments: dict) -> List[types.TextContent]:
"""Handle document ingestion requests."""
try:
from pathlib import Path
from .ingestion import get_loader_for_file
from .models.memory import Memory
from .utils import create_memory_from_chunk
import time
# Initialize storage lazily when needed
storage = await self._ensure_storage_initialized()
from .services.memory_service import normalize_tags
file_path = Path(arguments["file_path"])
tags = normalize_tags(arguments.get("tags", []))
chunk_size = arguments.get("chunk_size", 1000)
chunk_overlap = arguments.get("chunk_overlap", 200)
memory_type = arguments.get("memory_type", "document")
logger.info(f"Starting document ingestion: {file_path}")
start_time = time.time()
# Validate file exists and get appropriate document loader
if not file_path.exists():
return [types.TextContent(
type="text",
text=f"Error: File not found: {file_path.resolve()}"
)]
loader = get_loader_for_file(file_path)
if loader is None:
from .ingestion import SUPPORTED_FORMATS
supported_exts = ", ".join(f".{ext}" for ext in SUPPORTED_FORMATS.keys())
return [types.TextContent(
type="text",
text=f"Error: Unsupported file format: {file_path.suffix}. Supported formats: {supported_exts}"
)]
# Configure loader
loader.chunk_size = chunk_size
loader.chunk_overlap = chunk_overlap
chunks_processed = 0
chunks_stored = 0
errors = []
# Extract and store chunks
async for chunk in loader.extract_chunks(file_path):
chunks_processed += 1
try:
# Combine document tags with chunk metadata tags
all_tags = tags.copy()
if chunk.metadata.get('tags'):
# Handle tags from chunk metadata (can be string or list)
chunk_tags = chunk.metadata['tags']
if isinstance(chunk_tags, str):
# Split comma-separated string into list
chunk_tags = [tag.strip() for tag in chunk_tags.split(',') if tag.strip()]
all_tags.extend(chunk_tags)
# Create memory object
memory = Memory(
content=chunk.content,
content_hash=generate_content_hash(chunk.content, chunk.metadata),
tags=list(set(all_tags)), # Remove duplicates
memory_type=memory_type,
metadata=chunk.metadata
)
# Store the memory
success, error = await storage.store(memory)
if success:
chunks_stored += 1
else:
errors.append(f"Chunk {chunk.chunk_index}: {error}")
except Exception as e:
errors.append(f"Chunk {chunk.chunk_index}: {str(e)}")
processing_time = time.time() - start_time
success_rate = (chunks_stored / chunks_processed * 100) if chunks_processed > 0 else 0
# Prepare result message
result_lines = [
f"✅ Document ingestion completed: {file_path.name}",
f"📄 Chunks processed: {chunks_processed}",
f"💾 Chunks stored: {chunks_stored}",
f"⚡ Success rate: {success_rate:.1f}%",
f"⏱️ Processing time: {processing_time:.2f} seconds"
]
if errors:
result_lines.append(f"⚠️ Errors encountered: {len(errors)}")
if len(errors) <= 5: # Show first few errors
result_lines.extend([f" - {error}" for error in errors[:5]])
else:
result_lines.extend([f" - {error}" for error in errors[:3]])
result_lines.append(f" ... and {len(errors) - 3} more errors")
logger.info(f"Document ingestion completed: {chunks_stored}/{chunks_processed} chunks stored")
return [types.TextContent(type="text", text="\n".join(result_lines))]
except Exception as e:
logger.error(f"Error in document ingestion: {str(e)}")
return [types.TextContent(
type="text",
text=f"Error ingesting document: {str(e)}"
)]
async def handle_ingest_directory(self, arguments: dict) -> List[types.TextContent]:
"""Handle directory ingestion requests."""
try:
from pathlib import Path
from .ingestion import get_loader_for_file, is_supported_file
from .models.memory import Memory
from .utils import generate_content_hash
import time
# Initialize storage lazily when needed
storage = await self._ensure_storage_initialized()
from .services.memory_service import normalize_tags
directory_path = Path(arguments["directory_path"])
tags = normalize_tags(arguments.get("tags", []))
recursive = arguments.get("recursive", True)
file_extensions = arguments.get("file_extensions", ["pdf", "txt", "md", "json"])
chunk_size = arguments.get("chunk_size", 1000)
max_files = arguments.get("max_files", 100)
if not directory_path.exists() or not directory_path.is_dir():
return [types.TextContent(
type="text",
text=f"Error: Directory not found: {directory_path}"
)]
logger.info(f"Starting directory ingestion: {directory_path}")
start_time = time.time()
# Find all supported files
pattern = "**/*" if recursive else "*"
all_files = []
for ext in file_extensions:
ext_pattern = f"*.{ext.lstrip('.')}"
if recursive:
files = list(directory_path.rglob(ext_pattern))
else:
files = list(directory_path.glob(ext_pattern))
all_files.extend(files)
# Remove duplicates and filter supported files
unique_files = []
seen = set()
for file_path in all_files:
if file_path not in seen and is_supported_file(file_path):
unique_files.append(file_path)
seen.add(file_path)
# Limit number of files
files_to_process = unique_files[:max_files]
if not files_to_process:
return [types.TextContent(
type="text",
text=f"No supported files found in directory: {directory_path}"
)]
total_chunks_processed = 0
total_chunks_stored = 0
files_processed = 0
files_failed = 0
all_errors = []
# Process each file
for file_path in files_to_process:
try:
logger.info(f"Processing file {files_processed + 1}/{len(files_to_process)}: {file_path.name}")
# Get appropriate document loader
loader = get_loader_for_file(file_path)
if loader is None:
all_errors.append(f"{file_path.name}: Unsupported format")
files_failed += 1
continue
# Configure loader
loader.chunk_size = chunk_size
file_chunks_processed = 0
file_chunks_stored = 0
# Extract and store chunks from this file
async for chunk in loader.extract_chunks(file_path):
file_chunks_processed += 1
total_chunks_processed += 1
# Process and store the chunk
success, error = await _process_and_store_chunk(
chunk,
storage,
file_path.name,
base_tags=tags.copy(),
context_tags={
"source_dir": directory_path.name,
"file_type": file_path.suffix.lstrip('.')
}
)
if success:
file_chunks_stored += 1
total_chunks_stored += 1
else:
all_errors.append(error)
if file_chunks_stored > 0:
files_processed += 1
else:
files_failed += 1
except Exception as e:
files_failed += 1
all_errors.append(f"{file_path.name}: {str(e)}")
processing_time = time.time() - start_time
success_rate = (total_chunks_stored / total_chunks_processed * 100) if total_chunks_processed > 0 else 0
# Prepare result message
result_lines = [
f"✅ Directory ingestion completed: {directory_path.name}",
f"📁 Files processed: {files_processed}/{len(files_to_process)}",
f"📄 Total chunks processed: {total_chunks_processed}",
f"💾 Total chunks stored: {total_chunks_stored}",
f"⚡ Success rate: {success_rate:.1f}%",
f"⏱️ Processing time: {processing_time:.2f} seconds"
]
if files_failed > 0:
result_lines.append(f"❌ Files failed: {files_failed}")
if all_errors:
result_lines.append(f"⚠️ Total errors: {len(all_errors)}")
# Show first few errors
error_limit = 5
for error in all_errors[:error_limit]:
result_lines.append(f" - {error}")
if len(all_errors) > error_limit:
result_lines.append(f" ... and {len(all_errors) - error_limit} more errors")
logger.info(f"Directory ingestion completed: {total_chunks_stored}/{total_chunks_processed} chunks from {files_processed} files")
return [types.TextContent(type="text", text="\n".join(result_lines))]
except Exception as e:
logger.error(f"Error in directory ingestion: {str(e)}")
return [types.TextContent(
type="text",
text=f"Error ingesting directory: {str(e)}"
)]
async def async_main():
# Apply LM Studio compatibility patch before anything else
patch_mcp_for_lm_studio()
# Add Windows-specific timeout handling
add_windows_timeout_handling()
# Run dependency check before starting
run_dependency_check()
# Check if running with UV
check_uv_environment()
# Check for version mismatch (stale venv issue)
check_version_consistency()
# Debug logging is now handled by the CLI layer
# Print system diagnostics only for LM Studio (avoid JSON parsing errors in Claude Desktop)
system_info = get_system_info()
if MCP_CLIENT == 'lm_studio':
print("\n=== MCP Memory Service System Diagnostics ===", file=sys.stdout, flush=True)
print(f"OS: {system_info.os_name} {system_info.architecture}", file=sys.stdout, flush=True)
print(f"Python: {platform.python_version()}", file=sys.stdout, flush=True)
print(f"Hardware Acceleration: {system_info.accelerator}", file=sys.stdout, flush=True)
print(f"Memory: {system_info.memory_gb:.2f} GB", file=sys.stdout, flush=True)
print(f"Optimal Model: {system_info.get_optimal_model()}", file=sys.stdout, flush=True)
print(f"Optimal Batch Size: {system_info.get_optimal_batch_size()}", file=sys.stdout, flush=True)
print(f"Storage Backend: {STORAGE_BACKEND}", file=sys.stdout, flush=True)
print("================================================\n", file=sys.stdout, flush=True)
logger.info(f"Starting MCP Memory Service with storage backend: {STORAGE_BACKEND}")
try:
# Create server instance with hardware-aware configuration
memory_server = MemoryServer()
# Set up async initialization with timeout and retry logic
max_retries = 2
retry_count = 0
init_success = False
while retry_count <= max_retries and not init_success:
if retry_count > 0:
logger.warning(f"Retrying initialization (attempt {retry_count}/{max_retries})...")
init_task = asyncio.create_task(memory_server.initialize())
try:
# 30 second timeout for initialization
init_success = await asyncio.wait_for(init_task, timeout=30.0)
if init_success:
logger.info("Async initialization completed successfully")
else:
logger.warning("Initialization returned failure status")
retry_count += 1
except asyncio.TimeoutError:
logger.warning("Async initialization timed out. Continuing with server startup.")
# Don't cancel the task, let it complete in the background
break
except Exception as init_error:
logger.error(f"Initialization error: {str(init_error)}")
logger.error(traceback.format_exc())
retry_count += 1
if retry_count <= max_retries:
logger.info(f"Waiting 2 seconds before retry...")
await asyncio.sleep(2)
# Check if running in standalone mode (Docker without active client)
standalone_mode = os.environ.get('MCP_STANDALONE_MODE', '').lower() == '1'
running_in_docker = os.path.exists('/.dockerenv') or os.environ.get('DOCKER_CONTAINER', False)
if standalone_mode:
logger.info("Running in standalone mode - keeping server alive without active client")
if MCP_CLIENT == 'lm_studio':
print("MCP Memory Service running in standalone mode", file=sys.stdout, flush=True)
# Keep the server running indefinitely
try:
while True:
await asyncio.sleep(60) # Sleep for 60 seconds at a time
logger.debug("Standalone server heartbeat")
except asyncio.CancelledError:
logger.info("Standalone server cancelled")
raise
else:
# Start the server with stdio
async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
logger.info("Server started and ready to handle requests")
if running_in_docker:
logger.info("Detected Docker environment - ensuring proper stdio handling")
if MCP_CLIENT == 'lm_studio':
print("MCP Memory Service running in Docker container", file=sys.stdout, flush=True)
try:
await memory_server.server.run(
read_stream,
write_stream,
InitializationOptions(
server_name=SERVER_NAME,
server_version=SERVER_VERSION,
# Explicitly specify the protocol version that matches Claude's request
# Use the latest protocol version to ensure compatibility with all clients
protocol_version="2024-11-05",
capabilities=memory_server.server.get_capabilities(
notification_options=NotificationOptions(),
experimental_capabilities={
"hardware_info": {
"architecture": system_info.architecture,
"accelerator": system_info.accelerator,
"memory_gb": system_info.memory_gb,
"cpu_count": system_info.cpu_count
}
},
),
),
)
except asyncio.CancelledError:
logger.info("Server run cancelled")
raise
except BaseException as e:
# Handle ExceptionGroup specially (Python 3.11+)
if type(e).__name__ == 'ExceptionGroup' or 'ExceptionGroup' in str(type(e)):
error_str = str(e)
# Check if this contains the LM Studio cancelled notification error
if 'notifications/cancelled' in error_str or 'ValidationError' in error_str:
logger.info("LM Studio sent a cancelled notification - this is expected behavior")
logger.debug(f"Full error for debugging: {error_str}")
# Don't re-raise - just continue gracefully
else:
logger.error(f"ExceptionGroup in server.run: {str(e)}")
logger.error(traceback.format_exc())
raise
else:
logger.error(f"Error in server.run: {str(e)}")
logger.error(traceback.format_exc())
raise
finally:
logger.info("Server run completed")
except Exception as e:
logger.error(f"Server error: {str(e)}")
logger.error(traceback.format_exc())
print(f"Fatal server error: {str(e)}", file=sys.stderr, flush=True)
raise
def main():
import signal
# Set up signal handlers for graceful shutdown
def signal_handler(signum, frame):
logger.info(f"Received signal {signum}, shutting down gracefully...")
sys.exit(0)
signal.signal(signal.SIGTERM, signal_handler)
signal.signal(signal.SIGINT, signal_handler)
try:
# Check if running in Docker
if os.path.exists('/.dockerenv') or os.environ.get('DOCKER_CONTAINER', False):
logger.info("Running in Docker container")
if MCP_CLIENT == 'lm_studio':
print("MCP Memory Service starting in Docker mode", file=sys.stdout, flush=True)
asyncio.run(async_main())
except KeyboardInterrupt:
logger.info("Shutting down gracefully...")
except Exception as e:
logger.error(f"Fatal error: {str(e)}\n{traceback.format_exc()}")
sys.exit(1)
if __name__ == "__main__":
main()
```