Coverage for mcpgateway / cli.py: 100%

1# -*- coding: utf-8 -*- 

2"""Location: ./mcpgateway/cli.py 

3Copyright 2025 

4SPDX-License-Identifier: Apache-2.0 

5Authors: Mihai Criveti 

6 

7mcpgateway CLI ─ a thin wrapper around Uvicorn 

8This module is exposed as a **console-script** via: 

9 

10 [project.scripts] 

11 mcpgateway = "mcpgateway.cli:main" 

12 

13so that a user can simply type `mcpgateway ...` instead of the longer 

14`uvicorn mcpgateway.main:app ...`. 

15 

16Features 

17───────── 

18* Injects the default FastAPI application path (``mcpgateway.main:app``) 

19 when the user doesn't supply one explicitly. 

20* Adds sensible default host/port (127.0.0.1:4444) unless the user passes 

21 ``--host``/``--port`` or overrides them via the environment variables 

22 ``MCG_HOST`` and ``MCG_PORT``. 

23* Forwards *all* remaining arguments verbatim to Uvicorn's own CLI, so 

24 `--reload`, `--workers`, etc. work exactly the same. 

25 

26Typical usage 

27───────────── 

28```console 

29$ mcpgateway --reload # dev server on 127.0.0.1:4444 

30$ mcpgateway --workers 4 # production-style multiprocess 

31$ mcpgateway 127.0.0.1:8000 --reload # explicit host/port keeps defaults out 

32$ mcpgateway mypkg.other:app # run a different ASGI callable 

33``` 

34""" 

35 

36# Future 

37from __future__ import annotations 

38 

39# Standard 

40import os 

41from pathlib import Path 

42import sys 

43from typing import List, Optional 

44 

45# Third-Party 

46import orjson 

47from pydantic import ValidationError 

48import uvicorn 

49 

50# First-Party 

51from mcpgateway import __version__ 

52from mcpgateway.config import Settings 

53 

54# --------------------------------------------------------------------------- 

55# Configuration defaults (overridable via environment variables) 

56# --------------------------------------------------------------------------- 

57DEFAULT_APP = "mcpgateway.main:app" # dotted path to FastAPI instance 

58DEFAULT_HOST = os.getenv("MCG_HOST", "127.0.0.1") 

59DEFAULT_PORT = int(os.getenv("MCG_PORT", "4444")) 

60 

61# --------------------------------------------------------------------------- 

62# Helper utilities 

63# --------------------------------------------------------------------------- 

64 

65 

66def _needs_app(arg_list: List[str]) -> bool: 

67 """Return *True* when the CLI invocation has *no* positional APP path. 

68 

69 According to Uvicorn's argument grammar, the **first** non-flag token 

70 is taken as the application path. We therefore look at the first 

71 element of *arg_list* (if any) - if it *starts* with a dash it must be 

72 an option, hence the app path is missing and we should inject ours. 

73 

74 Args: 

75 arg_list (List[str]): List of arguments 

76 

77 Returns: 

78 bool: Returns *True* when the CLI invocation has *no* positional APP path 

79 

80 Examples: 

81 >>> _needs_app([]) 

82 True 

83 >>> _needs_app(["--reload"]) 

84 True 

85 >>> _needs_app(["myapp.main:app"]) 

86 False 

87 """ 

88 

89 return len(arg_list) == 0 or arg_list[0].startswith("-") 

90 

91 

92def _insert_defaults(raw_args: List[str]) -> List[str]: 

93 """Return a *new* argv with defaults sprinkled in where needed. 

94 

95 Args: 

96 raw_args (List[str]): List of input arguments to cli 

97 

98 Returns: 

99 List[str]: List of arguments 

100 

101 Examples: 

102 >>> result = _insert_defaults([]) 

103 >>> result[0] 

104 'mcpgateway.main:app' 

105 >>> result = _insert_defaults(["myapp.main:app", "--reload"]) 

106 >>> result[0] 

107 'myapp.main:app' 

108 """ 

109 

110 args = list(raw_args) # shallow copy - we'll mutate this 

111 

112 # 1️⃣ Ensure an application path is present. 

113 if _needs_app(args): 

114 args.insert(0, DEFAULT_APP) 

115 

116 # 2️⃣ Supply host/port if neither supplied nor UNIX domain socket. 

117 if "--uds" not in args: 

118 if "--host" not in args and "--http" not in args: 

119 args.extend(["--host", DEFAULT_HOST]) 

