This is page 26 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
--------------------------------------------------------------------------------
/docs/examples/analysis-scripts.js:
--------------------------------------------------------------------------------
```javascript
/**
* Memory Analysis Scripts
*
* A collection of JavaScript functions for analyzing and extracting insights
* from MCP Memory Service data. These scripts demonstrate practical approaches
* to memory data analysis, pattern recognition, and visualization preparation.
*
* Usage: Import individual functions or use as reference for building
* custom analysis pipelines.
*/
// =============================================================================
// TEMPORAL ANALYSIS FUNCTIONS
// =============================================================================
/**
* Analyze memory distribution over time periods
* @param {Array} memories - Array of memory objects with timestamps
* @returns {Object} Distribution data organized by time periods
*/
function analyzeTemporalDistribution(memories) {
const distribution = {
monthly: {},
weekly: {},
daily: {},
hourly: {}
};
memories.forEach(memory => {
const date = new Date(memory.timestamp);
// Monthly distribution
const monthKey = `${date.getFullYear()}-${String(date.getMonth() + 1).padStart(2, '0')}`;
if (!distribution.monthly[monthKey]) {
distribution.monthly[monthKey] = [];
}
distribution.monthly[monthKey].push(memory);
// Weekly distribution (week of year)
const weekKey = `${date.getFullYear()}-W${getWeekNumber(date)}`;
if (!distribution.weekly[weekKey]) {
distribution.weekly[weekKey] = [];
}
distribution.weekly[weekKey].push(memory);
// Daily distribution (day of week)
const dayKey = date.toLocaleDateString('en-US', { weekday: 'long' });
if (!distribution.daily[dayKey]) {
distribution.daily[dayKey] = [];
}
distribution.daily[dayKey].push(memory);
// Hourly distribution
const hourKey = date.getHours();
if (!distribution.hourly[hourKey]) {
distribution.hourly[hourKey] = [];
}
distribution.hourly[hourKey].push(memory);
});
return distribution;
}
/**
* Calculate week number for a given date
* @param {Date} date - Date object
* @returns {number} Week number
*/
function getWeekNumber(date) {
const firstDayOfYear = new Date(date.getFullYear(), 0, 1);
const pastDaysOfYear = (date - firstDayOfYear) / 86400000;
return Math.ceil((pastDaysOfYear + firstDayOfYear.getDay() + 1) / 7);
}
/**
* Prepare temporal data for chart visualization
* @param {Object} distribution - Distribution object from analyzeTemporalDistribution
* @param {string} period - Time period ('monthly', 'weekly', 'daily', 'hourly')
* @returns {Array} Chart-ready data array
*/
function prepareTemporalChartData(distribution, period = 'monthly') {
const data = distribution[period];
const chartData = Object.entries(data)
.map(([key, memories]) => ({
period: formatPeriodLabel(key, period),
count: memories.length,
memories: memories,
key: key
}))
.sort((a, b) => a.key.localeCompare(b.key));
return chartData;
}
/**
* Format period labels for display
* @param {string} key - Period key
* @param {string} period - Period type
* @returns {string} Formatted label
*/
function formatPeriodLabel(key, period) {
switch (period) {
case 'monthly':
const [year, month] = key.split('-');
const monthNames = ['Jan', 'Feb', 'Mar', 'Apr', 'May', 'Jun',
'Jul', 'Aug', 'Sep', 'Oct', 'Nov', 'Dec'];
return `${monthNames[parseInt(month) - 1]} ${year}`;
case 'weekly':
return key; // Already formatted as YYYY-WXX
case 'daily':
return key; // Day names are already formatted
case 'hourly':
const hour = parseInt(key);
return `${hour}:00`;
default:
return key;
}
}
// =============================================================================
// TAG ANALYSIS FUNCTIONS
// =============================================================================
/**
* Analyze tag usage frequency and patterns
* @param {Array} memories - Array of memory objects
* @returns {Object} Tag analysis results
*/
function analyzeTagUsage(memories) {
const tagFrequency = {};
const tagCombinations = {};
const categoryDistribution = {};
memories.forEach(memory => {
const tags = memory.tags || [];
// Tag frequency analysis
tags.forEach(tag => {
tagFrequency[tag] = (tagFrequency[tag] || 0) + 1;
// Categorize tags
const category = categorizeTag(tag);
if (!categoryDistribution[category]) {
categoryDistribution[category] = {};
}
categoryDistribution[category][tag] = (categoryDistribution[category][tag] || 0) + 1;
});
// Tag combination analysis
if (tags.length > 1) {
for (let i = 0; i < tags.length; i++) {
for (let j = i + 1; j < tags.length; j++) {
const combo = [tags[i], tags[j]].sort().join(' + ');
tagCombinations[combo] = (tagCombinations[combo] || 0) + 1;
}
}
}
});
return {
frequency: Object.entries(tagFrequency)
.sort(([,a], [,b]) => b - a),
combinations: Object.entries(tagCombinations)
.sort(([,a], [,b]) => b - a)
.slice(0, 20), // Top 20 combinations
categories: categoryDistribution,
totalTags: Object.keys(tagFrequency).length,
averageTagsPerMemory: memories.reduce((sum, m) => sum + (m.tags?.length || 0), 0) / memories.length
};
}
/**
* Categorize a tag based on common patterns
* @param {string} tag - Tag to categorize
* @returns {string} Category name
*/
function categorizeTag(tag) {
const patterns = {
'projects': /^(mcp-memory-service|memory-dashboard|github-integration|mcp-protocol)/,
'technologies': /^(python|react|typescript|chromadb|git|docker|aws|npm)/,
'activities': /^(testing|debugging|development|documentation|deployment|maintenance)/,
'status': /^(resolved|in-progress|blocked|verified|completed|experimental)/,
'content-types': /^(concept|architecture|tutorial|reference|example|guide)/,
'temporal': /^(january|february|march|april|may|june|q1|q2|2025)/,
'priorities': /^(urgent|high-priority|low-priority|critical)/
};
for (const [category, pattern] of Object.entries(patterns)) {
if (pattern.test(tag)) {
return category;
}
}
return 'other';
}
/**
* Find tag inconsistencies and suggest improvements
* @param {Array} memories - Array of memory objects
* @returns {Object} Consistency analysis results
*/
function analyzeTagConsistency(memories) {
const inconsistencies = [];
const suggestions = [];
const patterns = {};
memories.forEach((memory, index) => {
const content = memory.content || '';
const tags = memory.tags || [];
// Common content patterns that should have corresponding tags
const contentPatterns = {
'test': /\b(test|testing|TEST)\b/i,
'bug': /\b(bug|issue|error|problem)\b/i,
'debug': /\b(debug|debugging|fix|fixed)\b/i,
'documentation': /\b(document|guide|tutorial|readme)\b/i,
'concept': /\b(concept|idea|design|architecture)\b/i,
'implementation': /\b(implement|implementation|develop|development)\b/i
};
Object.entries(contentPatterns).forEach(([expectedTag, pattern]) => {
if (pattern.test(content)) {
const hasRelatedTag = tags.some(tag =>
tag.includes(expectedTag) ||
expectedTag.includes(tag.split('-')[0])
);
if (!hasRelatedTag) {
inconsistencies.push({
memoryIndex: index,
type: 'missing-tag',
expectedTag: expectedTag,
content: content.substring(0, 100) + '...',
currentTags: tags
});
}
}
});
// Check for overly generic tags
const genericTags = ['test', 'memory', 'note', 'temp', 'example'];
const hasGenericOnly = tags.length > 0 &&
tags.every(tag => genericTags.includes(tag));
if (hasGenericOnly) {
suggestions.push({
memoryIndex: index,
type: 'improve-specificity',
suggestion: 'Replace generic tags with specific categories',
currentTags: tags,
content: content.substring(0, 100) + '...'
});
}
});
return {
inconsistencies,
suggestions,
consistencyScore: ((memories.length - inconsistencies.length) / memories.length) * 100,
totalIssues: inconsistencies.length + suggestions.length
};
}
// =============================================================================
// CONTENT ANALYSIS FUNCTIONS
// =============================================================================
/**
* Analyze content patterns and themes
* @param {Array} memories - Array of memory objects
* @returns {Object} Content analysis results
*/
function analyzeContentPatterns(memories) {
const themes = {};
const contentTypes = {};
const wordFrequency = {};
const lengthDistribution = {};
memories.forEach(memory => {
const content = memory.content || '';
const words = extractKeywords(content);
const contentType = detectContentType(content);
// Theme analysis based on keywords
words.forEach(word => {
wordFrequency[word] = (wordFrequency[word] || 0) + 1;
});
// Content type distribution
contentTypes[contentType] = (contentTypes[contentType] || 0) + 1;
// Length distribution
const lengthCategory = categorizeContentLength(content.length);
lengthDistribution[lengthCategory] = (lengthDistribution[lengthCategory] || 0) + 1;
});
// Extract top themes from word frequency
const topWords = Object.entries(wordFrequency)
.sort(([,a], [,b]) => b - a)
.slice(0, 50);
return {
themes: extractThemes(topWords),
contentTypes,
lengthDistribution,
wordFrequency: topWords,
averageLength: memories.reduce((sum, m) => sum + (m.content?.length || 0), 0) / memories.length
};
}
/**
* Extract keywords from content
* @param {string} content - Memory content
* @returns {Array} Array of keywords
*/
function extractKeywords(content) {
const stopWords = new Set([
'the', 'a', 'an', 'and', 'or', 'but', 'in', 'on', 'at', 'to', 'for', 'of', 'with',
'by', 'from', 'up', 'about', 'into', 'through', 'during', 'before', 'after',
'above', 'below', 'between', 'among', 'is', 'are', 'was', 'were', 'be', 'been',
'have', 'has', 'had', 'do', 'does', 'did', 'will', 'would', 'could', 'should',
'may', 'might', 'must', 'can', 'this', 'that', 'these', 'those'
]);
return content
.toLowerCase()
.replace(/[^\w\s-]/g, ' ') // Remove punctuation except hyphens
.split(/\s+/)
.filter(word =>
word.length > 2 &&
!stopWords.has(word) &&
!word.match(/^\d+$/) // Exclude pure numbers
);
}
/**
* Detect content type based on patterns
* @param {string} content - Memory content
* @returns {string} Content type
*/
function detectContentType(content) {
const patterns = {
'code': /```|function\s*\(|class\s+\w+|import\s+\w+/,
'documentation': /^#+\s|README|GUIDE|TUTORIAL/i,
'issue': /issue|bug|error|problem|fix|resolved/i,
'concept': /concept|idea|design|architecture|approach/i,
'test': /test|testing|verify|validation|TEST/i,
'configuration': /config|setup|installation|environment/i,
'analysis': /analysis|report|summary|statistics|metrics/i
};
for (const [type, pattern] of Object.entries(patterns)) {
if (pattern.test(content)) {
return type;
}
}
return 'general';
}
/**
* Categorize content by length
* @param {number} length - Content length in characters
* @returns {string} Length category
*/
function categorizeContentLength(length) {
if (length < 100) return 'very-short';
if (length < 500) return 'short';
if (length < 1500) return 'medium';
if (length < 3000) return 'long';
return 'very-long';
}
/**
* Extract themes from word frequency data
* @param {Array} topWords - Array of [word, frequency] pairs
* @returns {Object} Organized themes
*/
function extractThemes(topWords) {
const themeCategories = {
technology: ['python', 'react', 'typescript', 'chromadb', 'git', 'docker', 'api', 'database'],
development: ['development', 'implementation', 'code', 'programming', 'build', 'deploy'],
testing: ['test', 'testing', 'debug', 'debugging', 'verification', 'quality'],
project: ['project', 'service', 'system', 'application', 'platform', 'tool'],
process: ['process', 'workflow', 'methodology', 'procedure', 'approach', 'strategy']
};
const themes = {};
const wordMap = new Map(topWords);
Object.entries(themeCategories).forEach(([theme, keywords]) => {
themes[theme] = keywords
.filter(keyword => wordMap.has(keyword))
.map(keyword => ({ word: keyword, frequency: wordMap.get(keyword) }))
.sort((a, b) => b.frequency - a.frequency);
});
return themes;
}
// =============================================================================
// QUALITY ANALYSIS FUNCTIONS
// =============================================================================
/**
* Assess overall memory quality and organization
* @param {Array} memories - Array of memory objects
* @returns {Object} Quality assessment results
*/
function assessMemoryQuality(memories) {
const metrics = {
tagging: assessTaggingQuality(memories),
content: assessContentQuality(memories),
organization: assessOrganizationQuality(memories),
searchability: assessSearchabilityQuality(memories)
};
// Calculate overall quality score
const overallScore = Object.values(metrics)
.reduce((sum, metric) => sum + metric.score, 0) / Object.keys(metrics).length;
return {
overallScore: Math.round(overallScore),
metrics,
recommendations: generateQualityRecommendations(metrics),
totalMemories: memories.length
};
}
/**
* Assess tagging quality
* @param {Array} memories - Array of memory objects
* @returns {Object} Tagging quality assessment
*/
function assessTaggingQuality(memories) {
let taggedCount = 0;
let wellTaggedCount = 0;
let totalTags = 0;
memories.forEach(memory => {
const tags = memory.tags || [];
totalTags += tags.length;
if (tags.length > 0) {
taggedCount++;
// Well-tagged: has 3+ tags from different categories
if (tags.length >= 3) {
const categories = new Set(tags.map(tag => categorizeTag(tag)));
if (categories.size >= 2) {
wellTaggedCount++;
}
}
}
});
const taggedPercentage = (taggedCount / memories.length) * 100;
const wellTaggedPercentage = (wellTaggedCount / memories.length) * 100;
const averageTagsPerMemory = totalTags / memories.length;
let score = 0;
if (taggedPercentage >= 90) score += 40;
else if (taggedPercentage >= 70) score += 30;
else if (taggedPercentage >= 50) score += 20;
if (wellTaggedPercentage >= 70) score += 30;
else if (wellTaggedPercentage >= 50) score += 20;
else if (wellTaggedPercentage >= 30) score += 10;
if (averageTagsPerMemory >= 4) score += 30;
else if (averageTagsPerMemory >= 3) score += 20;
else if (averageTagsPerMemory >= 2) score += 10;
return {
score,
taggedPercentage: Math.round(taggedPercentage),
wellTaggedPercentage: Math.round(wellTaggedPercentage),
averageTagsPerMemory: Math.round(averageTagsPerMemory * 10) / 10,
issues: {
untagged: memories.length - taggedCount,
poorlyTagged: taggedCount - wellTaggedCount
}
};
}
/**
* Assess content quality
* @param {Array} memories - Array of memory objects
* @returns {Object} Content quality assessment
*/
function assessContentQuality(memories) {
let substantialContent = 0;
let hasDescription = 0;
let totalLength = 0;
memories.forEach(memory => {
const content = memory.content || '';
totalLength += content.length;
if (content.length >= 50) {
substantialContent++;
}
if (content.length >= 200) {
hasDescription++;
}
});
const substantialPercentage = (substantialContent / memories.length) * 100;
const descriptivePercentage = (hasDescription / memories.length) * 100;
const averageLength = totalLength / memories.length;
let score = 0;
if (substantialPercentage >= 90) score += 50;
else if (substantialPercentage >= 70) score += 35;
else if (substantialPercentage >= 50) score += 20;
if (descriptivePercentage >= 60) score += 30;
else if (descriptivePercentage >= 40) score += 20;
else if (descriptivePercentage >= 20) score += 10;
if (averageLength >= 300) score += 20;
else if (averageLength >= 150) score += 10;
return {
score,
substantialPercentage: Math.round(substantialPercentage),
descriptivePercentage: Math.round(descriptivePercentage),
averageLength: Math.round(averageLength),
issues: {
tooShort: memories.length - substantialContent,
lackingDescription: memories.length - hasDescription
}
};
}
/**
* Assess organization quality
* @param {Array} memories - Array of memory objects
* @returns {Object} Organization quality assessment
*/
function assessOrganizationQuality(memories) {
const tagAnalysis = analyzeTagUsage(memories);
const categories = Object.keys(tagAnalysis.categories);
const topTags = tagAnalysis.frequency.slice(0, 10);
// Check for balanced tag distribution
const tagDistribution = tagAnalysis.frequency.map(([, count]) => count);
const maxUsage = Math.max(...tagDistribution);
const minUsage = Math.min(...tagDistribution);
const distributionBalance = minUsage / maxUsage;
let score = 0;
// Category diversity
if (categories.length >= 5) score += 30;
else if (categories.length >= 3) score += 20;
else if (categories.length >= 2) score += 10;
// Tag usage balance
if (distributionBalance >= 0.3) score += 25;
else if (distributionBalance >= 0.2) score += 15;
else if (distributionBalance >= 0.1) score += 5;
// Consistent tag combinations
if (tagAnalysis.combinations.length >= 10) score += 25;
else if (tagAnalysis.combinations.length >= 5) score += 15;
// Avoid over-concentration
const topTagUsagePercentage = (topTags[0]?.[1] || 0) / memories.length * 100;
if (topTagUsagePercentage <= 30) score += 20;
else if (topTagUsagePercentage <= 40) score += 10;
return {
score,
categoryCount: categories.length,
tagDistributionBalance: Math.round(distributionBalance * 100),
topTagUsagePercentage: Math.round(topTagUsagePercentage),
consistentCombinations: tagAnalysis.combinations.length,
issues: {
fewCategories: categories.length < 3,
imbalancedDistribution: distributionBalance < 0.2,
overConcentration: topTagUsagePercentage > 40
}
};
}
/**
* Assess searchability quality
* @param {Array} memories - Array of memory objects
* @returns {Object} Searchability quality assessment
*/
function assessSearchabilityQuality(memories) {
const contentAnalysis = analyzeContentPatterns(memories);
const tagAnalysis = analyzeTagUsage(memories);
// Calculate searchability metrics
const keywordDiversity = Object.keys(contentAnalysis.wordFrequency).length;
const tagDiversity = tagAnalysis.totalTags;
const averageTagsPerMemory = tagAnalysis.averageTagsPerMemory;
let score = 0;
// Keyword diversity
if (keywordDiversity >= 100) score += 25;
else if (keywordDiversity >= 50) score += 15;
else if (keywordDiversity >= 25) score += 5;
// Tag diversity
if (tagDiversity >= 50) score += 25;
else if (tagDiversity >= 30) score += 15;
else if (tagDiversity >= 15) score += 5;
// Tag coverage
if (averageTagsPerMemory >= 4) score += 25;
else if (averageTagsPerMemory >= 3) score += 15;
else if (averageTagsPerMemory >= 2) score += 5;
// Content type diversity
const contentTypes = Object.keys(contentAnalysis.contentTypes).length;
if (contentTypes >= 5) score += 25;
else if (contentTypes >= 3) score += 15;
else if (contentTypes >= 2) score += 5;
return {
score,
keywordDiversity,
tagDiversity,
averageTagsPerMemory: Math.round(averageTagsPerMemory * 10) / 10,
contentTypeDiversity: contentTypes,
issues: {
lowKeywordDiversity: keywordDiversity < 25,
lowTagDiversity: tagDiversity < 15,
poorTagCoverage: averageTagsPerMemory < 2
}
};
}
/**
* Generate quality improvement recommendations
* @param {Object} metrics - Quality metrics object
* @returns {Array} Array of recommendations
*/
function generateQualityRecommendations(metrics) {
const recommendations = [];
// Tagging recommendations
if (metrics.tagging.taggedPercentage < 90) {
recommendations.push({
category: 'tagging',
priority: 'high',
issue: `${metrics.tagging.issues.untagged} memories are untagged`,
action: 'Run memory maintenance session to tag untagged memories',
expectedImprovement: 'Improve searchability and organization'
});
}
if (metrics.tagging.averageTagsPerMemory < 3) {
recommendations.push({
category: 'tagging',
priority: 'medium',
issue: 'Low average tags per memory',
action: 'Add more specific and categorical tags to existing memories',
expectedImprovement: 'Better categorization and discoverability'
});
}
// Content recommendations
if (metrics.content.substantialPercentage < 80) {
recommendations.push({
category: 'content',
priority: 'medium',
issue: `${metrics.content.issues.tooShort} memories have minimal content`,
action: 'Expand brief memories with more context and details',
expectedImprovement: 'Increased information value and searchability'
});
}
// Organization recommendations
if (metrics.organization.categoryCount < 3) {
recommendations.push({
category: 'organization',
priority: 'high',
issue: 'Limited tag category diversity',
action: 'Implement standardized tag schema with multiple categories',
expectedImprovement: 'Better knowledge organization structure'
});
}
if (metrics.organization.tagDistributionBalance < 20) {
recommendations.push({
category: 'organization',
priority: 'medium',
issue: 'Imbalanced tag usage distribution',
action: 'Review and balance tag usage across content types',
expectedImprovement: 'More consistent knowledge organization'
});
}
// Searchability recommendations
if (metrics.searchability.tagDiversity < 30) {
recommendations.push({
category: 'searchability',
priority: 'medium',
issue: 'Limited tag vocabulary',
action: 'Expand tag vocabulary with more specific and varied tags',
expectedImprovement: 'Enhanced search precision and recall'
});
}
return recommendations.sort((a, b) => {
const priorityOrder = { 'high': 3, 'medium': 2, 'low': 1 };
return priorityOrder[b.priority] - priorityOrder[a.priority];
});
}
// =============================================================================
// VISUALIZATION DATA PREPARATION
// =============================================================================
/**
* Prepare comprehensive data package for visualizations
* @param {Array} memories - Array of memory objects
* @returns {Object} Complete visualization data package
*/
function prepareVisualizationData(memories) {
const temporal = analyzeTemporalDistribution(memories);
const tags = analyzeTagUsage(memories);
const content = analyzeContentPatterns(memories);
const quality = assessMemoryQuality(memories);
return {
metadata: {
totalMemories: memories.length,
analysisDate: new Date().toISOString(),
dataVersion: '1.0'
},
// Chart data for different visualizations
charts: {
temporalDistribution: prepareTemporalChartData(temporal, 'monthly'),
weeklyPattern: prepareTemporalChartData(temporal, 'weekly'),
dailyPattern: prepareTemporalChartData(temporal, 'daily'),
hourlyPattern: prepareTemporalChartData(temporal, 'hourly'),
tagFrequency: tags.frequency.slice(0, 20).map(([tag, count]) => ({
tag,
count,
category: categorizeTag(tag)
})),
tagCombinations: tags.combinations.slice(0, 10).map(([combo, count]) => ({
combination: combo,
count,
tags: combo.split(' + ')
})),
contentTypes: Object.entries(content.contentTypes).map(([type, count]) => ({
type,
count,
percentage: Math.round((count / memories.length) * 100)
})),
contentLengths: Object.entries(content.lengthDistribution).map(([category, count]) => ({
category,
count,
percentage: Math.round((count / memories.length) * 100)
}))
},
// Summary statistics
statistics: {
temporal: {
peakMonth: findPeakPeriod(temporal.monthly),
mostActiveDay: findPeakPeriod(temporal.daily),
mostActiveHour: findPeakPeriod(temporal.hourly)
},
tags: {
totalUniqueTags: tags.totalTags,
averageTagsPerMemory: Math.round(tags.averageTagsPerMemory * 10) / 10,
mostUsedTag: tags.frequency[0],
categoryDistribution: Object.keys(tags.categories).length
},
content: {
averageLength: Math.round(content.averageLength),
mostCommonType: Object.entries(content.contentTypes)
.sort(([,a], [,b]) => b - a)[0],
keywordCount: Object.keys(content.wordFrequency).length
},
quality: {
overallScore: quality.overallScore,
taggedPercentage: quality.metrics.tagging.taggedPercentage,
organizationScore: quality.metrics.organization.score,
recommendationCount: quality.recommendations.length
}
},
// Raw analysis data for advanced processing
rawData: {
temporal,
tags,
content,
quality
}
};
}
/**
* Find peak period from distribution data
* @param {Object} distribution - Distribution object
* @returns {Object} Peak period information
*/
function findPeakPeriod(distribution) {
const entries = Object.entries(distribution);
if (entries.length === 0) return null;
const peak = entries.reduce((max, [period, memories]) =>
memories.length > max.count ? { period, count: memories.length } : max,
{ period: null, count: 0 }
);
return peak;
}
// =============================================================================
// EXPORT FUNCTIONS
// =============================================================================
/**
* Export analysis results to various formats
* @param {Object} analysisData - Complete analysis data
* @param {string} format - Export format ('json', 'csv', 'summary')
* @returns {string} Formatted export data
*/
function exportAnalysisData(analysisData, format = 'json') {
switch (format) {
case 'json':
return JSON.stringify(analysisData, null, 2);
case 'csv':
return exportToCSV(analysisData);
case 'summary':
return generateSummaryReport(analysisData);
default:
throw new Error(`Unsupported export format: ${format}`);
}
}
/**
* Export key metrics to CSV format
* @param {Object} analysisData - Analysis data
* @returns {string} CSV formatted data
*/
function exportToCSV(analysisData) {
const csvSections = [];
// Temporal data
csvSections.push('TEMPORAL DISTRIBUTION');
csvSections.push('Month,Count');
analysisData.charts.temporalDistribution.forEach(item => {
csvSections.push(`${item.period},${item.count}`);
});
csvSections.push('');
// Tag frequency
csvSections.push('TAG FREQUENCY');
csvSections.push('Tag,Count,Category');
analysisData.charts.tagFrequency.forEach(item => {
csvSections.push(`${item.tag},${item.count},${item.category}`);
});
csvSections.push('');
// Content types
csvSections.push('CONTENT TYPES');
csvSections.push('Type,Count,Percentage');
analysisData.charts.contentTypes.forEach(item => {
csvSections.push(`${item.type},${item.count},${item.percentage}%`);
});
return csvSections.join('\n');
}
/**
* Generate a human-readable summary report
* @param {Object} analysisData - Analysis data
* @returns {string} Summary report
*/
function generateSummaryReport(analysisData) {
const stats = analysisData.statistics;
const quality = analysisData.rawData.quality;
return `
MEMORY ANALYSIS SUMMARY REPORT
Generated: ${new Date().toLocaleDateString()}
DATABASE OVERVIEW:
- Total Memories: ${analysisData.metadata.totalMemories}
- Overall Quality Score: ${stats.quality.overallScore}/100
- Tagged Memories: ${stats.quality.taggedPercentage}%
TEMPORAL PATTERNS:
- Peak Activity: ${stats.temporal.peakMonth?.period} (${stats.temporal.peakMonth?.count} memories)
- Most Active Day: ${stats.temporal.mostActiveDay?.period}
- Most Active Hour: ${stats.temporal.mostActiveHour?.period}:00
TAG ANALYSIS:
- Unique Tags: ${stats.tags.totalUniqueTags}
- Average Tags per Memory: ${stats.tags.averageTagsPerMemory}
- Most Used Tag: ${stats.tags.mostUsedTag?.[0]} (${stats.tags.mostUsedTag?.[1]} uses)
- Tag Categories: ${stats.tags.categoryDistribution}
CONTENT INSIGHTS:
- Average Length: ${stats.content.averageLength} characters
- Most Common Type: ${stats.content.mostCommonType?.[0]}
- Unique Keywords: ${stats.content.keywordCount}
QUALITY RECOMMENDATIONS:
${quality.recommendations.slice(0, 3).map(rec =>
`- ${rec.priority.toUpperCase()}: ${rec.action}`
).join('\n')}
For detailed analysis, use the full JSON export or visualization tools.
`.trim();
}
// =============================================================================
// MAIN ANALYSIS PIPELINE
// =============================================================================
/**
* Run complete analysis pipeline on memory data
* @param {Array} memories - Array of memory objects
* @returns {Object} Complete analysis results
*/
async function runCompleteAnalysis(memories) {
console.log('Starting comprehensive memory analysis...');
const startTime = Date.now();
try {
// Run all analysis functions
const results = prepareVisualizationData(memories);
const endTime = Date.now();
const duration = endTime - startTime;
console.log(`Analysis complete in ${duration}ms`);
console.log(`Analyzed ${memories.length} memories`);
console.log(`Overall quality score: ${results.statistics.quality.overallScore}/100`);
return {
...results,
meta: {
analysisDuration: duration,
analysisTimestamp: new Date().toISOString(),
version: '1.0'
}
};
} catch (error) {
console.error('Analysis failed:', error);
throw error;
}
}
// Export all functions for use in other modules
if (typeof module !== 'undefined' && module.exports) {
module.exports = {
// Temporal analysis
analyzeTemporalDistribution,
prepareTemporalChartData,
// Tag analysis
analyzeTagUsage,
analyzeTagConsistency,
categorizeTag,
// Content analysis
analyzeContentPatterns,
detectContentType,
extractKeywords,
// Quality analysis
assessMemoryQuality,
generateQualityRecommendations,
// Visualization
prepareVisualizationData,
// Export utilities
exportAnalysisData,
generateSummaryReport,
// Main pipeline
runCompleteAnalysis
};
}
/**
* Usage Examples:
*
* // Basic usage with MCP Memory Service data
* const memories = await retrieve_memory({ query: "all memories", n_results: 500 });
* const analysis = await runCompleteAnalysis(memories);
*
* // Specific analyses
* const temporalData = analyzeTemporalDistribution(memories);
* const tagAnalysis = analyzeTagUsage(memories);
* const qualityReport = assessMemoryQuality(memories);
*
* // Export results
* const jsonExport = exportAnalysisData(analysis, 'json');
* const csvExport = exportAnalysisData(analysis, 'csv');
* const summary = exportAnalysisData(analysis, 'summary');
*
* // Prepare data for React charts
* const chartData = prepareVisualizationData(memories);
* // Use chartData.charts.temporalDistribution with the React component
*/
```
--------------------------------------------------------------------------------
/src/mcp_memory_service/web/api/analytics.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.
from __future__ import annotations
"""
Analytics endpoints for the HTTP interface.
Provides usage statistics, trends, and performance metrics for the memory system.
"""
import logging
from typing import List, Optional, Dict, Any, TYPE_CHECKING, Tuple
from datetime import datetime, timedelta, timezone
from collections import defaultdict
from dataclasses import dataclass
from enum import Enum
from fastapi import APIRouter, HTTPException, Depends, Query
from pydantic import BaseModel, Field
from ...storage.base import MemoryStorage
from ...config import OAUTH_ENABLED
from ..dependencies import get_storage
# OAuth authentication imports (conditional)
if OAUTH_ENABLED or TYPE_CHECKING:
from ..oauth.middleware import require_read_access, AuthenticationResult
else:
# Provide type stubs when OAuth is disabled
AuthenticationResult = None
require_read_access = None
router = APIRouter()
logger = logging.getLogger(__name__)
# Helper functions for analytics endpoints
async def fetch_storage_stats(storage: MemoryStorage) -> Dict[str, Any]:
"""Fetch storage statistics from the storage backend.
Args:
storage: MemoryStorage backend instance
Returns:
Dict containing storage stats, or empty dict if unavailable
"""
if hasattr(storage, 'get_stats'):
try:
return await storage.get_stats()
except Exception as e:
logger.warning(f"Failed to retrieve storage stats: {e}")
return {}
return {}
def calculate_tag_statistics(tag_data: List[Dict[str, Any]], total_memories: int) -> List[TagUsageStats]:
"""Calculate tag usage statistics with percentages.
Args:
tag_data: List of dicts with 'tag' and 'count' keys
total_memories: Total number of memories for percentage calculation
Returns:
List of TagUsageStats objects
"""
tags = []
for tag_item in tag_data:
percentage = (tag_item["count"] / total_memories * 100) if total_memories > 0 else 0
tags.append(TagUsageStats(
tag=tag_item["tag"],
count=tag_item["count"],
percentage=round(percentage, 1),
growth_rate=None # Would need historical data to calculate
))
return tags
def calculate_activity_time_ranges(timestamps: List[float], granularity: str) -> Tuple[List[ActivityBreakdown], set, List]:
"""Calculate activity breakdown by time range (hourly, daily, weekly).
Args:
timestamps: List of Unix timestamps
granularity: One of 'hourly', 'daily', 'weekly'
Returns:
Tuple of (breakdown_list, active_days_set, activity_dates_list)
"""
breakdown = []
active_days = set()
activity_dates = []
# Convert all timestamps to datetime objects and populate active_days/activity_dates once
dts = [datetime.fromtimestamp(ts, tz=timezone.utc) for ts in timestamps]
for dt in dts:
active_days.add(dt.date())
activity_dates.append(dt.date())
if granularity == "hourly":
hour_counts = defaultdict(int)
for dt in dts:
hour_counts[dt.hour] += 1
for hour in range(24):
count = hour_counts.get(hour, 0)
label = f"{hour:02d}:00"
breakdown.append(ActivityBreakdown(
period="hourly",
count=count,
label=label
))
elif granularity == "daily":
day_counts = defaultdict(int)
day_names = ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday"]
for dt in dts:
day_counts[dt.weekday()] += 1
for i, day_name in enumerate(day_names):
count = day_counts.get(i, 0)
breakdown.append(ActivityBreakdown(
period="daily",
count=count,
label=day_name
))
else: # weekly
week_counts = defaultdict(int)
for dt in dts:
# Get ISO week number with year
year, week_num, _ = dt.isocalendar()
week_key = f"{year}-W{week_num:02d}"
week_counts[week_key] += 1
# Last 12 weeks
now = datetime.now(timezone.utc)
for i in range(12):
# Calculate target date
target_date = now - timedelta(weeks=(11 - i))
year, week_num, _ = target_date.isocalendar()
week_key = f"{year}-W{week_num:02d}"
count = week_counts.get(week_key, 0)
breakdown.append(ActivityBreakdown(
period="weekly",
count=count,
label=f"Week {week_num} ({year})"
))
return breakdown, active_days, activity_dates
def aggregate_type_statistics(type_counts: Dict[str, int], total_memories: int) -> List[MemoryTypeDistribution]:
"""Aggregate memory type statistics with percentages.
Args:
type_counts: Dict mapping memory types to counts
total_memories: Total number of memories
Returns:
List of MemoryTypeDistribution objects, sorted by count descending
"""
types = []
for mem_type, count in type_counts.items():
percentage = (count / total_memories * 100) if total_memories > 0 else 0
types.append(MemoryTypeDistribution(
memory_type=mem_type,
count=count,
percentage=round(percentage, 1)
))
# Sort by count
types.sort(key=lambda x: x.count, reverse=True)
return types
# Period Configuration for Analytics
class PeriodType(str, Enum):
"""Valid time period types for analytics."""
WEEK = "week"
MONTH = "month"
QUARTER = "quarter"
YEAR = "year"
@dataclass
class PeriodConfig:
"""Configuration for time period analysis."""
days: int
interval_days: int
PERIOD_CONFIGS = {
PeriodType.WEEK: PeriodConfig(days=7, interval_days=1),
PeriodType.MONTH: PeriodConfig(days=30, interval_days=7), # Weekly aggregation for monthly view
PeriodType.QUARTER: PeriodConfig(days=90, interval_days=7),
PeriodType.YEAR: PeriodConfig(days=365, interval_days=30),
}
def get_period_config(period: PeriodType) -> PeriodConfig:
"""Get configuration for the specified time period.
Args:
period: Time period identifier (week, month, quarter, year)
Returns:
PeriodConfig for the specified period
Raises:
HTTPException: If period is invalid
"""
config = PERIOD_CONFIGS.get(period)
if not config:
valid_periods = ', '.join(p.value for p in PeriodType)
raise HTTPException(
status_code=400,
detail=f"Invalid period. Use: {valid_periods}"
)
return config
# Response Models
class AnalyticsOverview(BaseModel):
"""Overview statistics for the memory system."""
total_memories: int
memories_this_week: int
memories_this_month: int
unique_tags: int
database_size_mb: Optional[float]
uptime_seconds: Optional[float]
backend_type: str
class MemoryGrowthPoint(BaseModel):
"""Data point for memory growth over time."""
date: str # YYYY-MM-DD format
count: int
cumulative: int
label: Optional[str] = None # Human-readable label (e.g., "Week of Nov 1", "November 2024")
class MemoryGrowthData(BaseModel):
"""Memory growth data over time."""
data_points: List[MemoryGrowthPoint]
period: str # "week", "month", "quarter", "year"
class TagUsageStats(BaseModel):
"""Usage statistics for a specific tag."""
tag: str
count: int
percentage: float
growth_rate: Optional[float] # Growth rate compared to previous period
class TagUsageData(BaseModel):
"""Tag usage analytics."""
tags: List[TagUsageStats]
total_memories: int
period: str
class MemoryTypeDistribution(BaseModel):
"""Distribution of memories by type."""
memory_type: str
count: int
percentage: float
class MemoryTypeData(BaseModel):
"""Memory type distribution data."""
types: List[MemoryTypeDistribution]
total_memories: int
class SearchAnalytics(BaseModel):
"""Search usage analytics."""
total_searches: int = 0
avg_response_time: Optional[float] = None
popular_tags: List[Dict[str, Any]] = []
search_types: Dict[str, int] = {}
class PerformanceMetrics(BaseModel):
"""System performance metrics."""
avg_response_time: Optional[float] = None
memory_usage_mb: Optional[float] = None
storage_latency: Optional[float] = None
error_rate: Optional[float] = None
class ActivityHeatmapData(BaseModel):
"""Activity heatmap data for calendar view."""
date: str # YYYY-MM-DD format
count: int
level: int # 0-4 activity level for color coding
class ActivityHeatmapResponse(BaseModel):
"""Response containing activity heatmap data."""
data: List[ActivityHeatmapData]
total_days: int
max_count: int
class TopTagsReport(BaseModel):
"""Enhanced top tags report with trends and co-occurrence."""
tag: str
count: int
percentage: float
growth_rate: Optional[float]
trending: bool # Is usage increasing
co_occurring_tags: List[Dict[str, Any]] # Tags that appear with this tag
class TopTagsResponse(BaseModel):
"""Response for top tags report."""
tags: List[TopTagsReport]
period: str
class ActivityBreakdown(BaseModel):
"""Activity breakdown by time period."""
period: str # hour, day, week, month
count: int
label: str # e.g., "Monday", "10 AM", etc.
class ActivityReport(BaseModel):
"""Comprehensive activity report."""
breakdown: List[ActivityBreakdown]
peak_times: List[str]
active_days: int
total_days: int
current_streak: int
longest_streak: int
class LargestMemory(BaseModel):
"""A single large memory entry."""
content_hash: str
size_bytes: int
size_kb: float
created_at: Optional[str] = None
tags: List[str] = []
preview: str # First 100 chars
class GrowthTrendPoint(BaseModel):
"""Storage growth at a point in time."""
date: str # ISO format YYYY-MM-DD
total_size_mb: float
memory_count: int
class StorageStats(BaseModel):
"""Storage statistics and largest memories."""
total_size_mb: float
average_memory_size: float
largest_memories: List[LargestMemory]
growth_trend: List[GrowthTrendPoint]
storage_efficiency: float # Percentage of efficient storage
@router.get("/overview", response_model=AnalyticsOverview, tags=["analytics"])
async def get_analytics_overview(
storage: MemoryStorage = Depends(get_storage),
user: AuthenticationResult = Depends(require_read_access) if OAUTH_ENABLED else None
):
"""
Get overview analytics for the memory system.
Returns key metrics including total memories, recent activity, and system stats.
"""
try:
# Get detailed health data which contains most stats
if hasattr(storage, 'get_stats'):
try:
stats = await storage.get_stats()
logger.info(f"Storage stats: {stats}") # Debug logging
except Exception as e:
logger.warning(f"Failed to retrieve storage stats: {e}")
stats = {}
else:
stats = {}
# Get memories_this_week from storage stats (accurate for all memories)
memories_this_week = stats.get("memories_this_week", 0)
# Calculate memories this month
# TODO: Add memories_this_month to storage.get_stats() for consistency
month_ago = datetime.now(timezone.utc) - timedelta(days=30)
month_ago_ts = month_ago.timestamp()
memories_this_month = 0
try:
# Use larger sample for monthly calculation
# Note: This may be inaccurate if there are >5000 memories
recent_memories = await storage.get_recent_memories(n=5000)
memories_this_month = sum(1 for m in recent_memories if m.created_at and m.created_at > month_ago_ts)
except Exception as e:
logger.warning(f"Failed to calculate monthly memories: {e}")
memories_this_month = 0
return AnalyticsOverview(
total_memories=stats.get("total_memories", 0),
memories_this_week=memories_this_week,
memories_this_month=memories_this_month,
unique_tags=stats.get("unique_tags", 0),
database_size_mb=stats.get("primary_stats", {}).get("database_size_mb") or stats.get("database_size_mb"),
uptime_seconds=None, # Would need to be calculated from health endpoint
backend_type=stats.get("storage_backend", "unknown")
)
except Exception as e:
logger.error(f"Failed to get analytics overview: {str(e)}")
raise HTTPException(status_code=500, detail=f"Failed to get analytics overview: {str(e)}")
# Label formatters for each period type
PERIOD_LABEL_FORMATTERS = {
PeriodType.WEEK: lambda date: date.strftime("%b %d"), # "Nov 15"
PeriodType.MONTH: lambda date: f"Week of {date.strftime('%b %d')}", # "Week of Nov 15"
PeriodType.QUARTER: lambda date: f"Week of {date.strftime('%b %d')}", # "Week of Nov 15"
PeriodType.YEAR: lambda date: date.strftime("%B %Y"), # "November 2024"
}
def _generate_interval_label(date: datetime, period: PeriodType) -> str:
"""
Generate a human-readable label for a date interval based on the period type.
Args:
date: The date for the interval
period: The period type (week, month, quarter, year)
Returns:
A formatted label string
"""
formatter = PERIOD_LABEL_FORMATTERS.get(period)
if formatter:
return formatter(date)
# Fallback to ISO format
return date.strftime("%Y-%m-%d")
@router.get("/memory-growth", response_model=MemoryGrowthData, tags=["analytics"])
async def get_memory_growth(
period: PeriodType = Query(PeriodType.MONTH, description="Time period: week, month, quarter, year"),
storage: MemoryStorage = Depends(get_storage),
user: AuthenticationResult = Depends(require_read_access) if OAUTH_ENABLED else None
):
"""
Get memory growth data over time.
Returns data points showing how the memory count has grown over the specified period.
"""
try:
# Get period configuration
config = get_period_config(period)
days = config.days
interval_days = config.interval_days
# Calculate date ranges
end_date = datetime.now(timezone.utc)
start_date = end_date - timedelta(days=days)
# This is a simplified implementation
# In a real system, we'd need efficient date-range queries in the storage layer
data_points = []
cumulative = 0
try:
# Performance optimization: Use database-layer filtering instead of
# fetching all memories and filtering in Python (10x improvement)
# This pushes the date range filter to the storage backend (SQLite WHERE clause
# or Cloudflare D1 query), reducing memory consumption and network transfer
date_counts = defaultdict(int)
start_timestamp = start_date.timestamp()
end_timestamp = end_date.timestamp()
# Get memories in date range (database-filtered)
memories_in_range = await storage.get_memories_by_time_range(start_timestamp, end_timestamp)
# Group by date
for memory in memories_in_range:
if memory.created_at:
mem_date = datetime.fromtimestamp(memory.created_at, tz=timezone.utc).date()
date_counts[mem_date] += 1
# Create data points
current_date = start_date.date()
while current_date <= end_date.date():
# For intervals > 1 day, sum counts across the entire interval
interval_end = current_date + timedelta(days=interval_days)
count = 0
# Sum all memories within this interval
check_date = current_date
while check_date < interval_end and check_date <= end_date.date():
count += date_counts.get(check_date, 0)
check_date += timedelta(days=1)
cumulative += count
# Convert date to datetime for label generation
current_datetime = datetime.combine(current_date, datetime.min.time())
label = _generate_interval_label(current_datetime, period)
data_points.append(MemoryGrowthPoint(
date=current_date.isoformat(),
count=count,
cumulative=cumulative,
label=label
))
current_date += timedelta(days=interval_days)
except Exception as e:
logger.warning(f"Failed to calculate memory growth: {str(e)}")
# Return empty data if calculation fails
data_points = []
return MemoryGrowthData(
data_points=data_points,
period=period.value
)
except HTTPException:
raise
except Exception as e:
logger.error(f"Failed to get memory growth data: {str(e)}")
raise HTTPException(status_code=500, detail=f"Failed to get memory growth data: {str(e)}")
@router.get("/tag-usage", response_model=TagUsageData, tags=["analytics"])
async def get_tag_usage_analytics(
period: str = Query("all", description="Time period: week, month, all"),
limit: int = Query(20, description="Maximum number of tags to return"),
storage: MemoryStorage = Depends(get_storage),
user: AuthenticationResult = Depends(require_read_access) if OAUTH_ENABLED else None
):
"""
Get tag usage analytics.
Returns statistics about tag usage, optionally filtered by time period.
"""
try:
# Get all tags with counts
if hasattr(storage, 'get_all_tags_with_counts'):
tag_data = await storage.get_all_tags_with_counts()
else:
raise HTTPException(status_code=501, detail="Tag analytics not supported by storage backend")
# Get total memories for accurate percentage calculation
stats = await fetch_storage_stats(storage)
total_memories = stats.get("total_memories", 0)
if total_memories == 0:
# Fallback: count all memories directly for an accurate total.
total_memories = await storage.count_all_memories()
# Sort by count and limit
tag_data.sort(key=lambda x: x["count"], reverse=True)
tag_data = tag_data[:limit]
# Convert to response format using helper
tags = calculate_tag_statistics(tag_data, total_memories)
return TagUsageData(
tags=tags,
total_memories=total_memories,
period=period
)
except HTTPException:
raise
except Exception as e:
logger.error(f"Failed to get tag usage analytics: {str(e)}")
raise HTTPException(status_code=500, detail=f"Failed to get tag usage analytics: {str(e)}")
@router.get("/memory-types", response_model=MemoryTypeData, tags=["analytics"])
async def get_memory_type_distribution(
storage: MemoryStorage = Depends(get_storage),
user: AuthenticationResult = Depends(require_read_access) if OAUTH_ENABLED else None
):
"""
Get distribution of memories by type.
Returns statistics about how memories are categorized by type.
"""
try:
# Try to get accurate counts from storage layer if available
if hasattr(storage, 'get_type_counts'):
type_counts_data = await storage.get_type_counts()
type_counts = dict(type_counts_data)
total_memories = sum(type_counts.values())
# For Hybrid storage, access underlying SQLite primary storage
elif hasattr(storage, 'primary') and hasattr(storage.primary, 'conn') and storage.primary.conn:
# Hybrid storage - access underlying SQLite storage
import sqlite3
cursor = storage.primary.conn.cursor()
cursor.execute("""
SELECT
CASE
WHEN memory_type IS NULL OR memory_type = '' THEN 'untyped'
ELSE memory_type
END as mem_type,
COUNT(*) as count
FROM memories
GROUP BY mem_type
""")
type_counts = {row[0]: row[1] for row in cursor.fetchall()}
cursor.execute("SELECT COUNT(*) FROM memories")
total_memories = cursor.fetchone()[0]
elif hasattr(storage, 'conn') and storage.conn:
# Direct SQLite storage
import sqlite3
cursor = storage.conn.cursor()
cursor.execute("""
SELECT
CASE
WHEN memory_type IS NULL OR memory_type = '' THEN 'untyped'
ELSE memory_type
END as mem_type,
COUNT(*) as count
FROM memories
GROUP BY mem_type
""")
type_counts = {row[0]: row[1] for row in cursor.fetchall()}
cursor.execute("SELECT COUNT(*) FROM memories")
total_memories = cursor.fetchone()[0]
else:
# Fallback to sampling approach (less accurate for large databases)
logger.warning("Using sampling approach for memory type distribution - results may not reflect entire database")
memories = await storage.get_recent_memories(n=1000)
type_counts = defaultdict(int)
for memory in memories:
mem_type = memory.memory_type or "untyped"
type_counts[mem_type] += 1
total_memories = len(memories)
# Convert to response format using helper
types = aggregate_type_statistics(type_counts, total_memories)
return MemoryTypeData(
types=types,
total_memories=total_memories
)
except Exception as e:
logger.error(f"Failed to get memory type distribution: {str(e)}")
raise HTTPException(status_code=500, detail=f"Failed to get memory type distribution: {str(e)}")
@router.get("/search-analytics", response_model=SearchAnalytics, tags=["analytics"])
async def get_search_analytics(
user: AuthenticationResult = Depends(require_read_access) if OAUTH_ENABLED else None
):
"""
Get search usage analytics.
Returns statistics about search patterns and performance.
This is a placeholder - real implementation would need search logging.
"""
# Placeholder implementation
# In a real system, this would analyze search logs
return SearchAnalytics(
total_searches=0,
avg_response_time=None,
popular_tags=[],
search_types={}
)
@router.get("/performance", response_model=PerformanceMetrics, tags=["analytics"])
async def get_performance_metrics(
storage: MemoryStorage = Depends(get_storage),
user: AuthenticationResult = Depends(require_read_access) if OAUTH_ENABLED else None
):
"""
Get system performance metrics.
Returns performance statistics for the memory system.
"""
# Placeholder implementation
# In a real system, this would collect actual performance metrics
return PerformanceMetrics(
avg_response_time=None,
memory_usage_mb=None,
storage_latency=None,
error_rate=None
)
@router.get("/activity-heatmap", response_model=ActivityHeatmapResponse, tags=["analytics"])
async def get_activity_heatmap(
days: int = Query(365, description="Number of days to include in heatmap"),
storage: MemoryStorage = Depends(get_storage),
user: AuthenticationResult = Depends(require_read_access) if OAUTH_ENABLED else None
):
"""
Get activity heatmap data for calendar view.
Returns daily activity counts for the specified period, with activity levels for color coding.
"""
try:
# Use optimized timestamp-only fetching (v8.18.0+)
timestamps = await storage.get_memory_timestamps(days=days)
# Group by date
date_counts = defaultdict(int)
end_date = datetime.now(timezone.utc).date()
start_date = end_date - timedelta(days=days)
for timestamp in timestamps:
mem_date = datetime.fromtimestamp(timestamp, tz=timezone.utc).date()
if start_date <= mem_date <= end_date:
date_counts[mem_date] += 1
# Create heatmap data
heatmap_data = []
total_days = 0
max_count = 0
current_date = start_date
while current_date <= end_date:
count = date_counts.get(current_date, 0)
if count > 0:
total_days += 1
max_count = max(max_count, count)
# Calculate activity level (0-4)
if count == 0:
level = 0
elif count <= max_count * 0.25:
level = 1
elif count <= max_count * 0.5:
level = 2
elif count <= max_count * 0.75:
level = 3
else:
level = 4
heatmap_data.append(ActivityHeatmapData(
date=current_date.isoformat(),
count=count,
level=level
))
current_date += timedelta(days=1)
return ActivityHeatmapResponse(
data=heatmap_data,
total_days=total_days,
max_count=max_count
)
except Exception as e:
logger.error(f"Failed to get activity heatmap: {str(e)}")
raise HTTPException(status_code=500, detail=f"Failed to get activity heatmap: {str(e)}")
@router.get("/top-tags", response_model=TopTagsResponse, tags=["analytics"])
async def get_top_tags_report(
period: str = Query("30d", description="Time period: 7d, 30d, 90d, all"),
limit: int = Query(20, description="Maximum number of tags to return"),
storage: MemoryStorage = Depends(get_storage),
user: AuthenticationResult = Depends(require_read_access) if OAUTH_ENABLED else None
):
"""
Get enhanced top tags report with trends and co-occurrence patterns.
Returns detailed tag analytics including usage trends and related tags.
"""
try:
# Parse period
if period == "7d":
days = 7
elif period == "30d":
days = 30
elif period == "90d":
days = 90
else: # "all"
days = None
# Get tag usage data
if hasattr(storage, 'get_all_tags_with_counts'):
tag_data = await storage.get_all_tags_with_counts()
else:
raise HTTPException(status_code=501, detail="Tag analytics not supported by storage backend")
# Get total memories
if hasattr(storage, 'get_stats'):
stats = await storage.get_stats()
total_memories = stats.get("total_memories", 0)
else:
total_memories = sum(tag["count"] for tag in tag_data)
if total_memories == 0:
return TopTagsResponse(tags=[], period=period)
# Filter by time period if needed
if days is not None:
cutoff_ts = (datetime.now(timezone.utc) - timedelta(days=days)).timestamp()
# Get memories within the time range and count their tags
if hasattr(storage, 'get_memories_by_time_range'):
# Get memories from cutoff_ts to now
now_ts = datetime.now(timezone.utc).timestamp()
memories_in_period = await storage.get_memories_by_time_range(cutoff_ts, now_ts)
# Count tags from memories in this period
from collections import Counter
tag_counter = Counter()
period_memory_count = 0
for memory in memories_in_period:
period_memory_count += 1
if memory.tags:
for tag in memory.tags:
tag_counter[tag] += 1
# Convert to the expected format
tag_data = [{"tag": tag, "count": count} for tag, count in tag_counter.items()]
total_memories = period_memory_count
# If the storage backend doesn't support time range queries, fall back to all tags
# (This maintains backward compatibility with storage backends that don't implement the method)
# Sort and limit
tag_data.sort(key=lambda x: x["count"], reverse=True)
tag_data = tag_data[:limit]
# Calculate co-occurrence (simplified)
# In a real implementation, this would analyze memory-tag relationships
enhanced_tags = []
for tag_item in tag_data:
percentage = (tag_item["count"] / total_memories * 100) if total_memories > 0 else 0
# Placeholder co-occurrence data
# Real implementation would query the storage for tag co-occurrence
co_occurring = [
{"tag": "related-tag-1", "count": 5, "strength": 0.8},
{"tag": "related-tag-2", "count": 3, "strength": 0.6}
]
enhanced_tags.append(TopTagsReport(
tag=tag_item["tag"],
count=tag_item["count"],
percentage=round(percentage, 1),
growth_rate=None, # Would need historical data
trending=False, # Would need trend analysis
co_occurring_tags=co_occurring
))
return TopTagsResponse(
tags=enhanced_tags,
period=period
)
except HTTPException:
raise
except Exception as e:
logger.error(f"Failed to get top tags report: {str(e)}")
raise HTTPException(status_code=500, detail=f"Failed to get top tags report: {str(e)}")
@router.get("/activity-breakdown", response_model=ActivityReport, tags=["analytics"])
async def get_activity_breakdown(
granularity: str = Query("daily", description="Time granularity: hourly, daily, weekly"),
storage: MemoryStorage = Depends(get_storage),
user: AuthenticationResult = Depends(require_read_access) if OAUTH_ENABLED else None
):
"""
Get activity breakdown and patterns.
Returns activity statistics by time period, peak times, and streak information.
"""
try:
# Use optimized timestamp-only fetching (v8.18.0+)
# Get last 90 days of timestamps (adequate for all granularity levels)
timestamps = await storage.get_memory_timestamps(days=90)
# Group by granularity using helper function
breakdown, active_days, activity_dates = calculate_activity_time_ranges(timestamps, granularity)
# Calculate streaks
activity_dates = sorted(set(activity_dates))
current_streak = 0
longest_streak = 0
if activity_dates:
# Current streak - check backwards from today
today = datetime.now(timezone.utc).date()
activity_dates_set = set(activity_dates)
# A streak is only "current" if it includes today
if today in activity_dates_set:
day_to_check = today
while day_to_check in activity_dates_set:
current_streak += 1
day_to_check -= timedelta(days=1)
# Longest streak - iterate through sorted dates
temp_streak = 1 # Start at 1, not 0
longest_streak = 1 # At least 1 if there's any activity
for i in range(1, len(activity_dates)):
if activity_dates[i] == activity_dates[i-1] + timedelta(days=1):
temp_streak += 1
longest_streak = max(longest_streak, temp_streak)
else:
temp_streak = 1 # Reset to 1, not 0
# Find peak times (top 3)
sorted_breakdown = sorted(breakdown, key=lambda x: x.count, reverse=True)
peak_times = [item.label for item in sorted_breakdown[:3]]
# Calculate total_days as the span from oldest to newest memory
total_days = (activity_dates[-1] - activity_dates[0]).days + 1 if len(activity_dates) >= 2 else len(activity_dates)
return ActivityReport(
breakdown=breakdown,
peak_times=peak_times,
active_days=len(active_days),
total_days=total_days,
current_streak=current_streak,
longest_streak=max(longest_streak, current_streak)
)
except Exception as e:
logger.error(f"Failed to get activity breakdown: {str(e)}")
raise HTTPException(status_code=500, detail=f"Failed to get activity breakdown: {str(e)}")
@router.get("/storage-stats", response_model=StorageStats, tags=["analytics"])
async def get_storage_stats(
storage: MemoryStorage = Depends(get_storage),
user: AuthenticationResult = Depends(require_read_access) if OAUTH_ENABLED else None
):
"""
Get storage statistics and largest memories.
Returns comprehensive storage analytics including size trends and largest memories.
"""
try:
# Get basic stats
if hasattr(storage, 'get_stats'):
stats = await storage.get_stats()
else:
stats = {}
total_size_mb = stats.get("primary_stats", {}).get("database_size_mb") or stats.get("database_size_mb") or 0
total_memories = stats.get("primary_stats", {}).get("total_memories") or stats.get("total_memories") or 0
# Get recent memories for average size calculation (smaller sample)
recent_memories = await storage.get_recent_memories(n=100)
if recent_memories:
# Calculate average memory size from recent sample
total_content_length = sum(len(memory.content or "") for memory in recent_memories)
average_memory_size = total_content_length / len(recent_memories)
else:
average_memory_size = 0
# Get largest memories using efficient database query
largest_memories_objs = await storage.get_largest_memories(n=10)
largest_memories = []
for memory in largest_memories_objs:
size_bytes = len(memory.content or "")
content = memory.content or ""
largest_memories.append(LargestMemory(
content_hash=memory.content_hash,
size_bytes=size_bytes,
size_kb=round(size_bytes / 1024, 2),
created_at=datetime.fromtimestamp(memory.created_at, tz=timezone.utc).isoformat() if memory.created_at else None,
tags=memory.tags or [],
preview=content[:100] + "..." if len(content) > 100 else content
))
# Placeholder growth trend (would need historical data)
now = datetime.now(timezone.utc)
growth_trend = [
GrowthTrendPoint(
date=(now - timedelta(days=i)).date().isoformat(),
total_size_mb=round(total_size_mb * (0.9 + i * 0.01), 2),
memory_count=int(total_memories * (0.9 + i * 0.01))
)
for i in range(30, 0, -1)
]
# Storage efficiency (placeholder)
storage_efficiency = 85.0 # Would calculate based on deduplication, etc.
return StorageStats(
total_size_mb=round(total_size_mb, 2),
average_memory_size=round(average_memory_size, 2),
largest_memories=largest_memories,
growth_trend=growth_trend,
storage_efficiency=storage_efficiency
)
except Exception as e:
logger.error(f"Failed to get storage stats: {str(e)}")
raise HTTPException(status_code=500, detail=f"Failed to get storage stats: {str(e)}")
```
--------------------------------------------------------------------------------
/src/mcp_memory_service/config.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 Configuration
Environment Variables:
- MCP_MEMORY_STORAGE_BACKEND: Storage backend ('sqlite_vec', 'cloudflare', or 'hybrid')
- MCP_MEMORY_SQLITE_PATH: SQLite-vec database file path
- MCP_MEMORY_USE_ONNX: Use ONNX embeddings ('true'/'false')
Copyright (c) 2024 Heinrich Krupp
Licensed under the Apache License, Version 2.0
"""
import os
import sys
import secrets
from pathlib import Path
from typing import Optional
import time
import logging
# Load environment variables from .env file if it exists
try:
from dotenv import load_dotenv
env_file = Path(__file__).parent.parent.parent / ".env"
if env_file.exists():
load_dotenv(env_file)
logging.getLogger(__name__).info(f"Loaded environment from {env_file}")
except ImportError:
# dotenv not available, skip loading
pass
logger = logging.getLogger(__name__)
def safe_get_int_env(env_var: str, default: int, min_value: int = None, max_value: int = None) -> int:
"""
Safely parse an integer environment variable with validation and error handling.
Args:
env_var: Environment variable name
default: Default value if not set or invalid
min_value: Minimum allowed value (optional)
max_value: Maximum allowed value (optional)
Returns:
Parsed and validated integer value
Raises:
ValueError: If the value is outside the specified range
"""
env_value = os.getenv(env_var)
if not env_value:
return default
try:
value = int(env_value)
# Validate range if specified
if min_value is not None and value < min_value:
logger.error(f"Environment variable {env_var}={value} is below minimum {min_value}, using default {default}")
return default
if max_value is not None and value > max_value:
logger.error(f"Environment variable {env_var}={value} is above maximum {max_value}, using default {default}")
return default
logger.debug(f"Environment variable {env_var}={value} parsed successfully")
return value
except ValueError as e:
logger.error(f"Invalid integer value for {env_var}='{env_value}': {e}. Using default {default}")
return default
def safe_get_optional_int_env(env_var: str, default: Optional[int] = None, min_value: int = None, max_value: int = None, none_values: tuple = ('none', 'null', 'unlimited', '')) -> Optional[int]:
"""
Safely parse an optional integer environment variable with validation and error handling.
Args:
env_var: Environment variable name
default: Default value if not set or invalid (None for unlimited)
min_value: Minimum allowed value (optional)
max_value: Maximum allowed value (optional)
none_values: Tuple of string values that should be interpreted as None
Returns:
Parsed and validated integer value, or None if explicitly set to a none_value
"""
env_value = os.getenv(env_var)
if not env_value:
return default
# Check if value should be interpreted as None/unlimited
if env_value.lower().strip() in none_values:
return None
try:
value = int(env_value.strip())
# Validate range if specified
if min_value is not None and value < min_value:
logger.warning(f"Environment variable {env_var}={value} is below minimum {min_value}. Using default {default}")
return default
if max_value is not None and value > max_value:
logger.warning(f"Environment variable {env_var}={value} is above maximum {max_value}. Using default {default}")
return default
return value
except ValueError:
logger.warning(f"Invalid value for {env_var}='{env_value}'. Expected integer or {'/'.join(none_values)}. Using default {default}")
return default
def safe_get_bool_env(env_var: str, default: bool) -> bool:
"""
Safely parse a boolean environment variable with validation and error handling.
Args:
env_var: Environment variable name
default: Default value if not set or invalid
Returns:
Parsed boolean value
"""
env_value = os.getenv(env_var)
if not env_value:
return default
env_value_lower = env_value.lower().strip()
if env_value_lower in ('true', '1', 'yes', 'on', 'enabled'):
return True
elif env_value_lower in ('false', '0', 'no', 'off', 'disabled'):
return False
else:
logger.error(f"Invalid boolean value for {env_var}='{env_value}'. Expected true/false, 1/0, yes/no, on/off, enabled/disabled. Using default {default}")
return default
def validate_and_create_path(path: str) -> str:
"""Validate and create a directory path, ensuring it's writable.
This function ensures that the specified directory path exists and is writable.
It performs several checks and has a retry mechanism to handle potential race
conditions, especially when running in environments like Claude Desktop where
file system operations might be more restricted.
"""
try:
# Convert to absolute path and expand user directory if present (e.g. ~)
abs_path = os.path.abspath(os.path.expanduser(path))
logger.debug(f"Validating path: {abs_path}")
# Create directory and all parents if they don't exist
try:
os.makedirs(abs_path, exist_ok=True)
logger.debug(f"Created directory (or already exists): {abs_path}")
except Exception as e:
logger.error(f"Error creating directory {abs_path}: {str(e)}")
raise PermissionError(f"Cannot create directory {abs_path}: {str(e)}")
# Add small delay to prevent potential race conditions on macOS during initial write test
time.sleep(0.1)
# Verify that the path exists and is a directory
if not os.path.exists(abs_path):
logger.error(f"Path does not exist after creation attempt: {abs_path}")
raise PermissionError(f"Path does not exist: {abs_path}")
if not os.path.isdir(abs_path):
logger.error(f"Path is not a directory: {abs_path}")
raise PermissionError(f"Path is not a directory: {abs_path}")
# Write test with retry mechanism
max_retries = 3
retry_delay = 0.5
test_file = os.path.join(abs_path, '.write_test')
for attempt in range(max_retries):
try:
logger.debug(f"Testing write permissions (attempt {attempt+1}/{max_retries}): {test_file}")
with open(test_file, 'w') as f:
f.write('test')
if os.path.exists(test_file):
logger.debug(f"Successfully wrote test file: {test_file}")
os.remove(test_file)
logger.debug(f"Successfully removed test file: {test_file}")
logger.info(f"Directory {abs_path} is writable.")
return abs_path
else:
logger.warning(f"Test file was not created: {test_file}")
except Exception as e:
logger.warning(f"Error during write test (attempt {attempt+1}/{max_retries}): {str(e)}")
if attempt < max_retries - 1:
logger.debug(f"Retrying after {retry_delay}s...")
time.sleep(retry_delay)
else:
logger.error(f"All write test attempts failed for {abs_path}")
raise PermissionError(f"Directory {abs_path} is not writable: {str(e)}")
return abs_path
except Exception as e:
logger.error(f"Error validating path {path}: {str(e)}")
raise
# Determine base directory - prefer local over Cloud
def get_base_directory() -> str:
"""Get base directory for storage, with fallback options."""
# First choice: Environment variable
if base_dir := os.getenv('MCP_MEMORY_BASE_DIR'):
return validate_and_create_path(base_dir)
# Second choice: Local app data directory
home = str(Path.home())
if sys.platform == 'darwin': # macOS
base = os.path.join(home, 'Library', 'Application Support', 'mcp-memory')
elif sys.platform == 'win32': # Windows
base = os.path.join(os.getenv('LOCALAPPDATA', ''), 'mcp-memory')
else: # Linux and others
base = os.path.join(home, '.local', 'share', 'mcp-memory')
return validate_and_create_path(base)
# Initialize paths
try:
BASE_DIR = get_base_directory()
# Try multiple environment variable names for backups path
backups_path = None
for env_var in ['MCP_MEMORY_BACKUPS_PATH', 'mcpMemoryBackupsPath']:
if path := os.getenv(env_var):
backups_path = path
logger.info(f"Using {env_var}={path} for backups path")
break
# If no environment variable is set, use the default path
if not backups_path:
backups_path = os.path.join(BASE_DIR, 'backups')
logger.info(f"No backups path environment variable found, using default: {backups_path}")
BACKUPS_PATH = validate_and_create_path(backups_path)
# Print the final paths used
logger.info(f"Using backups path: {BACKUPS_PATH}")
except Exception as e:
logger.error(f"Fatal error initializing paths: {str(e)}")
sys.exit(1)
# Server settings
SERVER_NAME = "memory"
# Import version from main package for consistency
from . import __version__ as SERVER_VERSION
# Storage backend configuration
SUPPORTED_BACKENDS = ['sqlite_vec', 'sqlite-vec', 'cloudflare', 'hybrid']
STORAGE_BACKEND = os.getenv('MCP_MEMORY_STORAGE_BACKEND', 'sqlite_vec').lower()
# Normalize backend names (sqlite-vec -> sqlite_vec)
if STORAGE_BACKEND == 'sqlite-vec':
STORAGE_BACKEND = 'sqlite_vec'
# Validate backend selection
if STORAGE_BACKEND not in SUPPORTED_BACKENDS:
logger.warning(f"Unknown storage backend: {STORAGE_BACKEND}, falling back to sqlite_vec")
STORAGE_BACKEND = 'sqlite_vec'
logger.info(f"Using storage backend: {STORAGE_BACKEND}")
# =============================================================================
# Content Length Limits Configuration (v7.5.0+)
# =============================================================================
# Backend-specific content length limits based on embedding model constraints
# These limits prevent embedding failures and enable automatic content splitting
# Cloudflare: BGE-base-en-v1.5 model has 512 token limit
# Using 800 characters as safe limit (~400 tokens with overhead)
CLOUDFLARE_MAX_CONTENT_LENGTH = safe_get_int_env(
'MCP_CLOUDFLARE_MAX_CONTENT_LENGTH',
default=800,
min_value=100,
max_value=10000
)
# SQLite-vec: No inherent limit (local storage)
# Set to None for unlimited, or configure via environment variable
SQLITEVEC_MAX_CONTENT_LENGTH = safe_get_optional_int_env(
'MCP_SQLITEVEC_MAX_CONTENT_LENGTH',
default=None,
min_value=100,
max_value=10000
)
# Hybrid: Constrained by Cloudflare secondary storage (configurable)
HYBRID_MAX_CONTENT_LENGTH = safe_get_int_env(
'MCP_HYBRID_MAX_CONTENT_LENGTH',
default=CLOUDFLARE_MAX_CONTENT_LENGTH,
min_value=100,
max_value=10000
)
# Enable automatic content splitting when limits are exceeded
ENABLE_AUTO_SPLIT = safe_get_bool_env('MCP_ENABLE_AUTO_SPLIT', default=True)
# Content splitting configuration
CONTENT_SPLIT_OVERLAP = safe_get_int_env(
'MCP_CONTENT_SPLIT_OVERLAP',
default=50,
min_value=0,
max_value=500
)
CONTENT_PRESERVE_BOUNDARIES = safe_get_bool_env('MCP_CONTENT_PRESERVE_BOUNDARIES', default=True)
logger.info(f"Content length limits - Cloudflare: {CLOUDFLARE_MAX_CONTENT_LENGTH}, "
f"SQLite-vec: {'unlimited' if SQLITEVEC_MAX_CONTENT_LENGTH is None else SQLITEVEC_MAX_CONTENT_LENGTH}, "
f"Auto-split: {ENABLE_AUTO_SPLIT}")
# =============================================================================
# End Content Length Limits Configuration
# =============================================================================
# SQLite-vec specific configuration (also needed for hybrid backend)
if STORAGE_BACKEND == 'sqlite_vec' or STORAGE_BACKEND == 'hybrid':
# Try multiple environment variable names for SQLite-vec path
sqlite_vec_path = None
for env_var in ['MCP_MEMORY_SQLITE_PATH', 'MCP_MEMORY_SQLITEVEC_PATH']:
if path := os.getenv(env_var):
sqlite_vec_path = path
logger.info(f"Using {env_var}={path} for SQLite-vec database path")
break
# If no environment variable is set, use the default path
if not sqlite_vec_path:
sqlite_vec_path = os.path.join(BASE_DIR, 'sqlite_vec.db')
logger.info(f"No SQLite-vec path environment variable found, using default: {sqlite_vec_path}")
# Ensure directory exists for SQLite database
sqlite_dir = os.path.dirname(sqlite_vec_path)
if sqlite_dir:
os.makedirs(sqlite_dir, exist_ok=True)
SQLITE_VEC_PATH = sqlite_vec_path
logger.info(f"Using SQLite-vec database path: {SQLITE_VEC_PATH}")
else:
SQLITE_VEC_PATH = None
# ONNX Configuration
USE_ONNX = os.getenv('MCP_MEMORY_USE_ONNX', '').lower() in ('1', 'true', 'yes')
if USE_ONNX:
logger.info("ONNX embeddings enabled - using PyTorch-free embedding generation")
# ONNX model cache directory
ONNX_MODEL_CACHE = os.path.join(BASE_DIR, 'onnx_models')
os.makedirs(ONNX_MODEL_CACHE, exist_ok=True)
# Cloudflare specific configuration (also needed for hybrid backend)
if STORAGE_BACKEND == 'cloudflare' or STORAGE_BACKEND == 'hybrid':
# Required Cloudflare settings
CLOUDFLARE_API_TOKEN = os.getenv('CLOUDFLARE_API_TOKEN')
CLOUDFLARE_ACCOUNT_ID = os.getenv('CLOUDFLARE_ACCOUNT_ID')
CLOUDFLARE_VECTORIZE_INDEX = os.getenv('CLOUDFLARE_VECTORIZE_INDEX')
CLOUDFLARE_D1_DATABASE_ID = os.getenv('CLOUDFLARE_D1_DATABASE_ID')
# Optional Cloudflare settings
CLOUDFLARE_R2_BUCKET = os.getenv('CLOUDFLARE_R2_BUCKET') # For large content storage
CLOUDFLARE_EMBEDDING_MODEL = os.getenv('CLOUDFLARE_EMBEDDING_MODEL', '@cf/baai/bge-base-en-v1.5')
CLOUDFLARE_LARGE_CONTENT_THRESHOLD = int(os.getenv('CLOUDFLARE_LARGE_CONTENT_THRESHOLD', '1048576')) # 1MB
CLOUDFLARE_MAX_RETRIES = int(os.getenv('CLOUDFLARE_MAX_RETRIES', '3'))
CLOUDFLARE_BASE_DELAY = float(os.getenv('CLOUDFLARE_BASE_DELAY', '1.0'))
# Validate required settings
missing_vars = []
if not CLOUDFLARE_API_TOKEN:
missing_vars.append('CLOUDFLARE_API_TOKEN')
if not CLOUDFLARE_ACCOUNT_ID:
missing_vars.append('CLOUDFLARE_ACCOUNT_ID')
if not CLOUDFLARE_VECTORIZE_INDEX:
missing_vars.append('CLOUDFLARE_VECTORIZE_INDEX')
if not CLOUDFLARE_D1_DATABASE_ID:
missing_vars.append('CLOUDFLARE_D1_DATABASE_ID')
if missing_vars:
logger.error(f"Missing required environment variables for Cloudflare backend: {', '.join(missing_vars)}")
logger.error("Please set the required variables or switch to a different backend")
sys.exit(1)
logger.info(f"Using Cloudflare backend with:")
logger.info(f" Vectorize Index: {CLOUDFLARE_VECTORIZE_INDEX}")
logger.info(f" D1 Database: {CLOUDFLARE_D1_DATABASE_ID}")
logger.info(f" R2 Bucket: {CLOUDFLARE_R2_BUCKET or 'Not configured'}")
logger.info(f" Embedding Model: {CLOUDFLARE_EMBEDDING_MODEL}")
logger.info(f" Large Content Threshold: {CLOUDFLARE_LARGE_CONTENT_THRESHOLD} bytes")
else:
# Set Cloudflare variables to None when not using Cloudflare backend
CLOUDFLARE_API_TOKEN = None
CLOUDFLARE_ACCOUNT_ID = None
CLOUDFLARE_VECTORIZE_INDEX = None
CLOUDFLARE_D1_DATABASE_ID = None
CLOUDFLARE_R2_BUCKET = None
CLOUDFLARE_EMBEDDING_MODEL = None
CLOUDFLARE_LARGE_CONTENT_THRESHOLD = None
CLOUDFLARE_MAX_RETRIES = None
CLOUDFLARE_BASE_DELAY = None
# Hybrid backend specific configuration
if STORAGE_BACKEND == 'hybrid':
# Sync service configuration
HYBRID_SYNC_INTERVAL = int(os.getenv('MCP_HYBRID_SYNC_INTERVAL', '300')) # 5 minutes default
HYBRID_BATCH_SIZE = int(os.getenv('MCP_HYBRID_BATCH_SIZE', '50'))
HYBRID_MAX_QUEUE_SIZE = int(os.getenv('MCP_HYBRID_MAX_QUEUE_SIZE', '1000'))
HYBRID_MAX_RETRIES = int(os.getenv('MCP_HYBRID_MAX_RETRIES', '3'))
# Sync ownership control (v8.27.0+) - Prevents duplicate sync queues
# Values: "http" (HTTP server only), "mcp" (MCP server only), "both" (both servers sync)
# Recommended: "http" to avoid duplicate sync work
HYBRID_SYNC_OWNER = os.getenv('MCP_HYBRID_SYNC_OWNER', 'both').lower()
# Performance tuning
HYBRID_ENABLE_HEALTH_CHECKS = os.getenv('MCP_HYBRID_ENABLE_HEALTH_CHECKS', 'true').lower() == 'true'
HYBRID_HEALTH_CHECK_INTERVAL = int(os.getenv('MCP_HYBRID_HEALTH_CHECK_INTERVAL', '60')) # 1 minute
HYBRID_SYNC_ON_STARTUP = os.getenv('MCP_HYBRID_SYNC_ON_STARTUP', 'true').lower() == 'true'
# Drift detection and metadata sync (v8.25.0+)
HYBRID_SYNC_UPDATES = os.getenv('MCP_HYBRID_SYNC_UPDATES', 'true').lower() == 'true'
HYBRID_DRIFT_CHECK_INTERVAL = int(os.getenv('MCP_HYBRID_DRIFT_CHECK_INTERVAL', '3600')) # 1 hour default
HYBRID_DRIFT_BATCH_SIZE = int(os.getenv('MCP_HYBRID_DRIFT_BATCH_SIZE', '100'))
# Initial sync behavior tuning (v7.5.4+)
HYBRID_MAX_EMPTY_BATCHES = safe_get_int_env('MCP_HYBRID_MAX_EMPTY_BATCHES', 20, min_value=1) # Stop after N batches without new syncs
HYBRID_MIN_CHECK_COUNT = safe_get_int_env('MCP_HYBRID_MIN_CHECK_COUNT', 1000, min_value=1) # Minimum memories to check before early stop
# Fallback behavior
HYBRID_FALLBACK_TO_PRIMARY = os.getenv('MCP_HYBRID_FALLBACK_TO_PRIMARY', 'true').lower() == 'true'
HYBRID_WARN_ON_SECONDARY_FAILURE = os.getenv('MCP_HYBRID_WARN_ON_SECONDARY_FAILURE', 'true').lower() == 'true'
logger.info(f"Hybrid storage configuration: sync_interval={HYBRID_SYNC_INTERVAL}s, batch_size={HYBRID_BATCH_SIZE}")
# Cloudflare Service Limits (for validation and monitoring)
CLOUDFLARE_D1_MAX_SIZE_GB = 10 # D1 database hard limit
CLOUDFLARE_VECTORIZE_MAX_VECTORS = 5_000_000 # Maximum vectors per index
CLOUDFLARE_MAX_METADATA_SIZE_KB = 10 # Maximum metadata size per vector
CLOUDFLARE_MAX_FILTER_SIZE_BYTES = 2048 # Maximum filter query size
CLOUDFLARE_MAX_STRING_INDEX_SIZE_BYTES = 64 # Maximum indexed string size
CLOUDFLARE_BATCH_INSERT_LIMIT = 200_000 # Maximum batch insert size
# Limit warning thresholds (percentage)
CLOUDFLARE_WARNING_THRESHOLD_PERCENT = 80 # Warn at 80% capacity
CLOUDFLARE_CRITICAL_THRESHOLD_PERCENT = 95 # Critical at 95% capacity
# Validate Cloudflare configuration for hybrid mode
if not (CLOUDFLARE_API_TOKEN and CLOUDFLARE_ACCOUNT_ID and CLOUDFLARE_VECTORIZE_INDEX and CLOUDFLARE_D1_DATABASE_ID):
logger.warning("Hybrid mode requires Cloudflare configuration. Missing required variables:")
if not CLOUDFLARE_API_TOKEN:
logger.warning(" - CLOUDFLARE_API_TOKEN")
if not CLOUDFLARE_ACCOUNT_ID:
logger.warning(" - CLOUDFLARE_ACCOUNT_ID")
if not CLOUDFLARE_VECTORIZE_INDEX:
logger.warning(" - CLOUDFLARE_VECTORIZE_INDEX")
if not CLOUDFLARE_D1_DATABASE_ID:
logger.warning(" - CLOUDFLARE_D1_DATABASE_ID")
logger.warning("Hybrid mode will operate in SQLite-only mode until Cloudflare is configured")
else:
# Set hybrid-specific variables to None when not using hybrid backend
HYBRID_SYNC_INTERVAL = None
HYBRID_BATCH_SIZE = None
HYBRID_MAX_QUEUE_SIZE = None
HYBRID_MAX_RETRIES = None
HYBRID_SYNC_OWNER = None
HYBRID_ENABLE_HEALTH_CHECKS = None
HYBRID_HEALTH_CHECK_INTERVAL = None
HYBRID_SYNC_ON_STARTUP = None
HYBRID_SYNC_UPDATES = None
HYBRID_DRIFT_CHECK_INTERVAL = None
HYBRID_DRIFT_BATCH_SIZE = None
HYBRID_MAX_EMPTY_BATCHES = None
HYBRID_MIN_CHECK_COUNT = None
HYBRID_FALLBACK_TO_PRIMARY = None
HYBRID_WARN_ON_SECONDARY_FAILURE = None
# Also set limit constants to None
CLOUDFLARE_D1_MAX_SIZE_GB = None
CLOUDFLARE_VECTORIZE_MAX_VECTORS = None
CLOUDFLARE_MAX_METADATA_SIZE_KB = None
CLOUDFLARE_MAX_FILTER_SIZE_BYTES = None
CLOUDFLARE_MAX_STRING_INDEX_SIZE_BYTES = None
CLOUDFLARE_BATCH_INSERT_LIMIT = None
CLOUDFLARE_WARNING_THRESHOLD_PERCENT = None
CLOUDFLARE_CRITICAL_THRESHOLD_PERCENT = None
# HTTP Server Configuration
HTTP_ENABLED = os.getenv('MCP_HTTP_ENABLED', 'false').lower() == 'true'
HTTP_PORT = safe_get_int_env('MCP_HTTP_PORT', 8000, min_value=1024, max_value=65535) # Non-privileged ports only
HTTP_HOST = os.getenv('MCP_HTTP_HOST', '0.0.0.0')
CORS_ORIGINS = os.getenv('MCP_CORS_ORIGINS', '*').split(',')
SSE_HEARTBEAT_INTERVAL = safe_get_int_env('MCP_SSE_HEARTBEAT', 30, min_value=5, max_value=300) # 5 seconds to 5 minutes
API_KEY = os.getenv('MCP_API_KEY', None) # Optional authentication
# HTTPS Configuration
HTTPS_ENABLED = os.getenv('MCP_HTTPS_ENABLED', 'false').lower() == 'true'
SSL_CERT_FILE = os.getenv('MCP_SSL_CERT_FILE', None)
SSL_KEY_FILE = os.getenv('MCP_SSL_KEY_FILE', None)
# mDNS Service Discovery Configuration
MDNS_ENABLED = os.getenv('MCP_MDNS_ENABLED', 'true').lower() == 'true'
MDNS_SERVICE_NAME = os.getenv('MCP_MDNS_SERVICE_NAME', 'MCP Memory Service')
MDNS_SERVICE_TYPE = os.getenv('MCP_MDNS_SERVICE_TYPE', '_mcp-memory._tcp.local.')
MDNS_DISCOVERY_TIMEOUT = int(os.getenv('MCP_MDNS_DISCOVERY_TIMEOUT', '5'))
# Database path for HTTP interface (use SQLite-vec by default)
if (STORAGE_BACKEND in ['sqlite_vec', 'hybrid']) and SQLITE_VEC_PATH:
DATABASE_PATH = SQLITE_VEC_PATH
else:
# Fallback to a default SQLite-vec path for HTTP interface
DATABASE_PATH = os.path.join(BASE_DIR, 'memory_http.db')
# Embedding model configuration
EMBEDDING_MODEL_NAME = os.getenv('MCP_EMBEDDING_MODEL', 'all-MiniLM-L6-v2')
# =============================================================================
# Document Processing Configuration (Semtools Integration)
# =============================================================================
# Semtools configuration for enhanced document parsing
# LlamaParse API key for advanced OCR and table extraction
LLAMAPARSE_API_KEY = os.getenv('LLAMAPARSE_API_KEY', None)
# Document chunking configuration
DOCUMENT_CHUNK_SIZE = safe_get_int_env('MCP_DOCUMENT_CHUNK_SIZE', 1000, min_value=100, max_value=10000)
DOCUMENT_CHUNK_OVERLAP = safe_get_int_env('MCP_DOCUMENT_CHUNK_OVERLAP', 200, min_value=0, max_value=1000)
# Log semtools configuration
if LLAMAPARSE_API_KEY:
logger.info("LlamaParse API key configured - enhanced document parsing available")
else:
logger.debug("LlamaParse API key not set - semtools will use basic parsing mode")
logger.info(f"Document chunking: size={DOCUMENT_CHUNK_SIZE}, overlap={DOCUMENT_CHUNK_OVERLAP}")
# =============================================================================
# End Document Processing Configuration
# =============================================================================
# =============================================================================
# Automatic Backup Configuration
# =============================================================================
BACKUP_ENABLED = safe_get_bool_env('MCP_BACKUP_ENABLED', True)
BACKUP_INTERVAL = os.getenv('MCP_BACKUP_INTERVAL', 'daily').lower() # 'hourly', 'daily', 'weekly'
BACKUP_RETENTION = safe_get_int_env('MCP_BACKUP_RETENTION', 7, min_value=1, max_value=365) # days
BACKUP_MAX_COUNT = safe_get_int_env('MCP_BACKUP_MAX_COUNT', 10, min_value=1, max_value=100) # max backups to keep
# Validate backup interval
if BACKUP_INTERVAL not in ['hourly', 'daily', 'weekly']:
logger.warning(f"Invalid backup interval: {BACKUP_INTERVAL}, falling back to 'daily'")
BACKUP_INTERVAL = 'daily'
logger.info(f"Backup configuration: enabled={BACKUP_ENABLED}, interval={BACKUP_INTERVAL}, retention={BACKUP_RETENTION} days")
# =============================================================================
# End Automatic Backup Configuration
# =============================================================================
# Dream-inspired consolidation configuration
CONSOLIDATION_ENABLED = os.getenv('MCP_CONSOLIDATION_ENABLED', 'false').lower() == 'true'
# Machine identification configuration
INCLUDE_HOSTNAME = os.getenv('MCP_MEMORY_INCLUDE_HOSTNAME', 'false').lower() == 'true'
# Consolidation archive location
consolidation_archive_path = None
for env_var in ['MCP_CONSOLIDATION_ARCHIVE_PATH', 'MCP_MEMORY_ARCHIVE_PATH']:
if path := os.getenv(env_var):
consolidation_archive_path = path
logger.info(f"Using {env_var}={path} for consolidation archive path")
break
if not consolidation_archive_path:
consolidation_archive_path = os.path.join(BASE_DIR, 'consolidation_archive')
logger.info(f"No consolidation archive path environment variable found, using default: {consolidation_archive_path}")
try:
CONSOLIDATION_ARCHIVE_PATH = validate_and_create_path(consolidation_archive_path)
logger.info(f"Using consolidation archive path: {CONSOLIDATION_ARCHIVE_PATH}")
except Exception as e:
logger.error(f"Error creating consolidation archive path: {e}")
CONSOLIDATION_ARCHIVE_PATH = None
# Consolidation settings with environment variable overrides
CONSOLIDATION_CONFIG = {
# Decay settings
'decay_enabled': os.getenv('MCP_DECAY_ENABLED', 'true').lower() == 'true',
'retention_periods': {
'critical': int(os.getenv('MCP_RETENTION_CRITICAL', '365')),
'reference': int(os.getenv('MCP_RETENTION_REFERENCE', '180')),
'standard': int(os.getenv('MCP_RETENTION_STANDARD', '30')),
'temporary': int(os.getenv('MCP_RETENTION_TEMPORARY', '7'))
},
# Association settings
'associations_enabled': os.getenv('MCP_ASSOCIATIONS_ENABLED', 'true').lower() == 'true',
'min_similarity': float(os.getenv('MCP_ASSOCIATION_MIN_SIMILARITY', '0.3')),
'max_similarity': float(os.getenv('MCP_ASSOCIATION_MAX_SIMILARITY', '0.7')),
'max_pairs_per_run': int(os.getenv('MCP_ASSOCIATION_MAX_PAIRS', '100')),
# Clustering settings
'clustering_enabled': os.getenv('MCP_CLUSTERING_ENABLED', 'true').lower() == 'true',
'min_cluster_size': int(os.getenv('MCP_CLUSTERING_MIN_SIZE', '5')),
'clustering_algorithm': os.getenv('MCP_CLUSTERING_ALGORITHM', 'dbscan'), # 'dbscan', 'hierarchical', 'simple'
# Compression settings
'compression_enabled': os.getenv('MCP_COMPRESSION_ENABLED', 'true').lower() == 'true',
'max_summary_length': int(os.getenv('MCP_COMPRESSION_MAX_LENGTH', '500')),
'preserve_originals': os.getenv('MCP_COMPRESSION_PRESERVE_ORIGINALS', 'true').lower() == 'true',
# Forgetting settings
'forgetting_enabled': os.getenv('MCP_FORGETTING_ENABLED', 'true').lower() == 'true',
'relevance_threshold': float(os.getenv('MCP_FORGETTING_RELEVANCE_THRESHOLD', '0.1')),
'access_threshold_days': int(os.getenv('MCP_FORGETTING_ACCESS_THRESHOLD', '90')),
'archive_location': CONSOLIDATION_ARCHIVE_PATH,
# Incremental consolidation settings
'batch_size': int(os.getenv('MCP_CONSOLIDATION_BATCH_SIZE', '500')),
'incremental_mode': os.getenv('MCP_CONSOLIDATION_INCREMENTAL', 'true').lower() == 'true'
}
# Consolidation scheduling settings (for APScheduler integration)
CONSOLIDATION_SCHEDULE = {
'daily': os.getenv('MCP_SCHEDULE_DAILY', '02:00'), # 2 AM daily
'weekly': os.getenv('MCP_SCHEDULE_WEEKLY', 'SUN 03:00'), # 3 AM on Sundays
'monthly': os.getenv('MCP_SCHEDULE_MONTHLY', '01 04:00'), # 4 AM on 1st of month
'quarterly': os.getenv('MCP_SCHEDULE_QUARTERLY', 'disabled'), # Disabled by default
'yearly': os.getenv('MCP_SCHEDULE_YEARLY', 'disabled') # Disabled by default
}
logger.info(f"Consolidation enabled: {CONSOLIDATION_ENABLED}")
if CONSOLIDATION_ENABLED:
logger.info(f"Consolidation configuration: {CONSOLIDATION_CONFIG}")
logger.info(f"Consolidation schedule: {CONSOLIDATION_SCHEDULE}")
# OAuth 2.1 Configuration
OAUTH_ENABLED = safe_get_bool_env('MCP_OAUTH_ENABLED', True)
# RSA key pair configuration for JWT signing (RS256)
# Private key for signing tokens
OAUTH_PRIVATE_KEY = os.getenv('MCP_OAUTH_PRIVATE_KEY')
# Public key for verifying tokens
OAUTH_PUBLIC_KEY = os.getenv('MCP_OAUTH_PUBLIC_KEY')
# Generate RSA key pair if not provided
if not OAUTH_PRIVATE_KEY or not OAUTH_PUBLIC_KEY:
try:
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.hazmat.backends import default_backend
# Generate 2048-bit RSA key pair
private_key = rsa.generate_private_key(
public_exponent=65537,
key_size=2048,
backend=default_backend()
)
# Serialize private key to PEM format
OAUTH_PRIVATE_KEY = private_key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.PKCS8,
encryption_algorithm=serialization.NoEncryption()
).decode('utf-8')
# Serialize public key to PEM format
public_key = private_key.public_key()
OAUTH_PUBLIC_KEY = public_key.public_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PublicFormat.SubjectPublicKeyInfo
).decode('utf-8')
logger.info("Generated RSA key pair for OAuth JWT signing (set MCP_OAUTH_PRIVATE_KEY and MCP_OAUTH_PUBLIC_KEY for persistence)")
except ImportError:
logger.warning("cryptography package not available, falling back to HS256 symmetric key")
# Fallback to symmetric key for HS256
OAUTH_SECRET_KEY = os.getenv('MCP_OAUTH_SECRET_KEY')
if not OAUTH_SECRET_KEY:
OAUTH_SECRET_KEY = secrets.token_urlsafe(32)
logger.info("Generated random OAuth secret key (set MCP_OAUTH_SECRET_KEY for persistence)")
OAUTH_PRIVATE_KEY = None
OAUTH_PUBLIC_KEY = None
# JWT algorithm and key helper functions
def get_jwt_algorithm() -> str:
"""Get the JWT algorithm to use based on available keys."""
return "RS256" if OAUTH_PRIVATE_KEY and OAUTH_PUBLIC_KEY else "HS256"
def get_jwt_signing_key() -> str:
"""Get the appropriate key for JWT signing."""
if OAUTH_PRIVATE_KEY and OAUTH_PUBLIC_KEY:
return OAUTH_PRIVATE_KEY
elif hasattr(globals(), 'OAUTH_SECRET_KEY'):
return OAUTH_SECRET_KEY
else:
raise ValueError("No JWT signing key available")
def get_jwt_verification_key() -> str:
"""Get the appropriate key for JWT verification."""
if OAUTH_PRIVATE_KEY and OAUTH_PUBLIC_KEY:
return OAUTH_PUBLIC_KEY
elif hasattr(globals(), 'OAUTH_SECRET_KEY'):
return OAUTH_SECRET_KEY
else:
raise ValueError("No JWT verification key available")
def validate_oauth_configuration() -> None:
"""
Validate OAuth configuration at startup.
Raises:
ValueError: If OAuth configuration is invalid
"""
if not OAUTH_ENABLED:
logger.info("OAuth validation skipped: OAuth disabled")
return
errors = []
warnings = []
# Validate issuer URL
if not OAUTH_ISSUER:
errors.append("OAuth issuer URL is not configured")
elif not OAUTH_ISSUER.startswith(('http://', 'https://')):
errors.append(f"OAuth issuer URL must start with http:// or https://: {OAUTH_ISSUER}")
# Validate JWT configuration
try:
algorithm = get_jwt_algorithm()
logger.debug(f"OAuth JWT algorithm validation: {algorithm}")
# Test key access
signing_key = get_jwt_signing_key()
verification_key = get_jwt_verification_key()
if algorithm == "RS256":
if not OAUTH_PRIVATE_KEY or not OAUTH_PUBLIC_KEY:
errors.append("RS256 algorithm selected but RSA keys are missing")
elif len(signing_key) < 100: # Basic length check for PEM format
warnings.append("RSA private key appears to be too short")
elif algorithm == "HS256":
if not hasattr(globals(), 'OAUTH_SECRET_KEY') or not OAUTH_SECRET_KEY:
errors.append("HS256 algorithm selected but secret key is missing")
elif len(signing_key) < 32: # Basic length check for symmetric key
warnings.append("OAuth secret key is shorter than recommended (32+ characters)")
except Exception as e:
errors.append(f"JWT configuration error: {e}")
# Validate token expiry settings
if OAUTH_ACCESS_TOKEN_EXPIRE_MINUTES <= 0:
errors.append(f"OAuth access token expiry must be positive: {OAUTH_ACCESS_TOKEN_EXPIRE_MINUTES}")
elif OAUTH_ACCESS_TOKEN_EXPIRE_MINUTES > 1440: # 24 hours
warnings.append(f"OAuth access token expiry is very long: {OAUTH_ACCESS_TOKEN_EXPIRE_MINUTES} minutes")
if OAUTH_AUTHORIZATION_CODE_EXPIRE_MINUTES <= 0:
errors.append(f"OAuth authorization code expiry must be positive: {OAUTH_AUTHORIZATION_CODE_EXPIRE_MINUTES}")
elif OAUTH_AUTHORIZATION_CODE_EXPIRE_MINUTES > 60: # 1 hour
warnings.append(f"OAuth authorization code expiry is longer than recommended: {OAUTH_AUTHORIZATION_CODE_EXPIRE_MINUTES} minutes")
# Validate security settings
if "localhost" in OAUTH_ISSUER or "127.0.0.1" in OAUTH_ISSUER:
if not os.getenv('MCP_OAUTH_ISSUER'):
warnings.append("OAuth issuer contains localhost/127.0.0.1. For production, set MCP_OAUTH_ISSUER to external URL")
# Check for production readiness
if ALLOW_ANONYMOUS_ACCESS:
warnings.append("Anonymous access is enabled - consider disabling for production")
# Check for insecure transport in production
if OAUTH_ISSUER.startswith('http://') and not ("localhost" in OAUTH_ISSUER or "127.0.0.1" in OAUTH_ISSUER):
warnings.append("OAuth issuer uses HTTP (non-encrypted) transport - use HTTPS for production")
# Check for weak algorithm in production environments
if get_jwt_algorithm() == "HS256" and not os.getenv('MCP_OAUTH_SECRET_KEY'):
warnings.append("Using auto-generated HS256 secret key - set MCP_OAUTH_SECRET_KEY for production")
# Log validation results
if errors:
error_msg = "OAuth configuration validation failed:\n" + "\n".join(f" - {err}" for err in errors)
logger.error(error_msg)
raise ValueError(f"Invalid OAuth configuration: {'; '.join(errors)}")
if warnings:
warning_msg = "OAuth configuration warnings:\n" + "\n".join(f" - {warn}" for warn in warnings)
logger.warning(warning_msg)
logger.info("OAuth configuration validation successful")
# OAuth server configuration
def get_oauth_issuer() -> str:
"""
Get the OAuth issuer URL based on server configuration.
For reverse proxy deployments, set MCP_OAUTH_ISSUER environment variable
to override auto-detection (e.g., "https://api.example.com").
This ensures OAuth discovery endpoints return the correct external URLs
that clients can actually reach, rather than internal server addresses.
"""
scheme = "https" if HTTPS_ENABLED else "http"
host = "localhost" if HTTP_HOST == "0.0.0.0" else HTTP_HOST
# Only include port if it's not the standard port for the scheme
if (scheme == "https" and HTTP_PORT != 443) or (scheme == "http" and HTTP_PORT != 80):
return f"{scheme}://{host}:{HTTP_PORT}"
else:
return f"{scheme}://{host}"
# OAuth issuer URL - CRITICAL for reverse proxy deployments
# Production: Set MCP_OAUTH_ISSUER to external URL (e.g., "https://api.example.com")
# Development: Auto-detects from server configuration
OAUTH_ISSUER = os.getenv('MCP_OAUTH_ISSUER') or get_oauth_issuer()
# OAuth token configuration
OAUTH_ACCESS_TOKEN_EXPIRE_MINUTES = safe_get_int_env('MCP_OAUTH_ACCESS_TOKEN_EXPIRE_MINUTES', 60, min_value=1, max_value=1440) # 1 minute to 24 hours
OAUTH_AUTHORIZATION_CODE_EXPIRE_MINUTES = safe_get_int_env('MCP_OAUTH_AUTHORIZATION_CODE_EXPIRE_MINUTES', 10, min_value=1, max_value=60) # 1 minute to 1 hour
# OAuth security configuration
ALLOW_ANONYMOUS_ACCESS = safe_get_bool_env('MCP_ALLOW_ANONYMOUS_ACCESS', False)
logger.info(f"OAuth enabled: {OAUTH_ENABLED}")
if OAUTH_ENABLED:
logger.info(f"OAuth issuer: {OAUTH_ISSUER}")
logger.info(f"OAuth JWT algorithm: {get_jwt_algorithm()}")
logger.info(f"OAuth access token expiry: {OAUTH_ACCESS_TOKEN_EXPIRE_MINUTES} minutes")
logger.info(f"Anonymous access allowed: {ALLOW_ANONYMOUS_ACCESS}")
# Warn about potential reverse proxy configuration issues
if not os.getenv('MCP_OAUTH_ISSUER') and ("localhost" in OAUTH_ISSUER or "127.0.0.1" in OAUTH_ISSUER):
logger.warning(
"OAuth issuer contains localhost/127.0.0.1. For reverse proxy deployments, "
"set MCP_OAUTH_ISSUER to the external URL (e.g., 'https://api.example.com')"
)
# Validate OAuth configuration at startup
try:
validate_oauth_configuration()
except ValueError as e:
logger.error(f"OAuth configuration validation failed: {e}")
raise
```
--------------------------------------------------------------------------------
/docs/research/code-execution-interface-implementation.md:
--------------------------------------------------------------------------------
```markdown
# Code Execution Interface Implementation Research
## Issue #206: 90-95% Token Reduction Strategy
**Research Date:** November 6, 2025
**Target:** Implement Python code API for mcp-memory-service to reduce token consumption by 90-95%
**Current Status:** Research & Architecture Phase
---
## Executive Summary
This document provides comprehensive research findings and implementation recommendations for transitioning mcp-memory-service from tool-based MCP interactions to a direct code execution interface. Based on industry best practices, real-world examples, and analysis of current codebase architecture, this research identifies concrete strategies to achieve the target 90-95% token reduction.
### Key Findings
1. **Token Reduction Potential Validated**: Research confirms 75-90% reductions are achievable through code execution interfaces
2. **Industry Momentum**: Anthropic's November 2025 announcement of MCP code execution aligns with our proposal
3. **Proven Patterns**: Multiple successful implementations exist (python-interpreter MCP, CodeAgents framework)
4. **Architecture Ready**: Current codebase structure well-positioned for gradual migration
---
## 1. Current State Analysis
### Token Consumption Breakdown
**Current Architecture:**
- **33 MCP tools** generating ~4,125 tokens per interaction
- **Document ingestion:** 57,400 tokens for 50 PDFs
- **Session hooks:** 3,600-9,600 tokens per session start
- **Tool definitions:** Loaded upfront into context window
**Example: Session-Start Hook**
```javascript
// Current approach: MCP tool invocation
// Each tool call includes full schema (~125 tokens/tool)
await memoryClient.callTool('retrieve_memory', {
query: gitContext.query,
limit: 8,
similarity_threshold: 0.6
});
// Result includes full Memory objects with all fields:
// - content, content_hash, tags, memory_type, metadata
// - embedding (768 floats for all-MiniLM-L6-v2)
// - created_at, created_at_iso, updated_at, updated_at_iso
// - Total: ~500-800 tokens per memory
```
### Codebase Architecture Analysis
**Strengths:**
- ✅ Clean separation of concerns (storage/models/web layers)
- ✅ Abstract base class (`MemoryStorage`) for consistent interface
- ✅ Async/await throughout for performance
- ✅ Strong type hints (Python 3.10+)
- ✅ Multiple storage backends (SQLite-Vec, Cloudflare, Hybrid)
- ✅ Existing HTTP client (`HTTPClientStorage`) demonstrates remote access pattern
**Current Entry Points:**
```python
# Public API (src/mcp_memory_service/__init__.py)
__all__ = [
'Memory',
'MemoryQueryResult',
'MemoryStorage',
'generate_content_hash'
]
```
**Infrastructure Files:**
- `src/mcp_memory_service/server.py` - 3,721 lines (MCP server implementation)
- `src/mcp_memory_service/storage/base.py` - Abstract interface with 20+ methods
- `src/mcp_memory_service/models/memory.py` - Memory data model with timestamp handling
- `src/mcp_memory_service/web/api/mcp.py` - MCP protocol endpoints
---
## 2. Best Practices from Research
### 2.1 Token-Efficient API Design
**Key Principles from CodeAgents Framework:**
1. **Codified Structures Over Natural Language**
- Use pseudocode/typed structures instead of verbose descriptions
- Control structures (loops, conditionals) reduce repeated instructions
- Typed variables eliminate ambiguity and error-prone parsing
2. **Modular Subroutines**
- Encapsulate common patterns in reusable functions
- Single import replaces repeated tool definitions
- Function signatures convey requirements compactly
3. **Compact Result Types**
- Return only essential data fields
- Use structured types (namedtuple, TypedDict) for clarity
- Avoid redundant metadata in response payloads
**Anthropic's MCP Code Execution Approach (Nov 2025):**
```python
# Before: Tool invocation (125 tokens for schema + 500 tokens for result)
result = await call_tool("retrieve_memory", {
"query": "recent architecture decisions",
"limit": 5,
"similarity_threshold": 0.7
})
# After: Code execution (5 tokens import + 20 tokens call)
from mcp_memory_service.api import search
results = search("recent architecture decisions", limit=5)
```
### 2.2 Python Data Structure Performance
**Benchmark Results (from research):**
| Structure | Creation Speed | Access Speed | Memory | Immutability | Type Safety |
|-----------|---------------|--------------|---------|--------------|-------------|
| `dict` | Fastest | Fast | High | No | Runtime only |
| `dataclass` | 8% faster than NamedTuple | Fast | Medium (with `__slots__`) | Optional | Static + Runtime |
| `NamedTuple` | Fast | Fastest (C-based) | Low | Yes | Static + Runtime |
| `TypedDict` | Same as dict | Same as dict | High | No | Static only |
**Recommendation for Token Efficiency:**
```python
from typing import NamedTuple
class CompactMemory(NamedTuple):
"""Minimal memory representation for hooks (50-80 tokens vs 500-800)."""
hash: str # 8 chars
content: str # First 200 chars
tags: tuple[str] # Tag list
created: float # Unix timestamp
score: float # Relevance score
```
**Benefits:**
- ✅ **60-90% size reduction**: Essential fields only
- ✅ **Immutable**: Safer for concurrent access in hooks
- ✅ **Type-safe**: Static checking with mypy/pyright
- ✅ **Fast**: C-based tuple operations
- ✅ **Readable**: Named field access (`memory.hash` not `memory[0]`)
### 2.3 Migration Strategy Best Practices
**Lessons from Python 2→3 Migrations:**
1. **Compatibility Layers Work**
- `python-future` provided seamless 2.6/2.7/3.3+ compatibility
- Gradual migration reduced risk and allowed testing
- Tool-based automation (futurize) caught 65-80% of changes
2. **Feature Flags Enable Rollback**
- Dual implementations run side-by-side during transition
- Environment variable switches between old/new paths
- Observability metrics validate equivalence
3. **Incremental Adoption**
- Start with low-risk, high-value targets (session hooks)
- Gather metrics before expanding scope
- Maintain backward compatibility throughout
---
## 3. Architecture Recommendations
### 3.1 Filesystem Structure
```
src/mcp_memory_service/
├── api/ # NEW: Code execution interface
│ ├── __init__.py # Public API exports
│ ├── compact.py # Compact result types
│ ├── search.py # Search operations
│ ├── storage.py # Storage operations
│ └── utils.py # Helper functions
├── models/
│ ├── memory.py # EXISTING: Full Memory model
│ └── compact.py # NEW: CompactMemory types
├── storage/
│ ├── base.py # EXISTING: Abstract interface
│ ├── sqlite_vec.py # EXISTING: SQLite backend
│ └── ...
└── server.py # EXISTING: MCP server (keep for compatibility)
```
### 3.2 Compact Result Types
**Design Principles:**
1. Return minimal data for common use cases
2. Provide "expand" functions for full details when needed
3. Use immutable types (NamedTuple) for safety
**Implementation:**
```python
# src/mcp_memory_service/models/compact.py
from typing import NamedTuple, Optional
class CompactMemory(NamedTuple):
"""Minimal memory for efficient token usage (~80 tokens vs ~600)."""
hash: str # Content hash (8 chars)
preview: str # First 200 chars of content
tags: tuple[str, ...] # Immutable tag tuple
created: float # Unix timestamp
score: float = 0.0 # Relevance score
class CompactSearchResult(NamedTuple):
"""Search result with minimal overhead."""
memories: tuple[CompactMemory, ...]
total: int
query: str
def __repr__(self) -> str:
"""Compact string representation."""
return f"SearchResult(found={self.total}, shown={len(self.memories)})"
class CompactStorageInfo(NamedTuple):
"""Health check result (~20 tokens vs ~100)."""
backend: str # 'sqlite_vec' | 'cloudflare' | 'hybrid'
count: int # Total memories
ready: bool # Service operational
```
**Token Comparison:**
```python
# Full Memory object (current):
{
"content": "Long text...", # ~400 tokens
"content_hash": "abc123...", # ~10 tokens
"tags": ["tag1", "tag2"], # ~15 tokens
"memory_type": "note", # ~5 tokens
"metadata": {...}, # ~50 tokens
"embedding": [0.1, 0.2, ...], # ~300 tokens (768 dims)
"created_at": 1730928000.0, # ~8 tokens
"created_at_iso": "2025-11-06...", # ~12 tokens
"updated_at": 1730928000.0, # ~8 tokens
"updated_at_iso": "2025-11-06..." # ~12 tokens
}
# Total: ~820 tokens per memory
# CompactMemory (proposed):
CompactMemory(
hash='abc123', # ~5 tokens
preview='Long text...'[:200], # ~50 tokens
tags=('tag1', 'tag2'), # ~10 tokens
created=1730928000.0, # ~5 tokens
score=0.85 # ~3 tokens
)
# Total: ~73 tokens per memory (91% reduction!)
```
### 3.3 Core API Functions
**Design Goals:**
- Single import statement replaces tool definitions
- Type hints provide inline documentation
- Sync wrappers for non-async contexts (hooks)
- Automatic connection management
**Implementation:**
```python
# src/mcp_memory_service/api/__init__.py
"""
Code execution API for mcp-memory-service.
This module provides a lightweight, token-efficient interface
for direct Python code execution, replacing MCP tool calls.
Token Efficiency:
- Import: ~10 tokens (once per session)
- Function call: ~5-20 tokens (vs 125+ for MCP tools)
- Results: 73-200 tokens (vs 500-800 for full Memory objects)
Example:
from mcp_memory_service.api import search, store, health
# Search (20 tokens vs 625 tokens for MCP)
results = search("architecture decisions", limit=5)
for m in results.memories:
print(f"{m.hash}: {m.preview}")
# Store (15 tokens vs 150 tokens for MCP)
store("New memory", tags=['note', 'important'])
# Health (5 tokens vs 125 tokens for MCP)
info = health()
print(f"Backend: {info.backend}, Count: {info.count}")
"""
from typing import Optional, Union
from .compact import CompactMemory, CompactSearchResult, CompactStorageInfo
from .search import search, search_by_tag, recall
from .storage import store, delete, update
from .utils import health, expand_memory
__all__ = [
# Search operations
'search', # Semantic search with compact results
'search_by_tag', # Tag-based search
'recall', # Time-based natural language search
# Storage operations
'store', # Store new memory
'delete', # Delete by hash
'update', # Update metadata
# Utilities
'health', # Service health check
'expand_memory', # Get full Memory from hash
# Types
'CompactMemory',
'CompactSearchResult',
'CompactStorageInfo',
]
# Version for API compatibility tracking
__api_version__ = "1.0.0"
```
```python
# src/mcp_memory_service/api/search.py
"""Search operations with compact results."""
import asyncio
from typing import Optional, Union
from ..storage.factory import create_storage_backend
from ..models.compact import CompactMemory, CompactSearchResult
# Thread-local storage for connection reuse
_storage_instance = None
def _get_storage():
"""Get or create storage backend instance."""
global _storage_instance
if _storage_instance is None:
_storage_instance = create_storage_backend()
# Initialize in sync context (run once)
asyncio.run(_storage_instance.initialize())
return _storage_instance
def search(
query: str,
limit: int = 5,
threshold: float = 0.0
) -> CompactSearchResult:
"""
Search memories using semantic similarity.
Token efficiency: ~25 tokens (query + params + results)
vs ~625 tokens for MCP tool call with full Memory objects.
Args:
query: Search query text
limit: Maximum results to return (default: 5)
threshold: Minimum similarity score 0.0-1.0 (default: 0.0)
Returns:
CompactSearchResult with minimal memory representations
Example:
>>> results = search("recent architecture changes", limit=3)
>>> print(results)
SearchResult(found=3, shown=3)
>>> for m in results.memories:
... print(f"{m.hash}: {m.preview[:50]}...")
"""
storage = _get_storage()
# Run async operation in sync context
async def _search():
query_results = await storage.retrieve(query, n_results=limit)
# Convert to compact format
compact = [
CompactMemory(
hash=r.memory.content_hash[:8], # 8 char hash
preview=r.memory.content[:200], # First 200 chars
tags=tuple(r.memory.tags), # Immutable tuple
created=r.memory.created_at,
score=r.relevance_score
)
for r in query_results
if r.relevance_score >= threshold
]
return CompactSearchResult(
memories=tuple(compact),
total=len(compact),
query=query
)
return asyncio.run(_search())
def search_by_tag(
tags: Union[str, list[str]],
limit: Optional[int] = None
) -> CompactSearchResult:
"""
Search memories by tags.
Args:
tags: Single tag or list of tags
limit: Maximum results (None for all)
Returns:
CompactSearchResult with matching memories
"""
storage = _get_storage()
tag_list = [tags] if isinstance(tags, str) else tags
async def _search():
memories = await storage.search_by_tag(tag_list)
if limit:
memories = memories[:limit]
compact = [
CompactMemory(
hash=m.content_hash[:8],
preview=m.content[:200],
tags=tuple(m.tags),
created=m.created_at,
score=1.0 # Tag match = perfect relevance
)
for m in memories
]
return CompactSearchResult(
memories=tuple(compact),
total=len(compact),
query=f"tags:{','.join(tag_list)}"
)
return asyncio.run(_search())
def recall(query: str, n_results: int = 5) -> CompactSearchResult:
"""
Retrieve memories using natural language time expressions.
Examples:
- "last week"
- "yesterday afternoon"
- "this month"
- "2 days ago"
Args:
query: Natural language time query
n_results: Maximum results to return
Returns:
CompactSearchResult with time-filtered memories
"""
storage = _get_storage()
async def _recall():
memories = await storage.recall_memory(query, n_results)
compact = [
CompactMemory(
hash=m.content_hash[:8],
preview=m.content[:200],
tags=tuple(m.tags),
created=m.created_at,
score=1.0
)
for m in memories
]
return CompactSearchResult(
memories=tuple(compact),
total=len(compact),
query=query
)
return asyncio.run(_recall())
```
```python
# src/mcp_memory_service/api/storage.py
"""Storage operations (store, delete, update)."""
import asyncio
from typing import Optional, Union
from ..models.memory import Memory
from ..utils.hashing import generate_content_hash
from .search import _get_storage
def store(
content: str,
tags: Optional[Union[str, list[str]]] = None,
memory_type: Optional[str] = None,
metadata: Optional[dict] = None
) -> str:
"""
Store a new memory.
Token efficiency: ~15 tokens (params only)
vs ~150 tokens for MCP tool call with schema.
Args:
content: Memory content text
tags: Single tag or list of tags
memory_type: Memory type classification
metadata: Additional metadata dictionary
Returns:
Content hash of stored memory (8 chars)
Example:
>>> hash = store("Important decision", tags=['architecture', 'decision'])
>>> print(f"Stored: {hash}")
Stored: abc12345
"""
storage = _get_storage()
# Normalize tags
if isinstance(tags, str):
tag_list = [tags]
elif tags is None:
tag_list = []
else:
tag_list = list(tags)
# Create memory object
content_hash = generate_content_hash(content)
memory = Memory(
content=content,
content_hash=content_hash,
tags=tag_list,
memory_type=memory_type,
metadata=metadata or {}
)
# Store
async def _store():
success, message = await storage.store(memory)
if not success:
raise RuntimeError(f"Failed to store memory: {message}")
return content_hash[:8]
return asyncio.run(_store())
def delete(hash: str) -> bool:
"""
Delete a memory by its content hash.
Args:
hash: Content hash (8+ characters)
Returns:
True if deleted, False if not found
"""
storage = _get_storage()
async def _delete():
# If short hash provided, expand to full hash
if len(hash) == 8:
# Get full hash from short form (requires index lookup)
memories = await storage.get_recent_memories(n=10000)
full_hash = next(
(m.content_hash for m in memories if m.content_hash.startswith(hash)),
hash
)
else:
full_hash = hash
success, _ = await storage.delete(full_hash)
return success
return asyncio.run(_delete())
def update(
hash: str,
tags: Optional[list[str]] = None,
memory_type: Optional[str] = None,
metadata: Optional[dict] = None
) -> bool:
"""
Update memory metadata.
Args:
hash: Content hash (8+ characters)
tags: New tags (replaces existing)
memory_type: New memory type
metadata: New metadata (merges with existing)
Returns:
True if updated successfully
"""
storage = _get_storage()
updates = {}
if tags is not None:
updates['tags'] = tags
if memory_type is not None:
updates['memory_type'] = memory_type
if metadata is not None:
updates['metadata'] = metadata
async def _update():
success, _ = await storage.update_memory_metadata(hash, updates)
return success
return asyncio.run(_update())
```
```python
# src/mcp_memory_service/api/utils.py
"""Utility functions for API."""
import asyncio
from typing import Optional
from ..models.memory import Memory
from ..models.compact import CompactStorageInfo
from .search import _get_storage
def health() -> CompactStorageInfo:
"""
Get service health and status.
Token efficiency: ~20 tokens
vs ~125 tokens for MCP health check tool.
Returns:
CompactStorageInfo with backend, count, and ready status
Example:
>>> info = health()
>>> print(f"Using {info.backend}, {info.count} memories")
Using sqlite_vec, 1247 memories
"""
storage = _get_storage()
async def _health():
stats = await storage.get_stats()
return CompactStorageInfo(
backend=stats.get('storage_backend', 'unknown'),
count=stats.get('total_memories', 0),
ready=stats.get('status', 'unknown') == 'operational'
)
return asyncio.run(_health())
def expand_memory(hash: str) -> Optional[Memory]:
"""
Get full Memory object from compact hash.
Use when you need complete memory details (content, embedding, etc.)
after working with compact results.
Args:
hash: Content hash (8+ characters)
Returns:
Full Memory object or None if not found
Example:
>>> results = search("architecture", limit=5)
>>> full = expand_memory(results.memories[0].hash)
>>> print(full.content) # Complete content, not preview
"""
storage = _get_storage()
async def _expand():
# Handle short hash
if len(hash) == 8:
memories = await storage.get_recent_memories(n=10000)
full_hash = next(
(m.content_hash for m in memories if m.content_hash.startswith(hash)),
None
)
if full_hash is None:
return None
else:
full_hash = hash
return await storage.get_by_hash(full_hash)
return asyncio.run(_expand())
```
### 3.4 Hook Integration Pattern
**Before (MCP Tool Invocation):**
```javascript
// ~/.claude/hooks/core/session-start.js
const { MemoryClient } = require('../utilities/memory-client');
async function retrieveMemories(gitContext) {
const memoryClient = new MemoryClient(config);
// MCP tool call: ~625 tokens (tool def + result)
const result = await memoryClient.callTool('retrieve_memory', {
query: gitContext.query,
limit: 8,
similarity_threshold: 0.6
});
// Result parsing adds more tokens
const memories = parseToolResult(result);
return memories; // 8 full Memory objects = ~6,400 tokens
}
```
**After (Code Execution):**
```javascript
// ~/.claude/hooks/core/session-start.js
const { execSync } = require('child_process');
async function retrieveMemories(gitContext) {
// Execute Python code directly: ~25 tokens total
const pythonCode = `
from mcp_memory_service.api import search
results = search("${gitContext.query}", limit=8)
for m in results.memories:
print(f"{m.hash}|{m.preview}|{','.join(m.tags)}|{m.created}")
`;
const output = execSync(`python -c "${pythonCode}"`, {
encoding: 'utf8',
timeout: 5000
});
// Parse compact results (8 memories = ~600 tokens total)
const memories = output.trim().split('\n').map(line => {
const [hash, preview, tags, created] = line.split('|');
return { hash, preview, tags: tags.split(','), created: parseFloat(created) };
});
return memories; // 90% token reduction: 6,400 → 600 tokens
}
```
---
## 4. Implementation Examples from Similar Projects
### 4.1 MCP Python Interpreter (Nov 2024)
**Key Features:**
- Sandboxed code execution in isolated directories
- File read/write capabilities through code
- Iterative error correction (write → run → fix → repeat)
**Relevant Patterns:**
```python
# Tool exposure as filesystem
# Instead of: 33 tool definitions in context
# Use: import from known locations
from mcp_memory_service.api import search, store, health
# LLM can discover functions via IDE-like introspection
help(search) # Returns compact docstring
```
### 4.2 CodeAgents Framework
**Token Efficiency Techniques:**
1. **Typed Variables**: `memories: list[CompactMemory]` (10 tokens) vs "a list of memory objects with content and metadata" (15+ tokens)
2. **Control Structures**: `for m in memories if m.score > 0.7` (12 tokens) vs calling filter tool (125+ tokens)
3. **Reusable Subroutines**: Single function encapsulates common pattern
**Application to Memory Service:**
```python
# Compact search and filter in code (30 tokens total)
from mcp_memory_service.api import search
results = search("architecture", limit=20)
relevant = [m for m in results.memories if 'decision' in m.tags and m.score > 0.7]
print(f"Found {len(relevant)} relevant memories")
# vs MCP tools (625+ tokens)
# 1. retrieve_memory tool call (125 tokens)
# 2. Full results parsing (400 tokens)
# 3. search_by_tag tool call (125 tokens)
# 4. Manual filtering logic in prompt (100+ tokens)
```
---
## 5. Potential Challenges and Mitigation Strategies
### Challenge 1: Async/Sync Context Mismatch
**Problem:** Hooks run in Node.js (sync), storage backends use asyncio (async)
**Mitigation:**
```python
# Provide sync wrappers that handle async internally
import asyncio
def search(query: str, limit: int = 5):
"""Sync wrapper for async storage operations."""
async def _search():
storage = _get_storage()
results = await storage.retrieve(query, limit)
return _convert_to_compact(results)
# Run in event loop
return asyncio.run(_search())
```
**Trade-offs:**
- ✅ Simple API for hook developers
- ✅ No async/await in JavaScript
- ⚠️ Small overhead (~1-2ms) for event loop creation
- ✅ Acceptable for hooks (not high-frequency calls)
### Challenge 2: Connection Management
**Problem:** Multiple calls from hooks shouldn't create new connections each time
**Mitigation:**
```python
# Thread-local storage instance (reused across calls)
_storage_instance = None
def _get_storage():
global _storage_instance
if _storage_instance is None:
_storage_instance = create_storage_backend()
asyncio.run(_storage_instance.initialize())
return _storage_instance
```
**Benefits:**
- ✅ Single connection per process
- ✅ Automatic initialization on first use
- ✅ No manual connection cleanup needed
### Challenge 3: Backward Compatibility
**Problem:** Existing users rely on MCP tools, can't break them
**Mitigation Strategy:**
```python
# Phase 1: Add code execution API alongside MCP tools
# Both interfaces work simultaneously
# - MCP server (server.py) continues operating
# - New api/ module available for direct import
# - Users opt-in to new approach
# Phase 2: Encourage migration with documentation
# - Performance comparison benchmarks
# - Token usage metrics
# - Migration guide with examples
# Phase 3 (Optional): Deprecation path
# - Log warnings when MCP tools used
# - Offer automatic migration scripts
# - Eventually remove or maintain minimal MCP support
```
**Migration Timeline:**
```
Week 1-2: Core API implementation + tests
Week 3: Session hook migration + validation
Week 4-5: Search operation migration
Week 6+: Optional optimizations + additional operations
```
### Challenge 4: Error Handling in Compact Mode
**Problem:** Less context in compact results makes debugging harder
**Mitigation:**
```python
# Compact results for normal operation
results = search("query", limit=5)
# Expand individual memory for debugging
if results.memories:
full_memory = expand_memory(results.memories[0].hash)
print(full_memory.content) # Complete content
print(full_memory.metadata) # All metadata
# Health check provides diagnostics
info = health()
if not info.ready:
raise RuntimeError(f"Storage backend {info.backend} not ready")
```
### Challenge 5: Performance with Large Result Sets
**Problem:** Converting 1000s of memories to compact format
**Mitigation:**
```python
# Lazy evaluation for large queries
from typing import Iterator
def search_iter(query: str, batch_size: int = 50) -> Iterator[CompactMemory]:
"""Streaming search results for large queries."""
storage = _get_storage()
offset = 0
while True:
batch = storage.retrieve(query, n_results=batch_size, offset=offset)
if not batch:
break
for result in batch:
yield CompactMemory(...)
offset += batch_size
# Use in hooks
for memory in search_iter("query", batch_size=10):
if some_condition(memory):
break # Early termination saves processing
```
---
## 6. Recommended Tools and Libraries
### 6.1 Type Safety and Validation
**Pydantic v2** (optional, for advanced use cases)
```python
from pydantic import BaseModel, Field, field_validator
class SearchParams(BaseModel):
query: str = Field(min_length=1, max_length=1000)
limit: int = Field(default=5, ge=1, le=100)
threshold: float = Field(default=0.0, ge=0.0, le=1.0)
@field_validator('query')
def query_not_empty(cls, v):
if not v.strip():
raise ValueError('Query cannot be empty')
return v
```
**Benefits:**
- ✅ Runtime validation with clear error messages
- ✅ JSON schema generation for documentation
- ✅ 25-50% overhead acceptable for API boundaries
- ⚠️ Use NamedTuple for internal compact types (lighter weight)
### 6.2 Testing and Validation
**pytest-asyncio** (already in use)
```python
# tests/api/test_search.py
import pytest
from mcp_memory_service.api import search, store, health
def test_search_returns_compact_results():
"""Verify search returns CompactSearchResult."""
results = search("test query", limit=3)
assert results.total >= 0
assert len(results.memories) <= 3
assert all(isinstance(m.hash, str) for m in results.memories)
assert all(len(m.hash) == 8 for m in results.memories)
def test_token_efficiency():
"""Benchmark token usage vs MCP tools."""
import tiktoken
enc = tiktoken.encoding_for_model("gpt-4")
# Compact API
results = search("architecture", limit=5)
compact_repr = str(results.memories)
compact_tokens = len(enc.encode(compact_repr))
# Compare with full Memory objects
from mcp_memory_service.storage import get_storage
full_results = get_storage().retrieve("architecture", n_results=5)
full_repr = str([r.memory.to_dict() for r in full_results])
full_tokens = len(enc.encode(full_repr))
reduction = (1 - compact_tokens / full_tokens) * 100
assert reduction >= 85, f"Expected 85%+ reduction, got {reduction:.1f}%"
```
### 6.3 Documentation Generation
**Sphinx with autodoc** (existing infrastructure)
```python
# Docstrings optimized for both humans and LLMs
def search(query: str, limit: int = 5) -> CompactSearchResult:
"""
Search memories using semantic similarity.
This function provides a token-efficient alternative to the
retrieve_memory MCP tool, reducing token usage by ~90%.
Token Cost Analysis:
- Function call: ~20 tokens (import + call)
- Results: ~73 tokens per memory
- Total for 5 results: ~385 tokens
vs MCP Tool:
- Tool definition: ~125 tokens
- Full Memory results: ~500 tokens per memory
- Total for 5 results: ~2,625 tokens
Reduction: 85% (2,625 → 385 tokens)
Performance:
- Cold call: ~50ms (storage initialization)
- Warm call: ~5ms (connection reused)
Args:
query: Search query text. Supports natural language.
Examples: "recent architecture decisions",
"authentication implementation notes"
limit: Maximum number of results to return.
Higher values increase token cost proportionally.
Recommended: 3-8 for hooks, 10-20 for interactive use.
Returns:
CompactSearchResult containing:
- memories: Tuple of CompactMemory objects
- total: Number of results found
- query: Original query string
Raises:
RuntimeError: If storage backend not initialized
ValueError: If query empty or limit invalid
Example:
>>> from mcp_memory_service.api import search
>>> results = search("authentication setup", limit=3)
>>> print(results)
SearchResult(found=3, shown=3)
>>> for m in results.memories:
... print(f"{m.hash}: {m.preview[:50]}...")
abc12345: Implemented OAuth 2.1 authentication with...
def67890: Added JWT token validation middleware for...
ghi11121: Fixed authentication race condition in...
See Also:
- search_by_tag: Filter by specific tags
- recall: Time-based natural language queries
- expand_memory: Get full Memory object from hash
"""
...
```
### 6.4 Performance Monitoring
**structlog** (lightweight, JSON-compatible)
```python
import structlog
logger = structlog.get_logger(__name__)
def search(query: str, limit: int = 5):
with logger.contextualize(operation="search", query=query, limit=limit):
start = time.perf_counter()
try:
results = _do_search(query, limit)
duration_ms = (time.perf_counter() - start) * 1000
logger.info(
"search_completed",
duration_ms=duration_ms,
results_count=len(results.memories),
token_estimate=len(results.memories) * 73 # Compact token estimate
)
return results
except Exception as e:
logger.error("search_failed", error=str(e), exc_info=True)
raise
```
---
## 7. Migration Approach: Gradual Transition
### Phase 1: Core Infrastructure (Week 1-2)
**Deliverables:**
- ✅ `src/mcp_memory_service/api/` module structure
- ✅ `CompactMemory`, `CompactSearchResult`, `CompactStorageInfo` types
- ✅ `search()`, `store()`, `health()` functions
- ✅ Unit tests with 90%+ coverage
- ✅ Documentation with token usage benchmarks
**Success Criteria:**
- All functions work in sync context (no async/await in API)
- Connection reuse validated (single storage instance)
- Token reduction measured: 85%+ for search operations
- Performance overhead <5ms per call (warm)
**Risk: Low** - New code, no existing dependencies
### Phase 2: Session Hook Optimization (Week 3)
**Target:** Session-start hook (highest impact: 3,600-9,600 tokens → 900-2,400 tokens)
**Changes:**
```javascript
// Before: MCP tool invocation
const { MemoryClient } = require('../utilities/memory-client');
const memoryClient = new MemoryClient(config);
const result = await memoryClient.callTool('retrieve_memory', {...});
// After: Code execution with fallback
const { execSync } = require('child_process');
try {
// Try code execution first (fast, efficient)
const output = execSync('python -c "from mcp_memory_service.api import search; ..."');
const memories = parseCompactResults(output);
} catch (error) {
// Fallback to MCP if code execution fails
console.warn('Code execution failed, falling back to MCP:', error);
const result = await memoryClient.callTool('retrieve_memory', {...});
const memories = parseMCPResult(result);
}
```
**Success Criteria:**
- 75%+ token reduction measured in real sessions
- Fallback mechanism validates graceful degradation
- Hook execution time <500ms (no user-facing latency increase)
- Zero breaking changes for users
**Risk: Medium** - Touches production hook code, but has fallback
### Phase 3: Search Operation Optimization (Week 4-5)
**Target:** Mid-conversation and topic-change hooks
**Deliverables:**
- ✅ `search_by_tag()` implementation
- ✅ `recall()` natural language time queries
- ✅ Streaming search (`search_iter()`) for large results
- ✅ Migration guide with side-by-side examples
**Success Criteria:**
- 90%+ token reduction for search-heavy workflows
- Documentation shows before/after comparison
- Community feedback collected and addressed
**Risk: Low** - Builds on Phase 1 foundation
### Phase 4: Extended Operations (Week 6+)
**Optional Enhancements:**
- Document ingestion API
- Batch operations (store/delete multiple)
- Memory consolidation triggers
- Advanced filtering (memory_type, time ranges)
---
## 8. Success Metrics and Validation
### 8.1 Token Reduction Targets
| Operation | Current (MCP) | Target (Code Exec) | Reduction |
|-----------|---------------|-------------------|-----------|
| Session start hook | 3,600-9,600 | 900-2,400 | 75% |
| Search (5 results) | 2,625 | 385 | 85% |
| Store memory | 150 | 15 | 90% |
| Health check | 125 | 20 | 84% |
| Document ingestion (50 PDFs) | 57,400 | 8,610 | 85% |
**Annual Savings (Conservative):**
- 10 users x 5 sessions/day x 365 days x 6,000 tokens saved = **109.5M tokens/year**
- At $0.15/1M tokens (Claude Opus input): **$16.43/year saved** per 10-user deployment
**Annual Savings (Aggressive - 100 users):**
- 100 users x 10 sessions/day x 365 days x 6,000 tokens = **2.19B tokens/year**
- At $0.15/1M tokens: **$328.50/year saved**
### 8.2 Performance Metrics
**Latency Targets:**
- Cold start (first call): <100ms
- Warm calls: <10ms
- Hook total execution: <500ms (no degradation from current)
**Memory Usage:**
- Compact result set (5 memories): <5KB
- Full result set (5 memories): ~50KB
- 90% memory reduction for hook injection
### 8.3 Compatibility Validation
**Testing Matrix:**
- ✅ Existing MCP tools continue working (100% backward compat)
- ✅ New code execution API available alongside MCP
- ✅ Fallback mechanism activates on code execution failure
- ✅ All storage backends compatible (SQLite-Vec, Cloudflare, Hybrid)
- ✅ No breaking changes to server.py or existing APIs
---
## 9. Implementation Timeline
```
Week 1: Core Infrastructure
├── Design compact types (CompactMemory, CompactSearchResult)
├── Implement api/__init__.py with public exports
├── Create search.py with search(), search_by_tag(), recall()
├── Add storage.py with store(), delete(), update()
└── Write utils.py with health(), expand_memory()
Week 2: Testing & Documentation
├── Unit tests for all API functions
├── Integration tests with storage backends
├── Token usage benchmarking
├── API documentation with examples
└── Migration guide draft
Week 3: Session Hook Migration
├── Update session-start.js to use code execution
├── Add fallback to MCP tools
├── Test with SQLite-Vec and Cloudflare backends
├── Validate token reduction (target: 75%+)
└── Deploy to beta testers
Week 4-5: Search Operations
├── Update mid-conversation.js
├── Update topic-change.js
├── Implement streaming search for large queries
├── Document best practices
└── Gather community feedback
Week 6+: Polish & Extensions
├── Additional API functions (batch ops, etc.)
├── Performance optimizations
├── Developer tools (token calculators, debuggers)
└── Comprehensive documentation
```
---
## 10. Recommendations Summary
### Immediate Next Steps (Week 1)
1. **Create API Module Structure**
```bash
mkdir -p src/mcp_memory_service/api
touch src/mcp_memory_service/api/{__init__,compact,search,storage,utils}.py
```
2. **Implement Compact Types**
- `CompactMemory` with NamedTuple
- `CompactSearchResult` with tuple of memories
- `CompactStorageInfo` for health checks
3. **Core Functions**
- `search()` - Semantic search with compact results
- `store()` - Store with minimal params
- `health()` - Quick status check
4. **Testing Infrastructure**
- Unit tests for each function
- Token usage benchmarks
- Performance profiling
### Key Design Decisions
1. **Use NamedTuple for Compact Types**
- Fast (C-based), immutable, type-safe
- 60-90% size reduction vs dataclass
- Clear field names (`.hash` not `[0]`)
2. **Sync Wrappers for Async Operations**
- Hide asyncio complexity from hooks
- Use `asyncio.run()` internally
- Connection reuse via global instance
3. **Graceful Degradation**
- Code execution primary
- MCP tools fallback
- Zero breaking changes
4. **Incremental Migration**
- Start with session hooks (high impact)
- Gather metrics and feedback
- Expand to other operations
### Expected Outcomes
**Token Efficiency:**
- ✅ 75% reduction in session hooks
- ✅ 85-90% reduction in search operations
- ✅ 13-183 million tokens saved annually
**Performance:**
- ✅ <10ms per API call (warm)
- ✅ <500ms hook execution (no degradation)
- ✅ 90% memory footprint reduction
**Compatibility:**
- ✅ 100% backward compatible
- ✅ Opt-in adoption model
- ✅ MCP tools continue working
---
## 11. References and Further Reading
### Research Sources
1. **Anthropic Resources:**
- "Code execution with MCP: Building more efficient agents" (Nov 2025)
- "Claude Code Best Practices" - Token efficiency guidelines
- MCP Protocol Documentation - Tool use patterns
2. **Academic Research:**
- "CodeAgents: A Token-Efficient Framework for Codified Multi-Agent Reasoning" (arxiv.org/abs/2507.03254)
- Token consumption analysis in LLM agent systems
3. **Python Best Practices:**
- "Dataclasses vs NamedTuple vs TypedDict" performance comparisons
- Python API design patterns (hakibenita.com)
- Async/sync bridging patterns
4. **Real-World Implementations:**
- mcp-python-interpreter server
- Anthropic's MCP server examples
- LangChain compact result types
### Internal Documentation
- `src/mcp_memory_service/storage/base.py` - Storage interface
- `src/mcp_memory_service/models/memory.py` - Memory model
- `src/mcp_memory_service/server.py` - MCP server (3,721 lines)
- `~/.claude/hooks/core/session-start.js` - Current hook implementation
---
## Conclusion
The research validates the feasibility and high value of implementing a code execution interface for mcp-memory-service. Industry trends (Anthropic's MCP code execution announcement, CodeAgents framework) align with the proposal, and the current codebase architecture provides a solid foundation for gradual migration.
**Key Takeaways:**
1. **85-90% token reduction is achievable** through compact types and direct function calls
2. **Backward compatibility is maintained** via fallback mechanisms and parallel operation
3. **Proven patterns exist** in mcp-python-interpreter and similar projects
4. **Incremental approach reduces risk** while delivering immediate value
5. **Annual savings of 13-183M tokens** justify development investment
**Recommended Action:** Proceed with Phase 1 implementation (Core Infrastructure) targeting Week 1-2 completion, with session hook migration as first production use case.
---
**Document Version:** 1.0
**Last Updated:** November 6, 2025
**Author:** Research conducted for Issue #206
**Status:** Ready for Review and Implementation
```
--------------------------------------------------------------------------------
/src/mcp_memory_service/web/app.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.
"""
FastAPI application for MCP Memory Service HTTP/SSE interface.
Provides REST API and Server-Sent Events using SQLite-vec backend.
"""
import asyncio
import logging
import os
from contextlib import asynccontextmanager
from typing import Optional, Any
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
from fastapi.responses import HTMLResponse
from .. import __version__
from ..config import (
HTTP_PORT,
HTTP_HOST,
CORS_ORIGINS,
DATABASE_PATH,
EMBEDDING_MODEL_NAME,
MDNS_ENABLED,
HTTPS_ENABLED,
OAUTH_ENABLED,
CONSOLIDATION_ENABLED,
CONSOLIDATION_CONFIG,
CONSOLIDATION_SCHEDULE
)
from .dependencies import set_storage, get_storage, create_storage_backend
from .api.health import router as health_router
from .api.memories import router as memories_router
from .api.search import router as search_router
from .api.events import router as events_router
from .api.sync import router as sync_router
from .api.manage import router as manage_router
from .api.analytics import router as analytics_router
from .api.documents import router as documents_router
from .api.mcp import router as mcp_router
from .api.consolidation import router as consolidation_router
from .api.backup import router as backup_router
from .sse import sse_manager
logger = logging.getLogger(__name__)
# Global storage instance
storage: Optional["MemoryStorage"] = None
# Global mDNS advertiser instance
mdns_advertiser: Optional[Any] = None
# Global OAuth cleanup task
oauth_cleanup_task: Optional[asyncio.Task] = None
# Global consolidation instances
consolidator: Optional["DreamInspiredConsolidator"] = None
consolidation_scheduler: Optional["ConsolidationScheduler"] = None
async def oauth_cleanup_background_task():
"""Background task to periodically clean up expired OAuth tokens and codes."""
from .oauth.storage import oauth_storage
while True:
try:
# Clean up expired tokens every 5 minutes
await asyncio.sleep(300) # 5 minutes
cleanup_stats = await oauth_storage.cleanup_expired()
if cleanup_stats["expired_codes_cleaned"] > 0 or cleanup_stats["expired_tokens_cleaned"] > 0:
logger.info(f"OAuth cleanup: removed {cleanup_stats['expired_codes_cleaned']} codes, "
f"{cleanup_stats['expired_tokens_cleaned']} tokens")
except asyncio.CancelledError:
logger.info("OAuth cleanup task cancelled")
break
except Exception as e:
logger.error(f"Error in OAuth cleanup task: {e}")
# Continue running even if there's an error
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Application lifespan management."""
global storage, mdns_advertiser, oauth_cleanup_task, consolidator, consolidation_scheduler
# Startup
logger.info("Starting MCP Memory Service HTTP interface...")
try:
storage = await create_storage_backend()
set_storage(storage) # Set the global storage instance
# Initialize consolidation system if enabled
if CONSOLIDATION_ENABLED:
try:
from ..consolidation.base import ConsolidationConfig
from ..consolidation.consolidator import DreamInspiredConsolidator
from ..consolidation.scheduler import ConsolidationScheduler
from ..api import set_consolidator, set_scheduler
# Create consolidation config
config = ConsolidationConfig(**CONSOLIDATION_CONFIG)
# Initialize consolidator with storage
consolidator = DreamInspiredConsolidator(storage, config)
logger.info("Dream-inspired consolidator initialized")
# Set global consolidator for API access
set_consolidator(consolidator)
# Initialize scheduler if any schedules are enabled
if any(schedule != 'disabled' for schedule in CONSOLIDATION_SCHEDULE.values()):
consolidation_scheduler = ConsolidationScheduler(
consolidator,
CONSOLIDATION_SCHEDULE,
enabled=True
)
# Start the scheduler
if await consolidation_scheduler.start():
logger.info("Consolidation scheduler started successfully")
# Set global scheduler for API access
set_scheduler(consolidation_scheduler)
else:
logger.warning("Failed to start consolidation scheduler")
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}")
consolidator = None
consolidation_scheduler = None
else:
logger.info("Consolidation system disabled")
# Start SSE manager
await sse_manager.start()
logger.info("SSE Manager started")
# Start OAuth cleanup task if enabled
if OAUTH_ENABLED:
oauth_cleanup_task = asyncio.create_task(oauth_cleanup_background_task())
logger.info("OAuth cleanup background task started")
# Start mDNS service advertisement if enabled
if MDNS_ENABLED:
try:
from ..discovery.mdns_service import ServiceAdvertiser
mdns_advertiser = ServiceAdvertiser(
host=HTTP_HOST,
port=HTTP_PORT,
https_enabled=HTTPS_ENABLED
)
success = await mdns_advertiser.start()
if success:
logger.info("mDNS service advertisement started")
else:
logger.warning("Failed to start mDNS service advertisement")
mdns_advertiser = None
except ImportError:
logger.warning("mDNS support not available (zeroconf not installed)")
mdns_advertiser = None
except Exception as e:
logger.error(f"Error starting mDNS advertisement: {e}")
mdns_advertiser = None
else:
logger.info("mDNS service advertisement disabled")
except Exception as e:
logger.error(f"Failed to initialize storage: {e}")
raise
yield
# Shutdown
logger.info("Shutting down MCP Memory Service HTTP interface...")
# Stop consolidation scheduler
if consolidation_scheduler:
try:
await consolidation_scheduler.stop()
logger.info("Consolidation scheduler stopped")
except Exception as e:
logger.error(f"Error stopping consolidation scheduler: {e}")
# Stop mDNS advertisement
if mdns_advertiser:
try:
await mdns_advertiser.stop()
logger.info("mDNS service advertisement stopped")
except Exception as e:
logger.error(f"Error stopping mDNS advertisement: {e}")
# Stop OAuth cleanup task
if oauth_cleanup_task:
try:
oauth_cleanup_task.cancel()
await oauth_cleanup_task
logger.info("OAuth cleanup task stopped")
except asyncio.CancelledError:
logger.info("OAuth cleanup task cancelled successfully")
except Exception as e:
logger.error(f"Error stopping OAuth cleanup task: {e}")
# Stop SSE manager
await sse_manager.stop()
logger.info("SSE Manager stopped")
if storage:
await storage.close()
def create_app() -> FastAPI:
"""Create and configure the FastAPI application."""
app = FastAPI(
title="MCP Memory Service",
description="HTTP REST API and SSE interface for semantic memory storage",
version=__version__,
lifespan=lifespan,
docs_url="/api/docs",
redoc_url="/api/redoc"
)
# CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=CORS_ORIGINS,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Include API routers
logger.info("Including API routers...")
app.include_router(health_router, prefix="/api", tags=["health"])
logger.info(f"✓ Included health router with {len(health_router.routes)} routes")
app.include_router(memories_router, prefix="/api", tags=["memories"])
logger.info(f"✓ Included memories router with {len(memories_router.routes)} routes")
app.include_router(search_router, prefix="/api", tags=["search"])
logger.info(f"✓ Included search router with {len(search_router.routes)} routes")
app.include_router(manage_router, prefix="/api/manage", tags=["management"])
logger.info(f"✓ Included manage router with {len(manage_router.routes)} routes")
app.include_router(analytics_router, prefix="/api/analytics", tags=["analytics"])
logger.info(f"✓ Included analytics router with {len(analytics_router.routes)} routes")
app.include_router(events_router, prefix="/api", tags=["events"])
logger.info(f"✓ Included events router with {len(events_router.routes)} routes")
app.include_router(sync_router, prefix="/api", tags=["sync"])
logger.info(f"✓ Included sync router with {len(sync_router.routes)} routes")
app.include_router(backup_router, prefix="/api", tags=["backup"])
logger.info(f"✓ Included backup router with {len(backup_router.routes)} routes")
try:
app.include_router(documents_router, prefix="/api/documents", tags=["documents"])
logger.info(f"✓ Included documents router with {len(documents_router.routes)} routes")
except Exception as e:
logger.error(f"✗ Failed to include documents router: {e}")
import traceback
logger.error(traceback.format_exc())
# Include consolidation router
app.include_router(consolidation_router, tags=["consolidation"])
logger.info(f"✓ Included consolidation router with {len(consolidation_router.routes)} routes")
# Include MCP protocol router
app.include_router(mcp_router, tags=["mcp-protocol"])
# Include OAuth routers if enabled
if OAUTH_ENABLED:
from .oauth.discovery import router as oauth_discovery_router
from .oauth.registration import router as oauth_registration_router
from .oauth.authorization import router as oauth_authorization_router
app.include_router(oauth_discovery_router, tags=["oauth-discovery"])
app.include_router(oauth_registration_router, prefix="/oauth", tags=["oauth"])
app.include_router(oauth_authorization_router, prefix="/oauth", tags=["oauth"])
logger.info("OAuth 2.1 endpoints enabled")
else:
logger.info("OAuth 2.1 endpoints disabled")
# Serve static files (dashboard)
static_path = os.path.join(os.path.dirname(__file__), "static")
if os.path.exists(static_path):
app.mount("/static", StaticFiles(directory=static_path), name="static")
def get_api_overview_html():
"""Generate the API overview HTML template."""
return """
<!DOCTYPE html>
<html lang="en">
<head>
<title>MCP Memory Service v""" + __version__ + """</title>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<style>
* {
margin: 0;
padding: 0;
box-sizing: border-box;
}
:root {
--primary: #3b82f6;
--primary-dark: #2563eb;
--secondary: #8b5cf6;
--success: #10b981;
--warning: #f59e0b;
--danger: #ef4444;
--dark: #1e293b;
--gray: #64748b;
--light: #f8fafc;
--white: #ffffff;
--shadow: 0 1px 3px 0 rgba(0, 0, 0, 0.1), 0 1px 2px 0 rgba(0, 0, 0, 0.06);
--shadow-lg: 0 10px 15px -3px rgba(0, 0, 0, 0.1), 0 4px 6px -2px rgba(0, 0, 0, 0.05);
}
body {
font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, 'Helvetica Neue', Arial, sans-serif;
background: linear-gradient(135deg, #f5f7fa 0%, #c3cfe2 100%);
min-height: 100vh;
color: var(--dark);
line-height: 1.6;
}
.container {
max-width: 1200px;
margin: 0 auto;
padding: 2rem;
}
header {
text-align: center;
margin-bottom: 3rem;
padding: 2rem;
background: var(--white);
border-radius: 1rem;
box-shadow: var(--shadow-lg);
}
.logo {
display: inline-flex;
align-items: center;
gap: 1rem;
margin-bottom: 1rem;
}
.logo-icon {
width: 60px;
height: 60px;
background: linear-gradient(135deg, var(--primary) 0%, var(--secondary) 100%);
border-radius: 1rem;
display: flex;
align-items: center;
justify-content: center;
color: var(--white);
font-size: 2rem;
font-weight: bold;
}
h1 {
font-size: 2.5rem;
font-weight: 800;
background: linear-gradient(135deg, var(--primary) 0%, var(--secondary) 100%);
-webkit-background-clip: text;
-webkit-text-fill-color: transparent;
background-clip: text;
margin-bottom: 0.5rem;
}
.subtitle {
color: var(--gray);
font-size: 1.25rem;
margin-bottom: 1rem;
}
.version-badge {
display: inline-flex;
align-items: center;
gap: 0.5rem;
background: var(--success);
color: var(--white);
padding: 0.25rem 1rem;
border-radius: 2rem;
font-size: 0.875rem;
font-weight: 600;
}
.stats {
display: grid;
grid-template-columns: repeat(auto-fit, minmax(200px, 1fr));
gap: 1rem;
margin-bottom: 3rem;
}
.stat-card {
background: var(--white);
padding: 1.5rem;
border-radius: 0.75rem;
box-shadow: var(--shadow);
text-align: center;
transition: transform 0.2s ease, box-shadow 0.2s ease;
}
.stat-card:hover {
transform: translateY(-2px);
box-shadow: var(--shadow-lg);
}
.stat-value {
font-size: 2rem;
font-weight: 700;
color: var(--primary);
margin-bottom: 0.25rem;
}
.stat-label {
color: var(--gray);
font-size: 0.875rem;
text-transform: uppercase;
letter-spacing: 0.05em;
}
.endpoint-grid {
display: grid;
grid-template-columns: repeat(auto-fit, minmax(350px, 1fr));
gap: 1.5rem;
margin-bottom: 3rem;
}
.endpoint-card {
background: var(--white);
border-radius: 0.75rem;
box-shadow: var(--shadow);
overflow: hidden;
transition: transform 0.2s ease, box-shadow 0.2s ease;
}
.endpoint-card:hover {
transform: translateY(-4px);
box-shadow: var(--shadow-lg);
}
.endpoint-header {
padding: 1.5rem;
background: linear-gradient(135deg, var(--primary) 0%, var(--primary-dark) 100%);
color: var(--white);
}
.endpoint-header h3 {
font-size: 1.25rem;
margin-bottom: 0.5rem;
display: flex;
align-items: center;
gap: 0.5rem;
}
.endpoint-icon {
font-size: 1.5rem;
}
.endpoint-description {
opacity: 0.9;
font-size: 0.875rem;
}
.endpoint-list {
padding: 1.5rem;
}
.endpoint-item {
padding: 0.75rem;
border-radius: 0.5rem;
margin-bottom: 0.5rem;
background: var(--light);
transition: background-color 0.2s ease;
cursor: pointer;
}
.endpoint-item:hover {
background: #e2e8f0;
}
.method {
display: inline-block;
padding: 0.125rem 0.5rem;
border-radius: 0.25rem;
font-size: 0.75rem;
font-weight: 700;
margin-right: 0.5rem;
text-transform: uppercase;
}
.method-get { background: var(--success); color: var(--white); }
.method-post { background: var(--primary); color: var(--white); }
.method-delete { background: var(--danger); color: var(--white); }
.endpoint-path {
font-family: 'Courier New', monospace;
font-size: 0.875rem;
color: var(--dark);
}
.endpoint-desc {
font-size: 0.75rem;
color: var(--gray);
margin-top: 0.25rem;
}
.action-buttons {
display: flex;
gap: 1rem;
justify-content: center;
margin-bottom: 3rem;
}
.btn {
display: inline-flex;
align-items: center;
gap: 0.5rem;
padding: 0.75rem 1.5rem;
border-radius: 0.5rem;
font-weight: 600;
text-decoration: none;
transition: all 0.2s ease;
border: none;
cursor: pointer;
}
.btn-primary {
background: var(--primary);
color: var(--white);
}
.btn-primary:hover {
background: var(--primary-dark);
transform: translateY(-2px);
box-shadow: var(--shadow-lg);
}
.btn-secondary {
background: var(--white);
color: var(--primary);
border: 2px solid var(--primary);
}
.btn-secondary:hover {
background: var(--primary);
color: var(--white);
transform: translateY(-2px);
box-shadow: var(--shadow-lg);
}
footer {
text-align: center;
padding: 2rem;
color: var(--gray);
}
.tech-stack {
display: flex;
justify-content: center;
gap: 2rem;
margin-top: 1rem;
flex-wrap: wrap;
}
.tech-badge {
display: flex;
align-items: center;
gap: 0.5rem;
padding: 0.5rem 1rem;
background: var(--white);
border-radius: 0.5rem;
box-shadow: var(--shadow);
font-size: 0.875rem;
font-weight: 600;
}
.nav-buttons {
display: flex;
gap: 1rem;
margin-top: 1rem;
justify-content: center;
}
.nav-btn {
display: inline-flex;
align-items: center;
gap: 0.5rem;
padding: 0.75rem 1.5rem;
background: var(--primary);
color: var(--white);
text-decoration: none;
border-radius: 0.5rem;
font-weight: 600;
transition: background-color 0.2s;
box-shadow: var(--shadow);
}
.nav-btn:hover {
background: var(--primary-dark);
text-decoration: none;
color: var(--white);
}
.nav-btn.secondary {
background: var(--gray);
}
.nav-btn.secondary:hover {
background: #475569;
}
.loading {
display: inline-block;
width: 1rem;
height: 1rem;
border: 2px solid var(--light);
border-top-color: var(--primary);
border-radius: 50%;
animation: spin 0.6s linear infinite;
}
@keyframes spin {
to { transform: rotate(360deg); }
}
@media (max-width: 768px) {
.container { padding: 1rem; }
h1 { font-size: 2rem; }
.endpoint-grid { grid-template-columns: 1fr; }
.stats { grid-template-columns: 1fr; }
.action-buttons {
flex-direction: column;
align-items: center;
gap: 0.75rem;
}
.btn {
width: 100%;
max-width: 300px;
justify-content: center;
}
}
</style>
</head>
<body>
<div class="container">
<header>
<div class="logo">
<div class="logo-icon">🧠</div>
<div>
<h1>MCP Memory Service</h1>
<p class="subtitle" id="subtitle">Intelligent Semantic Memory with <span id="backend-name">Loading...</span></p>
</div>
</div>
<div class="version-badge">
<span>✅</span> <span id="version-display">Loading...</span> - Latest Release
</div>
<div class="nav-buttons">
<a href="/" class="nav-btn">
<svg width="20" height="20" fill="currentColor" viewBox="0 0 24 24">
<path d="M10,20V14H14V20H19V12H22L12,3L2,12H5V20H10Z"/>
</svg>
Interactive Dashboard
</a>
<a href="/api/docs" class="nav-btn secondary" target="_blank">
<svg width="20" height="20" fill="currentColor" viewBox="0 0 24 24">
<path d="M14,17H7V15H14M17,13H7V11H17M17,9H7V7H17M19,3H5C3.89,3 3,3.89 3,5V19A2,2 0 0,0 5,21H19A2,2 0 0,0 21,19V5C21,3.89 20.1,3 19,3Z"/>
</svg>
Swagger UI
</a>
</div>
</header>
<div class="stats" id="stats">
<div class="stat-card">
<div class="stat-value"><span class="loading"></span></div>
<div class="stat-label">Total Memories</div>
</div>
<div class="stat-card">
<div class="stat-value"><span class="loading"></span></div>
<div class="stat-label">Embedding Model</div>
</div>
<div class="stat-card">
<div class="stat-value"><span class="loading"></span></div>
<div class="stat-label">Server Status</div>
</div>
<div class="stat-card">
<div class="stat-value"><span class="loading"></span></div>
<div class="stat-label">Response Time</div>
</div>
</div>
<div class="action-buttons">
<a href="/api/docs" class="btn btn-primary">
<span>📚</span> Interactive API Docs
</a>
<a href="/api/redoc" class="btn btn-secondary">
<span>📖</span> ReDoc Documentation
</a>
<a href="https://github.com/doobidoo/mcp-memory-service" class="btn btn-secondary" target="_blank">
<span>🚀</span> GitHub Repository
</a>
</div>
<div class="endpoint-grid">
<div class="endpoint-card">
<div class="endpoint-header">
<h3><span class="endpoint-icon">💾</span> Memory Management</h3>
<p class="endpoint-description">Store, retrieve, and manage semantic memories</p>
</div>
<div class="endpoint-list">
<div class="endpoint-item" onclick="window.location.href='/api/docs#/memories/store_memory_api_memories_post'">
<span class="method method-post">POST</span>
<span class="endpoint-path">/api/memories</span>
<div class="endpoint-desc">Store a new memory with automatic embedding generation</div>
</div>
<div class="endpoint-item" onclick="window.location.href='/api/docs#/memories/list_memories_api_memories_get'">
<span class="method method-get">GET</span>
<span class="endpoint-path">/api/memories</span>
<div class="endpoint-desc">List all memories with pagination support</div>
</div>
<div class="endpoint-item" onclick="window.location.href='/api/docs#/memories/get_memory_api_memories__content_hash__get'">
<span class="method method-get">GET</span>
<span class="endpoint-path">/api/memories/{hash}</span>
<div class="endpoint-desc">Retrieve a specific memory by content hash</div>
</div>
<div class="endpoint-item" onclick="window.location.href='/api/docs#/memories/delete_memory_api_memories__content_hash__delete'">
<span class="method method-delete">DELETE</span>
<span class="endpoint-path">/api/memories/{hash}</span>
<div class="endpoint-desc">Delete a memory and its embeddings</div>
</div>
</div>
</div>
<div class="endpoint-card">
<div class="endpoint-header">
<h3><span class="endpoint-icon">🔍</span> Search Operations</h3>
<p class="endpoint-description">Powerful semantic and tag-based search</p>
</div>
<div class="endpoint-list">
<div class="endpoint-item" onclick="window.location.href='/api/docs#/search/semantic_search_api_search_post'">
<span class="method method-post">POST</span>
<span class="endpoint-path">/api/search</span>
<div class="endpoint-desc">Semantic similarity search using embeddings</div>
</div>
<div class="endpoint-item" onclick="window.location.href='/api/docs#/search/tag_search_api_search_by_tag_post'">
<span class="method method-post">POST</span>
<span class="endpoint-path">/api/search/by-tag</span>
<div class="endpoint-desc">Search memories by tags (AND/OR logic)</div>
</div>
<div class="endpoint-item" onclick="window.location.href='/api/docs#/search/time_search_api_search_by_time_post'">
<span class="method method-post">POST</span>
<span class="endpoint-path">/api/search/by-time</span>
<div class="endpoint-desc">Natural language time-based queries</div>
</div>
<div class="endpoint-item" onclick="window.location.href='/api/docs#/search/find_similar_api_search_similar__content_hash__get'">
<span class="method method-get">GET</span>
<span class="endpoint-path">/api/search/similar/{hash}</span>
<div class="endpoint-desc">Find memories similar to a specific one</div>
</div>
</div>
</div>
<div class="endpoint-card">
<div class="endpoint-header">
<h3><span class="endpoint-icon">📡</span> Real-time Events</h3>
<p class="endpoint-description">Server-Sent Events for live updates</p>
</div>
<div class="endpoint-list">
<div class="endpoint-item" onclick="window.location.href='/api/events'">
<span class="method method-get">GET</span>
<span class="endpoint-path">/api/events</span>
<div class="endpoint-desc">Subscribe to real-time memory events stream</div>
</div>
<div class="endpoint-item" onclick="window.location.href='/api/events/stats'">
<span class="method method-get">GET</span>
<span class="endpoint-path">/api/events/stats</span>
<div class="endpoint-desc">View SSE connection statistics</div>
</div>
<div class="endpoint-item" onclick="window.location.href='/static/sse_test.html'">
<span class="method method-get">GET</span>
<span class="endpoint-path">/static/sse_test.html</span>
<div class="endpoint-desc">Interactive SSE testing interface</div>
</div>
</div>
</div>
<div class="endpoint-card">
<div class="endpoint-header">
<h3><span class="endpoint-icon">🏥</span> Health & Status</h3>
<p class="endpoint-description">Monitor service health and performance</p>
</div>
<div class="endpoint-list">
<div class="endpoint-item" onclick="window.location.href='/api/health'">
<span class="method method-get">GET</span>
<span class="endpoint-path">/api/health</span>
<div class="endpoint-desc">Quick health check endpoint</div>
</div>
<div class="endpoint-item" onclick="window.location.href='/api/health/detailed'">
<span class="method method-get">GET</span>
<span class="endpoint-path">/api/health/detailed</span>
<div class="endpoint-desc">Detailed health with database statistics</div>
</div>
<div class="endpoint-item" onclick="window.location.href='/api/docs'">
<span class="method method-get">GET</span>
<span class="endpoint-path">/api/docs</span>
<div class="endpoint-desc">Interactive Swagger UI documentation</div>
</div>
<div class="endpoint-item" onclick="window.location.href='/api/redoc'">
<span class="method method-get">GET</span>
<span class="endpoint-path">/api/redoc</span>
<div class="endpoint-desc">Alternative ReDoc documentation</div>
</div>
</div>
</div>
</div>
<footer>
<p>Powered by cutting-edge technology</p>
<div class="tech-stack">
<div class="tech-badge">
<span>🐍</span> FastAPI
</div>
<div class="tech-badge">
<span>🗄️</span> SQLite-vec
</div>
<div class="tech-badge">
<span>🧠</span> Sentence Transformers
</div>
<div class="tech-badge">
<span>🔥</span> PyTorch
</div>
<div class="tech-badge">
<span>🌐</span> mDNS Discovery
</div>
</div>
<p style="margin-top: 2rem; opacity: 0.8;">
© 2025 MCP Memory Service | Apache 2.0 License
</p>
</footer>
</div>
<script>
// Fetch and display live stats
async function updateStats() {
try {
const healthResponse = await fetch('/api/health');
const health = await healthResponse.json();
const detailedResponse = await fetch('/api/health/detailed');
const detailed = await detailedResponse.json();
const stats = document.getElementById('stats');
stats.innerHTML = `
<div class="stat-card">
<div class="stat-value">${detailed.statistics?.total_memories || 0}</div>
<div class="stat-label">Total Memories</div>
</div>
<div class="stat-card">
<div class="stat-value">all-MiniLM-L6-v2</div>
<div class="stat-label">Embedding Model</div>
</div>
<div class="stat-card">
<div class="stat-value" style="color: var(--success);">● Healthy</div>
<div class="stat-label">Server Status</div>
</div>
<div class="stat-card">
<div class="stat-value"><1ms</div>
<div class="stat-label">Response Time</div>
</div>
`;
} catch (error) {
console.error('Failed to fetch stats:', error);
}
}
// Update stats on page load
updateStats();
// Update stats every 30 seconds
setInterval(updateStats, 30000);
</script>
<script>
// Dynamic content loading for API overview
function getBackendDisplayName(backend) {
const backendMap = {
'sqlite-vec': 'SQLite-vec',
'sqlite_vec': 'SQLite-vec',
'cloudflare': 'Cloudflare D1 + Vectorize',
'hybrid': 'Hybrid (SQLite-vec + Cloudflare)'
};
return backendMap[backend] || backend || 'Unknown Backend';
}
async function loadDynamicInfo() {
try {
// Load detailed health information
const response = await fetch('/api/health/detailed');
if (!response.ok) {
throw new Error(`HTTP ${response.status}`);
}
const healthData = await response.json();
// Update version display
const versionEl = document.getElementById('version-display');
if (versionEl && healthData.version) {
versionEl.textContent = `v${healthData.version}`;
}
// Update backend name and subtitle
const backendNameEl = document.getElementById('backend-name');
const subtitleEl = document.getElementById('subtitle');
if (healthData.storage && healthData.storage.backend) {
const backendDisplay = getBackendDisplayName(healthData.storage.backend);
if (backendNameEl) {
backendNameEl.textContent = backendDisplay;
}
if (subtitleEl) {
subtitleEl.innerHTML = `Intelligent Semantic Memory with <span id="backend-name">${backendDisplay}</span>`;
}
}
} catch (error) {
console.error('Error loading dynamic info:', error);
// Fallback values on error
const versionEl = document.getElementById('version-display');
const backendNameEl = document.getElementById('backend-name');
const subtitleEl = document.getElementById('subtitle');
if (versionEl) {
versionEl.textContent = 'v?.?.?';
}
if (backendNameEl) {
backendNameEl.textContent = 'Unknown Backend';
}
if (subtitleEl) {
subtitleEl.innerHTML = 'Intelligent Semantic Memory with <span id="backend-name">Unknown Backend</span>';
}
}
}
// Load dynamic content when page loads
document.addEventListener('DOMContentLoaded', loadDynamicInfo);
</script>
</body>
</html>
"""
@app.get("/api-overview", response_class=HTMLResponse)
async def api_overview():
"""Serve the API documentation overview page."""
return get_api_overview_html()
@app.get("/", response_class=HTMLResponse)
async def dashboard():
"""Serve the dashboard homepage."""
# Serve the migrated interactive dashboard instead of hardcoded template
try:
# Path to the migrated dashboard HTML file
dashboard_path = os.path.join(os.path.dirname(__file__), "static", "index.html")
if os.path.exists(dashboard_path):
# Read and serve the migrated dashboard
with open(dashboard_path, 'r', encoding='utf-8') as f:
return f.read()
else:
# Fallback to original template if dashboard not found
return html_template
except Exception as e:
# Error fallback to original template
logger.warning(f"Error loading migrated dashboard: {e}")
return html_template
return app
# Create the app instance
app = create_app()
# Storage getter is now in dependencies.py
```