Coverage for mcpgateway / tools / builder / cli.py: 100%

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

2""" 

3Location: ./mcpgateway/tools/builder/cli.py 

4Copyright 2025 

5SPDX-License-Identifier: Apache-2.0 

6Authors: Teryl Taylor 

7 

8MCP Stack Deployment Tool - Hybrid Dagger/Python Implementation 

9 

10This script can run in two modes: 

111. Plain Python mode (default) - No external dependencies 

122. Dagger mode (opt-in) - Requires dagger-io package, auto-downloads CLI 

13 

14Usage: 

15 # Local execution (plain Python mode) 

16 cforge deploy deploy.yaml 

17 

18 # Use Dagger mode for optimization (requires dagger-io, auto-downloads CLI) 

19 cforge --dagger deploy deploy.yaml 

20 

21 # Inside container 

22 docker run -v $PWD:/workspace mcpgateway/mcp-builder:latest deploy deploy.yaml 

23 

24Features: 

25 - Validates deploy.yaml configuration 

26 - Builds plugin containers from git repos 

27 - Generates mTLS certificates 

28 - Deploys to Kubernetes or Docker Compose 

29 - Integrates with CI/CD vault secrets 

30 

31Examples: 

32 >>> # Test that IN_CONTAINER detection works 

33 >>> import os 

34 >>> isinstance(IN_CONTAINER, bool) 

35 True 

36 

37 >>> # Test that BUILDER_DIR is a Path 

38 >>> from pathlib import Path 

39 >>> isinstance(BUILDER_DIR, Path) 

40 True 

41 

42 >>> # Test IMPL_MODE is set 

43 >>> isinstance(IMPL_MODE, str) 

44 True 

45""" 

46 

47# Standard 

48import asyncio 

49import os 

50from pathlib import Path 

51import sys 

52from typing import Optional 

53 

54# Third-Party 

55from rich.console import Console 

56from rich.panel import Panel 

57import typer 

58from typing_extensions import Annotated 

59 

60# First-Party 

61from mcpgateway.tools.builder.factory import DeployFactory 

62 

63app = typer.Typer( 

64 help="Command line tools for deploying the gateway and plugins via a config file.", 

65) 

66 

67console = Console() 

68 

69deployer = None 

70 

71IN_CONTAINER = os.path.exists("/.dockerenv") or os.environ.get("CONTAINER") == "true" 

72BUILDER_DIR = Path(__file__).parent / "builder" 

73IMPL_MODE = "plain" 

74 

75 

76@app.callback() 

77def cli( 

78 ctx: typer.Context, 

79 dagger: Annotated[bool, typer.Option("--dagger", help="Use Dagger mode (requires dagger-io package)")] = False, 

80 verbose: Annotated[bool, typer.Option("--verbose", "-v", help="Verbose output")] = False, 

81): 

82 """MCP Stack deployment tool 

83 

84 Deploys MCP Gateway + external plugins from a single YAML configuration. 

85 

86 By default, uses plain Python mode. Use --dagger to enable Dagger optimization. 

87 

88 Args: 

89 ctx: Typer context object 

90 dagger: Enable Dagger mode (requires dagger-io package and auto-downloads CLI) 

91 verbose: Enable verbose output 

92 """ 

93 ctx.ensure_object(dict) 

94 ctx.obj["verbose"] = verbose 

95 ctx.obj["dagger"] = dagger 

96 

97 if ctx.invoked_subcommand != "version": 

98 # Show execution mode - default to Python, opt-in to Dagger 

99 mode = "dagger" if dagger else "python" 

100 ctx.obj["deployer"], ctx.obj["mode"] = DeployFactory.create_deployer(mode, verbose) 

101 mode_color = "green" if ctx.obj["mode"] == "dagger" else "yellow" 

102 env_text = "container" if IN_CONTAINER else "local" 

103 

104 if verbose: 

105 console.print(Panel(f"[bold]Mode:[/bold] [{mode_color}]{ctx.obj['mode']}[/{mode_color}]\n" f"[bold]Environment:[/bold] {env_text}\n", title="MCP Deploy", border_style=mode_color)) 

106 

107 

108@app.command() 

109def validate(ctx: typer.Context, config_file: Annotated[Path, typer.Argument(help="The deployment configuration file.")]): 

110 """Validate mcp-stack.yaml configuration 

111 

112 Args: 

113 ctx: Typer context object 

114 config_file: Path to the deployment configuration file 

115 """ 

