"""MCP server command with streamable HTTP transport."""
import os
from typing import Any, Optional
import typer
from loguru import logger
from basic_memory.cli.app import app
from basic_memory.config import ConfigManager, init_mcp_logging
class _DeferredMcpServer:
def run(self, *args: Any, **kwargs: Any) -> None: # pragma: no cover
from basic_memory.mcp.server import mcp as live_mcp_server
live_mcp_server.run(*args, **kwargs)
# Keep module-level attribute for tests/monkeypatching while deferring heavy import.
mcp_server = _DeferredMcpServer()
@app.command()
def mcp(
transport: str = typer.Option("stdio", help="Transport type: stdio, streamable-http, or sse"),
host: str = typer.Option(
"0.0.0.0", help="Host for HTTP transports (use 0.0.0.0 to allow external connections)"
),
port: int = typer.Option(8000, help="Port for HTTP transports"),
path: str = typer.Option("/mcp", help="Path prefix for streamable-http transport"),
project: Optional[str] = typer.Option(None, help="Restrict MCP server to single project"),
): # pragma: no cover
"""Run the MCP server with configurable transport options.
This command starts an MCP server using one of three transport options:
- stdio: Standard I/O (good for local usage)
- streamable-http: Recommended for web deployments (default)
- sse: Server-Sent Events (for compatibility with existing clients)
Initialization, file sync, and cleanup are handled by the MCP server's lifespan.
Note: This command is available regardless of cloud mode setting.
Users who have cloud mode enabled can still use local MCP for Claude Code
and Claude Desktop while using cloud MCP for web and mobile access.
"""
# --- Routing setup ---
# Trigger: MCP server command invocation.
# Why: HTTP/SSE transports serve as local API endpoints and must never
# route through cloud. Stdio is a client-facing protocol that
# should honor per-project routing (local or cloud).
# Outcome: HTTP/SSE get explicit local override; stdio passes through
# whatever env vars are already set (honoring external overrides)
# and defaults to per-project routing resolution.
if transport in ("streamable-http", "sse"):
os.environ["BASIC_MEMORY_FORCE_LOCAL"] = "true"
os.environ.pop("BASIC_MEMORY_FORCE_CLOUD", None)
os.environ["BASIC_MEMORY_EXPLICIT_ROUTING"] = "true"
# stdio: no env var manipulation — per-project routing applies by default,
# and externally-set env vars (e.g. BASIC_MEMORY_FORCE_CLOUD) are honored.
# Import mcp tools/prompts to register them with the server
import basic_memory.mcp.tools # noqa: F401 # pragma: no cover
import basic_memory.mcp.prompts # noqa: F401 # pragma: no cover
import basic_memory.mcp.resources # noqa: F401 # pragma: no cover
# Initialize logging for MCP (file only, stdout breaks protocol)
init_mcp_logging()
# Validate and set project constraint if specified
if project:
config_manager = ConfigManager()
project_name, _ = config_manager.get_project(project)
if not project_name:
typer.echo(f"No project found named: {project}", err=True)
raise typer.Exit(1)
# Set env var with validated project name
os.environ["BASIC_MEMORY_MCP_PROJECT"] = project_name
logger.info(f"MCP server constrained to project: {project_name}")
# Run the MCP server (blocks)
# Lifespan handles: initialization, migrations, file sync, cleanup
logger.info(f"Starting MCP server with {transport.upper()} transport")
if transport == "stdio":
mcp_server.run(
transport=transport,
)
elif transport == "streamable-http" or transport == "sse":
mcp_server.run(
transport=transport,
host=host,
port=port,
path=path,
log_level="INFO",
)