120 if "--port" not in args: 

121 args.extend(["--port", str(DEFAULT_PORT)]) 

122 

123 return args 

124 

125 

126def _handle_validate_config(path: str = ".env") -> None: 

127 """ 

128 Validate the application's environment configuration file. 

129 

130 Attempts to load settings from the specified .env file using Pydantic. 

131 If validation fails, prints the errors and exits with code 1. 

132 On success, prints a confirmation message. 

133 

134 Args: 

135 path (str): Path to the .env file to validate. Defaults to ".env". 

136 

137 Raises: 

138 SystemExit: Exits with code 1 if the configuration is invalid. 

139 

140 Examples: 

141 >>> _handle_validate_config(".env.example") 

142 ✅ Configuration in .env.example is valid 

143 """ 

144 

145 try: 

146 Settings(_env_file=path) 

147 except ValidationError as exc: 

148 print(f"❌ Invalid configuration in {path}", file=sys.stderr) 

149 print(exc.json(indent=2), file=sys.stderr) 

150 raise SystemExit(1) 

151 

152 print(f"✅ Configuration in {path} is valid") 

153 

154 

155def _handle_config_schema(output: Optional[str] = None) -> None: 

156 """ 

157 Export the JSON schema for MCP Gateway Settings. 

158 

159 This function serializes the Pydantic Settings model into a JSON Schema 

160 suitable for validation or documentation purposes. 

161 

162 Args: 

163 output (Optional[str]): Optional file path to write the schema. 

164 If None, prints to stdout. 

165 

166 Examples: 

167 >>> # Print schema to stdout (output truncated for doctest) 

168 >>> _handle_config_schema() # doctest: +ELLIPSIS 

169 {... 

170 

171 >>> # Write schema to a file (creates 'schema.json'), skip doctest 

172 >>> _handle_config_schema("schema.json") # doctest: +SKIP 

173 ✅ Schema written to schema.json 

174 """ 

175 schema = Settings.model_json_schema(mode="validation") 

176 data = orjson.dumps(schema, option=orjson.OPT_INDENT_2 | orjson.OPT_SORT_KEYS).decode() 

177 

178 if output: 

179 path = Path(output) 

180 path.write_text(data, encoding="utf-8") 

181 print(f"✅ Schema written to {path}") 

182 else: 

183 print(data) 

184 

185 

186def _handle_support_bundle( 

187 output_dir: Optional[str] = None, 

188 log_lines: int = 1000, 

189 include_logs: bool = True, 

190 include_env: bool = True, 

191 include_system: bool = True, 

192) -> None: 

193 """ 

194 Generate a support bundle containing diagnostics and logs. 

195 

196 Creates a ZIP file with version info, system diagnostics, configuration, 

197 and logs - all automatically sanitized to remove sensitive data like 

198 passwords, tokens, and API keys. 

199 

200 Args: 

201 output_dir (Optional[str]): Directory for bundle output (default: /tmp) 

202 log_lines (int): Number of log lines to include (default: 1000, 0 = all) 

203 include_logs (bool): Include log files (default: True) 

204 include_env (bool): Include environment config (default: True) 

205 include_system (bool): Include system info (default: True) 

206 

207 Raises: 

208 SystemExit: If bundle generation fails 

209 

210 Examples: 

211 >>> # Generate bundle with default settings 

212 >>> _handle_support_bundle() # doctest: +SKIP 

213 ✅ Support bundle created: /tmp/mcpgateway-support-2025-01-09-120000.zip 

214 

215 >>> # Generate bundle with custom settings 

216 >>> _handle_support_bundle(output_dir="/tmp", log_lines=500) # doctest: +SKIP 

217 ✅ Support bundle created: /tmp/mcpgateway-support-2025-01-09-120000.zip 

218 """ 

219 # First-Party 

220 from mcpgateway.services.support_bundle_service import SupportBundleConfig, SupportBundleService # pylint: disable=import-outside-toplevel 

221 

222 try: 

223 config = SupportBundleConfig( 

224 include_logs=include_logs, 

225 include_env=include_env, 

226 include_system_info=include_system, 

227 log_tail_lines=log_lines, 

228 output_dir=Path(output_dir) if output_dir else None, 

229 ) 

230 

231 service = SupportBundleService() 

232 bundle_path = service.generate_bundle(config) 

233 

234 print(f"✅ Support bundle created: {bundle_path}") 

235 print(f"📦 Bundle size: {bundle_path.stat().st_size / 1024:.2f} KB") 