116 impl = ctx.obj["deployer"] 

117 

118 try: 

119 impl.validate(config_file) 

120 console.print("[green]✓ Configuration valid[/green]") 

121 except Exception as e: 

122 console.print(f"[red]✗ Validation failed: {e}[/red]") 

123 sys.exit(1) 

124 

125 

126@app.command() 

127def build( 

128 ctx: typer.Context, 

129 config_file: Annotated[Path, typer.Argument(help="The deployment configuration file")], 

130 plugins_only: Annotated[bool, typer.Option("--plugins-only", help="Only build plugin containers")] = False, 

131 plugin: Annotated[Optional[list[str]], typer.Option("--plugin", "-p", help="Build specific plugin(s)")] = None, 

132 no_cache: Annotated[bool, typer.Option("--no-cache", help="Disable build cache")] = False, 

133 copy_env_templates: Annotated[bool, typer.Option("--copy-env-templates", help="Copy .env.template files from plugin repos")] = True, 

134): 

135 """Build containers 

136 

137 Args: 

138 ctx: Typer context object 

139 config_file: Path to the deployment configuration file 

140 plugins_only: Only build plugin containers, skip gateway 

141 plugin: List of specific plugin names to build 

142 no_cache: Disable build cache 

143 copy_env_templates: Copy .env.template files from plugin repos 

144 """ 

145 impl = ctx.obj["deployer"] 

146 

147 try: 

148 asyncio.run(impl.build(config_file, plugins_only=plugins_only, specific_plugins=list(plugin) if plugin else None, no_cache=no_cache, copy_env_templates=copy_env_templates)) 

149 console.print("[green]✓ Build complete[/green]") 

150 

151 if copy_env_templates: 

152 console.print("[yellow]⚠ IMPORTANT: Review .env files in deploy/env/ before deploying![/yellow]") 

153 console.print("[yellow] Update any required configuration values.[/yellow]") 

154 except Exception as e: 

155 console.print(f"[red]✗ Build failed: {e}[/red]") 

156 sys.exit(1) 

157 

158 

159@app.command() 

160def certs(ctx: typer.Context, config_file: Annotated[Path, typer.Argument(help="The deployment configuration file")]): 

161 """Generate mTLS certificates 

162 

163 Args: 

164 ctx: Typer context object 

165 config_file: Path to the deployment configuration file 

166 """ 

167 impl = ctx.obj["deployer"] 

168 

169 try: 

170 asyncio.run(impl.generate_certificates(config_file)) 

171 console.print("[green]✓ Certificates generated[/green]") 

172 except Exception as e: 

173 console.print(f"[red]✗ Certificate generation failed: {e}[/red]") 

174 sys.exit(1) 

175 

176 

177@app.command() 

178def deploy( 

179 ctx: typer.Context, 

180 config_file: Annotated[Path, typer.Argument(help="The deployment configuration file")], 

181 output_dir: Annotated[Optional[Path], typer.Option("--output-dir", "-o", help="The deployment configuration file")] = None, 

182 dry_run: Annotated[bool, typer.Option("--dry-run", help="Generate manifests without deploying")] = False, 

183 skip_build: Annotated[bool, typer.Option("--skip-build", help="Skip building containers")] = False, 

184 skip_certs: Annotated[bool, typer.Option("--skip-certs", help="Skip certificate generation")] = False, 

185): 

186 """Deploy MCP stack 

187 

188 Args: 

189 ctx: Typer context object 

190 config_file: Path to the deployment configuration file 

191 output_dir: Custom output directory for manifests 

192 dry_run: Generate manifests without deploying 

193 skip_build: Skip building containers 

194 skip_certs: Skip certificate generation 

195 """ 

196 impl = ctx.obj["deployer"] 

197 

198 try: 

199 asyncio.run(impl.deploy(config_file, dry_run=dry_run, skip_build=skip_build, skip_certs=skip_certs, output_dir=output_dir)) 

200 if dry_run: 

201 console.print("[yellow]✓ Dry-run complete (no changes made)[/yellow]") 

202 else: 

203 console.print("[green]✓ Deployment complete[/green]") 

204 except Exception as e: 

205 console.print(f"[red]✗ Deployment failed: {e}[/red]") 

206 sys.exit(1) 

207 

208 

209@app.command() 

