Coverage for src/qdrant_loader_mcp_server/cli.py: 79%
80 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-20 10:15 +0000
1"""CLI module for QDrant Loader MCP Server."""
3import json
4import logging
5import os
6import sys
7from pathlib import Path
9import click
10from click.decorators import option
11from click.types import Choice
12from click.types import Path as ClickPath
13from dotenv import load_dotenv
15from .config_loader import load_config, redact_effective_config
16from .utils import LoggingConfig, get_version
18# Suppress asyncio debug messages to reduce noise in logs.
19logging.getLogger("asyncio").setLevel(logging.WARNING)
22def _setup_logging(log_level: str, transport: str | None = None) -> None:
23 """Set up logging configuration."""
24 try:
25 # Force-disable console logging in stdio mode to avoid polluting stdout
26 if transport and transport.lower() == "stdio":
27 os.environ["MCP_DISABLE_CONSOLE_LOGGING"] = "true"
29 # Check if console logging is disabled via environment variable (after any override)
30 disable_console_logging = (
31 os.getenv("MCP_DISABLE_CONSOLE_LOGGING", "").lower() == "true"
32 )
34 # Reset any pre-existing handlers to prevent duplicate logs when setup() is
35 # invoked implicitly during module imports before CLI config is applied.
36 root_logger = logging.getLogger()
37 for h in list(root_logger.handlers):
38 try:
39 root_logger.removeHandler(h)
40 except Exception:
41 pass
43 # Use reconfigure if available to avoid stacking handlers on repeated setup
44 level = log_level.upper()
45 if getattr(LoggingConfig, "reconfigure", None): # type: ignore[attr-defined]
46 if getattr(LoggingConfig, "_initialized", False): # type: ignore[attr-defined]
47 # Only switch file target (none in stdio; may be env provided)
48 LoggingConfig.reconfigure(file=os.getenv("MCP_LOG_FILE")) # type: ignore[attr-defined]
49 else:
50 LoggingConfig.setup(
51 level=level,
52 format=("json" if disable_console_logging else "console"),
53 )
54 else:
55 # Force replace handlers on older versions
56 logging.getLogger().handlers = []
57 LoggingConfig.setup(
58 level=level, format=("json" if disable_console_logging else "console")
59 )
60 except Exception as e:
61 print(f"Failed to setup logging: {e}", file=sys.stderr)
64@click.command(name="mcp-qdrant-loader")
65@option(
66 "--log-level",
67 type=Choice(
68 ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"], case_sensitive=False
69 ),
70 default="INFO",
71 help="Set the logging level.",
72)
73# Hidden option to print effective config (redacts secrets)
74@option(
75 "--print-config",
76 is_flag=True,
77 default=False,
78 help="Print the effective configuration (secrets redacted) and exit.",
79)
80@option(
81 "--config",
82 type=ClickPath(exists=True, path_type=Path),
83 help="Path to configuration file.",
84)
85@option(
86 "--transport",
87 type=Choice(["stdio", "http"], case_sensitive=False),
88 default="stdio",
89 help="Transport protocol to use (stdio for JSON-RPC over stdin/stdout, http for streamable HTTP)",
90)
91@option(
92 "--host",
93 type=str,
94 default="127.0.0.1",
95 help="Host to bind HTTP server to (only used with --transport http)",
96)
97@option(
98 "--port",
99 type=int,
100 default=8080,
101 help="Port to bind HTTP server to (only used with --transport http)",
102)
103@option(
104 "--workers",
105 type=int,
106 default=1,
107 help="Number of uvicorn worker processes (only used with --transport http)",
108)
109@option(
110 "--env",
111 type=ClickPath(exists=True, path_type=Path),
112 help="Path to .env file to load environment variables from",
113)
114@click.version_option(
115 version=get_version(),
116 message="QDrant Loader MCP Server v%(version)s",
117)
118def cli(
119 log_level: str = "INFO",
120 config: Path | None = None,
121 transport: str = "stdio",
122 host: str = "127.0.0.1",
123 port: int = 8080,
124 workers: int = 1,
125 env: Path | None = None,
126 print_config: bool = False,
127) -> None:
128 """QDrant Loader MCP Server.
130 A Model Context Protocol (MCP) server that provides RAG capabilities
131 to Cursor and other LLM applications using Qdrant vector database.
133 The server is built on FastMCP and supports both stdio (JSON-RPC over
134 stdin/stdout) and HTTP (streamable) transports.
136 Environment Variables:
137 QDRANT_URL: URL of your QDrant instance (required)
138 QDRANT_API_KEY: API key for QDrant authentication
139 QDRANT_COLLECTION_NAME: Name of the collection to use (default: "documents")
140 OPENAI_API_KEY: OpenAI API key for embeddings (required)
141 MCP_DISABLE_CONSOLE_LOGGING: Set to "true" to disable console logging
143 Examples:
144 # Start with stdio transport (default, for Cursor/Claude Desktop)
145 mcp-qdrant-loader
147 # Start with HTTP transport (for web clients)
148 mcp-qdrant-loader --transport http --port 8080
150 # Start with environment variables from .env file
151 mcp-qdrant-loader --transport http --env /path/to/.env
153 # Start with debug logging
154 mcp-qdrant-loader --log-level DEBUG --transport http
156 # Show help
157 mcp-qdrant-loader --help
159 # Show version
160 mcp-qdrant-loader --version
161 """
162 try:
163 # Load environment variables from .env file if specified
164 if env:
165 load_dotenv(env)
167 # Setup logging (force-disable console logging in stdio transport)
168 _setup_logging(log_level, transport)
170 # Log env file load after logging is configured to avoid duplicate handler setup
171 if env:
172 LoggingConfig.get_logger(__name__).info(
173 "Loaded environment variables", env=str(env)
174 )
176 # If a config file was provided, propagate it via MCP_CONFIG so that
177 # the FastMCP lifespan (which resolves config independently) can find it.
178 if config is not None:
179 try:
180 os.environ["MCP_CONFIG"] = str(config)
181 except Exception:
182 # Best-effort; continue without blocking startup
183 pass
185 # Resolve configuration early (fail fast; also powers --print-config).
186 _, effective_cfg, _ = load_config(config)
188 if print_config:
189 redacted = redact_effective_config(effective_cfg)
190 click.echo(json.dumps(redacted, indent=2))
191 return
193 if transport.lower() == "http":
194 import uvicorn
196 os.environ["MCP_LOG_LEVEL"] = log_level
197 os.environ["MCP_HOST"] = host
198 os.environ["MCP_PORT"] = str(port)
200 logger = LoggingConfig.get_logger(__name__)
201 logger.info(
202 "Starting HTTP server (FastMCP)",
203 host=host,
204 port=port,
205 log_level=log_level,
206 )
208 uvicorn.run(
209 "qdrant_loader_mcp_server.fastmcp_app:http_app",
210 host=host,
211 port=port,
212 workers=workers,
213 log_level=log_level.lower(),
214 access_log=(log_level.upper() == "DEBUG"),
215 )
216 elif transport.lower() == "stdio":
217 logger = LoggingConfig.get_logger(__name__)
218 logger.info("Starting stdio transport (FastMCP)")
219 from .fastmcp_app import mcp
221 mcp.run(transport="stdio", show_banner=False)
222 else:
223 raise ValueError(f"Unsupported transport: {transport}")
224 except Exception:
225 logger = LoggingConfig.get_logger(__name__)
226 logger.error("Error in main", exc_info=True)
227 sys.exit(1)
230if __name__ == "__main__":
231 cli()