236 print() 

237 print("⚠️ Security Notice:") 

238 print(" The bundle has been sanitized, but please review before sharing.") 

239 print(" Sensitive data (passwords, tokens, secrets) have been redacted.") 

240 except Exception as exc: 

241 print(f"❌ Failed to create support bundle: {exc}", file=sys.stderr) 

242 raise SystemExit(1) 

243 

244 

245# --------------------------------------------------------------------------- 

246# Public entry-point 

247# --------------------------------------------------------------------------- 

248 

249 

250def main() -> None: # noqa: D401 - imperative mood is fine here 

251 """Entry point for the *mcpgateway* console script (delegates to Uvicorn). 

252 

253 Processes command line arguments, handles version requests, and forwards 

254 all other arguments to Uvicorn with sensible defaults injected. 

255 

256 Also supports export/import subcommands for configuration management. 

257 

258 Environment Variables: 

259 MCG_HOST: Default host (default: "127.0.0.1") 

260 MCG_PORT: Default port (default: "4444") 

261 

262 Usage: 

263 mcpgateway --reload 

264 mcpgateway --workers 4 

265 mcpgateway --validate-config [path] 

266 mcpgateway --config-schema [output] 

267 mcpgateway --support-bundle [options] 

268 

269 Flags: 

270 --validate-config [path] Validate .env file (default: .env) 

271 --config-schema [output] Print or write JSON schema for Settings 

272 --support-bundle Generate support bundle for troubleshooting 

273 --output-dir [path] Output directory (default: /tmp) 

274 --log-lines [n] Number of log lines (default: 1000, 0 = all) 

275 --no-logs Exclude log files 

276 --no-env Exclude environment config 

277 --no-system Exclude system info 

278 """ 

279 

280 # Check for export/import commands first 

281 if len(sys.argv) > 1 and sys.argv[1] in ["export", "import"]: 

282 # Avoid cyclic import by importing only when needed 

283 # First-Party 

284 from mcpgateway.cli_export_import import main_with_subcommands # pylint: disable=import-outside-toplevel,cyclic-import 

285 

286 main_with_subcommands() 

287 return 

288 

289 # Check for version flag 

290 if "--version" in sys.argv or "-V" in sys.argv: 

291 print(f"mcpgateway {__version__}") 

292 return 

293 

294 # Handle config-related flags 

295 if len(sys.argv) > 1: 

296 cmd = sys.argv[1] 

297 

298 if cmd == "--validate-config": 

299 env_path = sys.argv[2] if len(sys.argv) > 2 else ".env" 

300 _handle_validate_config(env_path) 

301 return 

302 

303 if cmd == "--config-schema": 

304 output = sys.argv[2] if len(sys.argv) > 2 else None 

305 _handle_config_schema(output) 

306 return 

307 

308 if cmd == "--support-bundle": 

309 # Parse support bundle options 

310 output_dir = None 

311 log_lines = 1000 

312 include_logs = True 

313 include_env = True 

314 include_system = True 

315 

316 i = 2 

317 while i < len(sys.argv): 

318 arg = sys.argv[i] 

319 if arg == "--output-dir" and i + 1 < len(sys.argv): 

320 output_dir = sys.argv[i + 1] 

321 i += 2 

322 elif arg == "--log-lines" and i + 1 < len(sys.argv): 

323 log_lines = int(sys.argv[i + 1]) 

324 i += 2 

325 elif arg == "--no-logs": 

326 include_logs = False 

327 i += 1 

328 elif arg == "--no-env": 

329 include_env = False 

330 i += 1 

331 elif arg == "--no-system": 

332 include_system = False 

333 i += 1 

334 else: 

335 i += 1 

336 

337 _handle_support_bundle( 

338 output_dir=output_dir, 

339 log_lines=log_lines, 

340 include_logs=include_logs, 

341 include_env=include_env, 

342 include_system=include_system, 

343 ) 

344 return 

345 

346 # Discard the program name and inspect the rest. 

347 user_args = sys.argv[1:] 

348 uvicorn_argv = _insert_defaults(user_args) 

349 

350 # Uvicorn's `main()` uses sys.argv - patch it in and run. 

351 sys.argv = ["mcpgateway", *uvicorn_argv] 

352 uvicorn.main() # pylint: disable=no-value-for-parameter 

353 

354 

355if __name__ == "__main__": # pragma: no cover - executed only when run directly 

356 main()