210def verify( 

211 ctx: typer.Context, 

212 config_file: Annotated[Path, typer.Argument(help="The deployment configuration file")], 

213 wait: Annotated[bool, typer.Option("--wait", help="Wait for deployment to be ready")] = True, 

214 timeout: Annotated[int, typer.Option("--timeout", help="Wait timeout in seconds")] = 300, 

215): 

216 """Verify deployment health 

217 

218 Args: 

219 ctx: Typer context object 

220 config_file: Path to the deployment configuration file 

221 wait: Wait for deployment to be ready 

222 timeout: Wait timeout in seconds 

223 """ 

224 impl = ctx.obj["deployer"] 

225 

226 try: 

227 asyncio.run(impl.verify(config_file, wait=wait, timeout=timeout)) 

228 console.print("[green]✓ Deployment healthy[/green]") 

229 except Exception as e: 

230 console.print(f"[red]✗ Verification failed: {e}[/red]") 

231 sys.exit(1) 

232 

233 

234@app.command() 

235def destroy( 

236 ctx: typer.Context, 

237 config_file: Annotated[Path, typer.Argument(help="The deployment configuration file")], 

238 force: Annotated[bool, typer.Option("--force", help="Force destruction without confirmation")] = False, 

239): 

240 """Destroy deployed MCP stack 

241 

242 Args: 

243 ctx: Typer context object 

244 config_file: Path to the deployment configuration file 

245 force: Force destruction without confirmation 

246 """ 

247 impl = ctx.obj["deployer"] 

248 

249 if not force: 

250 if not typer.confirm("Are you sure you want to destroy the deployment?"): 

251 console.print("[yellow]Aborted[/yellow]") 

252 return 

253 

254 try: 

255 asyncio.run(impl.destroy(config_file)) 

256 console.print("[green]✓ Deployment destroyed[/green]") 

257 except Exception as e: 

258 console.print(f"[red]✗ Destruction failed: {e}[/red]") 

259 sys.exit(1) 

260 

261 

262@app.command() 

263def version(): 

264 """Show version information 

265 

266 Examples: 

267 >>> # Test that version function exists 

268 >>> callable(version) 

269 True 

270 

271 >>> # Test that it accesses module constants 

272 >>> IMPL_MODE in ['plain', 'dagger'] 

273 True 

274 """ 

275 console.print( 

276 Panel(f"[bold]MCP Deploy[/bold]\n" f"Version: 1.0.0\n" f"Mode: {IMPL_MODE}\n" f"Environment: {'container' if IN_CONTAINER else 'local'}\n", title="Version Info", border_style="blue") 

277 ) 

278 

279 

280@app.command() 

281def generate( 

282 ctx: typer.Context, 

283 config_file: Annotated[Path, typer.Argument(help="The deployment configuration file")], 

284 output: Annotated[Optional[Path], typer.Option("--output", "-o", help="Output directory for manifests")] = None, 

285): 

286 """Generate deployment manifests (k8s or compose) 

287 

288 Args: 

289 ctx: Typer context object 

290 config_file: Path to the deployment configuration file 

291 output: Output directory for manifests 

292 """ 

293 impl = ctx.obj["deployer"] 

294 

295 try: 

296 manifests_dir = impl.generate_manifests(config_file, output_dir=output) 

297 console.print(f"[green]✓ Manifests generated: {manifests_dir}[/green]") 

298 except Exception as e: 

299 console.print(f"[red]✗ Manifest generation failed: {e}[/red]") 

300 sys.exit(1) 

301 

302 

303def main(): 

304 """Main entry point 

305 

306 Raises: 

307 Exception: Any unhandled exception from subcommands (re-raised in debug mode) 

308 

309 Examples: 

310 >>> # Test that main function exists and is callable 

311 >>> callable(main) 

312 True 

313 

314 >>> # Test that app is a Typer instance 

315 >>> import typer 

316 >>> isinstance(app, typer.Typer) 

317 True 

318 

319 >>> # Test that console is available 

320 >>> from rich.console import Console 

321 >>> isinstance(console, Console) 

322 True 

323 """ 

324 try: 

325 app(obj={}) 

326 except KeyboardInterrupt: 

327 console.print("\n[yellow]Interrupted by user[/yellow]") 

328 sys.exit(130) 

329 except Exception as e: 

330 console.print(f"[red]Fatal error: {e}[/red]") 

331 if os.environ.get("MCP_DEBUG"): 

332 raise 

333 sys.exit(1) 

334 

335 

336if __name__ == "__main__": 

337 main()