Coverage for fastblocks/mcp/config

1"""Interactive CLI for FastBlocks adapter configuration management."""

3import asyncio

4import json

5import os

6import sys

7import typing as t

8from pathlib import Path

9from typing import Any

11import click

12from acb.console import Console

13from acb.depends import depends

14from mcp_common.ui import ServerPanels

15from rich.panel import Panel

16from rich.prompt import Confirm, Prompt

18from .configuration import (

19 ConfigurationManager,

20 ConfigurationProfile,

21 ConfigurationSchema,

22 ConfigurationStatus,

23 ConfigurationValidationResult,

24 EnvironmentVariable,

25)

26from .health import HealthCheckSystem

27from .registry import AdapterRegistry

29console = depends.get_sync(Console)

32class InteractiveConfigurationCLI:

33 """Interactive CLI for adapter configuration."""

35 def __init__(self) -> None:

36 """Initialize the interactive CLI."""

37 self.registry = AdapterRegistry()

38 self.health = HealthCheckSystem(self.registry)

39 self.config_manager = ConfigurationManager(self.registry)

41 async def initialize(self) -> None:

42 """Initialize all systems."""

43 await self.registry.initialize()

44 await self.config_manager.initialize()

46 async def run_configuration_wizard(self) -> ConfigurationSchema:

47 """Run the interactive configuration wizard."""

48 console.print(

49 Panel.fit(

50 "[bold blue]FastBlocks Adapter Configuration Wizard[/bold blue]\n"

51 "This wizard will help you configure FastBlocks adapters for your project.",

52 title="Welcome",

53 )

54 )

56 # Step 1: Select profile

57 profile = self._select_profile()

59 # Step 2: Select adapters

60 selected_adapters = await self._select_adapters()

62 # Step 3: Configure selected adapters

63 config = await self.config_manager.create_configuration(

64 profile=profile, adapters=selected_adapters

65 )

67 # Step 4: Configure each adapter

68 for adapter_name in selected_adapters:

69 await self._configure_adapter_interactive(config, adapter_name)

71 # Step 5: Configure global settings

72 await self._configure_global_settings(config)

74 # Step 6: Validate configuration

75 console.print("\n[yellow]Validating configuration...[/yellow]")

76 validation_result = await self.config_manager.validate_configuration(config)

77 self._display_validation_result(validation_result)

79 if validation_result.status == ConfigurationStatus.ERROR:

80 if not Confirm.ask("Configuration has errors. Continue anyway?"):

81 console.print("[red]Configuration cancelled.[/red]")

82 sys.exit(1)

84 # Step 7: Save configuration

85 config_name = Prompt.ask(

86 "Configuration name", default=f"{profile.value}_config"

87 )

89 config_file = await self.config_manager.save_configuration(config, config_name)

90 console.print(f"\n[green]✓[/green] Configuration saved to: {config_file}")

92 # Step 8: Generate environment file

93 if Confirm.ask("Generate .env file?", default=True):

94 env_file = await self.config_manager.generate_environment_file(config)

95 console.print(f"[green]✓[/green] Environment file created: {env_file}")

97 # Step 9: Create backup

98 if Confirm.ask("Create backup?", default=True):

99 backup_description = Prompt.ask(

100 "Backup description", default="Initial configuration"

101 )

102 backup = await self.config_manager.backup_configuration(

103 config, config_name, backup_description

104 )

105 console.print(f"[green]✓[/green] Backup created: {backup.id}")

106

107 return config

108

109 def _select_profile(self) -> ConfigurationProfile:

110 """Interactive profile selection."""

111 console.print("\n[bold]Select deployment profile:[/bold]")

112

113 profiles = {

114 "1": (

115 ConfigurationProfile.DEVELOPMENT,

116 "Development - Debug enabled, relaxed security",

117 ),

118 "2": (

119 ConfigurationProfile.STAGING,

120 "Staging - Production-like with debug options",

121 ),

122 "3": (

123 ConfigurationProfile.PRODUCTION,

124 "Production - Optimized for performance and security",

125 ),

126 }

127

128 for key, (profile, description) in profiles.items():

129 console.print(

130 f" {key}. [cyan]{profile.value.title()}[/cyan] - {description}"

131 )

132

133 choice = Prompt.ask(

134 "Choose profile", choices=list(profiles.keys()), default="1"

135 )

136

137 selected_profile, _ = profiles[choice]

138 console.print(

139 f"[green]✓[/green] Selected profile: [cyan]{selected_profile.value.title()}[/cyan]"

140 )

141 return selected_profile

142

143 def _group_adapters_by_category(

144 self, available_adapters: dict[str, Any]

145 ) -> dict[str, list[tuple[str, Any]]]:

146 """Group adapters by category.

147

148 Args:

149 available_adapters: Dict of adapter name -> adapter info

150

151 Returns:

152 Dict of category -> list of (adapter_name, adapter_info)

153 """

154 categories: dict[str, list[tuple[str, Any]]] = {}

155

156 for adapter_name, adapter_info in available_adapters.items():

157 category = adapter_info.category

158 if category not in categories:

159 categories[category] = []

160 categories[category].append((adapter_name, adapter_info))

161

162 return categories

163

164 def _display_adapter_choices(

165 self, categories: dict[str, list[tuple[str, Any]]]

166 ) -> dict[str, str]:

167 """Display adapters by category and build choice mapping.

168

169 Args:

170 categories: Dict of category -> list of (adapter_name, adapter_info)

171

172 Returns:

173 Dict mapping choice number to adapter name

174 """

175 adapter_choices = {}

176 choice_num = 1

177

178 for category, adapters in sorted(categories.items()):

179 console.print(f"\n[bold yellow]{category.title()}:[/bold yellow]")

180 for adapter_name, adapter_info in sorted(adapters):

181 status_color = (

182 "green" if adapter_info.module_status == "stable" else "yellow"

183 )

184 console.print(

185 f" {choice_num:2d}. [cyan]{adapter_name:<20}[/cyan] "

186 f"[{status_color}]{adapter_info.module_status:<12}[/{status_color}] "

187 f"{adapter_info.description or 'No description'}"

188 )

189 adapter_choices[str(choice_num)] = adapter_name

190 choice_num += 1

191

192 return adapter_choices

193

194 def _get_recommended_adapters(

195 self, available_adapters: dict[str, Any]

196 ) -> list[str]:

197 """Get recommended default adapters.

198

199 Args:

200 available_adapters: Dict of available adapters

201

202 Returns:

203 List of recommended adapter names

204 """

205 recommended = ["app", "templates", "routes"]

206 selected = [name for name in recommended if name in available_adapters]

207 console.print(

208 f"[green]✓[/green] Using recommended adapters: {', '.join(selected)}"

209 )

210 return selected

211

212 def _parse_adapter_selection(

213 self, selection: str, adapter_choices: dict[str, str]

214 ) -> list[str]:

215 """Parse user's adapter selection.

216

217 Args:

218 selection: Comma-separated choice numbers

219 adapter_choices: Dict mapping choice number to adapter name

220

221 Returns:

222 List of selected adapter names

223 """

224 selected_adapters = []

225

226 for choice in selection.split(","):

227 choice = choice.strip()

228 if choice in adapter_choices:

229 selected_adapters.append(adapter_choices[choice])

230 else:

231 console.print(f"[red]Warning:[/red] Invalid choice '{choice}' ignored")

232

233 # Print result

234 if selected_adapters:

235 console.print(

236 f"[green]✓[/green] Selected adapters: {', '.join(selected_adapters)}"

237 )

238 else:

239 console.print("[red]No adapters selected![/red]")

240

241 return selected_adapters

242

243 async def _select_adapters(self) -> list[str]:

244 """Interactive adapter selection."""

245 console.print("\n[bold]Available Adapters:[/bold]")

246

247 available_adapters = await self.config_manager.get_available_adapters()

248

249 # Group and display adapters

250 categories = self._group_adapters_by_category(available_adapters)

251 adapter_choices = self._display_adapter_choices(categories)

252

253 # Get user selection

254 console.print(

255 "\n[bold]Select adapters (comma-separated numbers, e.g., 1,3,5):[/bold]"

256 )

257 console.print("[dim]Leave empty for recommended defaults[/dim]")

258 selection = Prompt.ask("Adapter numbers", default="")

259

260 # Return recommended or parsed selection

261 if not selection.strip():

262 return self._get_recommended_adapters(available_adapters)

263

264 return self._parse_adapter_selection(selection, adapter_choices)

265

266 def _configure_adapter_env_vars(self, adapter_config: t.Any) -> None:

267 """Configure adapter environment variables."""

268 if adapter_config.environment_variables:

269 console.print("\n[yellow]Environment Variables:[/yellow]")

270 for env_var in adapter_config.environment_variables:

271 self._configure_environment_variable(env_var)

272

273 def _configure_required_settings(

274 self, adapter_config: t.Any, schema: dict[str, t.Any]

275 ) -> None:

276 """Configure required adapter settings."""

277 for setting in schema.get("required_settings", []):

278 value = Prompt.ask(

279 f"[red]*[/red] {setting['name']} ({setting['type']})",

280 default=str(setting.get("default", "")),

281 )

282 adapter_config.settings[setting["name"]] = self._parse_setting_value(

283 value, setting["type"]

284 )

285

286 def _configure_optional_settings(

287 self, adapter_config: t.Any, schema: dict[str, t.Any]

288 ) -> None:

289 """Configure optional adapter settings."""

290 if schema.get("optional_settings") and Confirm.ask(

291 "Configure optional settings?"

292 ):

293 for setting in schema.get("optional_settings", []):

294 if Confirm.ask(f"Configure {setting['name']}?"):

295 value = Prompt.ask(

296 f"{setting['name']} ({setting['type']})",

297 default=str(setting.get("default", "")),

298 )

299 adapter_config.settings[setting["name"]] = (

300 self._parse_setting_value(value, setting["type"])

301 )

302

303 async def _configure_adapter_interactive(

304 self, config: ConfigurationSchema, adapter_name: str

305 ) -> None:

306 """Configure a specific adapter interactively."""

307 console.print(f"\n[bold]Configuring {adapter_name} adapter:[/bold]")

308

309 adapter_config = config.adapters[adapter_name]

310 schema = await self.config_manager.get_adapter_configuration_schema(

311 adapter_name

312 )

313

314 # Show adapter information

315 console.print(f"[dim]Category: {schema.get('category', 'Unknown')}[/dim]")

316 if schema.get("description"):

317 console.print(f"[dim]Description: {schema['description']}[/dim]")

318

319 # Configure environment variables

320 self._configure_adapter_env_vars(adapter_config)

321

322 # Configure settings

323 if schema.get("required_settings") or schema.get("optional_settings"):

324 console.print("\n[yellow]Adapter Settings:[/yellow]")

325 self._configure_required_settings(adapter_config, schema)

326 self._configure_optional_settings(adapter_config, schema)

327

328 # Configure dependencies

329 if schema.get("dependencies"):

330 console.print(

331 f"\n[yellow]Dependencies: {', '.join(schema['dependencies'])}[/yellow]"

332 )

333 adapter_config.dependencies.update(schema["dependencies"])

334

335 def _configure_environment_variable(self, env_var: EnvironmentVariable) -> None:

336 """Configure a single environment variable."""

337 current_value = os.environ.get(

338 env_var.name, env_var.value or env_var.default or ""

339 )

340

341 if env_var.required:

342 prompt_text = f"[red]*[/red] {env_var.name}"

343 else:

344 prompt_text = env_var.name

345

346 if env_var.description:

347 console.print(f"[dim] {env_var.description}[/dim]")

348

349 if env_var.secret:

350 value = Prompt.ask(prompt_text, password=True, default=current_value)

351 else:

352 value = Prompt.ask(prompt_text, default=current_value)

353

354 env_var.value = value

355

356 def _parse_setting_value(self, value: str, type_name: str) -> Any:

357 """Parse setting value based on type."""

358 if not value:

359 return None

360

361 try:

362 if type_name in ("int", "integer"):

363 return int(value)

364 elif type_name in ("float", "number"):

365 return float(value)

366 elif type_name in ("bool", "boolean"):

367 return value.lower() in ("true", "1", "yes", "on")

368 elif type_name in ("list", "array"):

369 return [item.strip() for item in value.split(",") if item.strip()]

370 elif type_name in ("dict", "object"):

371 return json.loads(value)

372

373 return value

374 except (ValueError, json.JSONDecodeError):

375 console.print(

376 f"[red]Warning:[/red] Could not parse '{value}' as {type_name}, using as string"

377 )

378 return value

379

380 async def _configure_global_settings(self, config: ConfigurationSchema) -> None:

381 """Configure global settings."""

382 console.print("\n[bold]Global Settings:[/bold]")

383

384 if Confirm.ask("Configure global settings?"):

385 # Common global settings

386 settings_to_configure = [

387 ("debug", "bool", "Enable debug mode"),

388 ("log_level", "str", "Logging level (DEBUG, INFO, WARNING, ERROR)"),

389 ("secret_key", "str", "Application secret key"),

390 ("database_url", "str", "Database connection URL"),

391 ]

392

393 for setting_name, setting_type, description in settings_to_configure:

394 if Confirm.ask(f"Configure {setting_name}?"):

395 console.print(f"[dim]{description}[/dim]")

396 value = Prompt.ask(

397 setting_name,

398 password=(setting_name in ("secret_key", "database_url")),

399 )

400 config.global_settings[setting_name] = self._parse_setting_value(

401 value, setting_type

402 )

403

404 def _display_validation_result(self, result: ConfigurationValidationResult) -> None:

405 """Display configuration validation results."""

406 if result.status == ConfigurationStatus.VALID:

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

408 elif result.status == ConfigurationStatus.WARNING:

409 console.print("[yellow]⚠ Configuration has warnings[/yellow]")

410 elif result.status == ConfigurationStatus.ERROR:

411 console.print("[red]✗ Configuration has errors[/red]")

412

413 if result.errors:

414 console.print("\n[red]Errors:[/red]")

415 for error in result.errors:

416 console.print(f" [red]•[/red] {error}")

417

418 if result.warnings:

419 console.print("\n[yellow]Warnings:[/yellow]")

420 for warning in result.warnings:

421 console.print(f" [yellow]•[/yellow] {warning}")

424# CLI Commands

427@click.group()

428def config_cli() -> None:

429 """FastBlocks Configuration Management CLI."""

430 pass

431

432

433@config_cli.command()

434def wizard() -> None:

435 """Launch the interactive configuration wizard."""

436

437 async def _wizard() -> None:

438 cli = InteractiveConfigurationCLI()

439 await cli.initialize()

440 await cli.run_configuration_wizard()

441

442 asyncio.run(_wizard())

443

444

445@config_cli.command()

446@click.option(

447 "--profile",

448 type=click.Choice(["development", "staging", "production"]),

449 default="development",

450 help="Configuration profile",

451)

452@click.option("--adapters", help="Comma-separated list of adapters to include")

453@click.option("--output", "-o", help="Output file path")

454def create(profile: str, adapters: str | None, output: str | None) -> None:

455 """Create a new configuration file."""

456

457 async def _create() -> None:

458 registry = AdapterRegistry()

459 config_manager = ConfigurationManager(registry)

460 await registry.initialize()

461 await config_manager.initialize()

462

463 profile_enum = ConfigurationProfile(profile)

464 adapter_list = adapters.split(",") if adapters else None

465

466 config = await config_manager.create_configuration(

467 profile=profile_enum, adapters=adapter_list

468 )

469

470 config_file = await config_manager.save_configuration(config, output)

471 console.print(f"[green]✓[/green] Configuration created: {config_file}")

472

473 asyncio.run(_create())

474

475

476def _format_json_output(result: ConfigurationValidationResult) -> None:

477 """Format validation result as JSON."""

478 output = {

479 "status": result.status.value,

480 "errors": result.errors,

481 "warnings": result.warnings,

482 "adapter_results": result.adapter_results,

483 }

484 click.echo(json.dumps(output, indent=2))

485

486

487def _format_text_output(result: ConfigurationValidationResult) -> None:

488 """Format validation result as human-readable text."""

489 # Print status

490 status_messages = {

491 ConfigurationStatus.VALID: "[green]✓ Configuration is valid[/green]",

492 ConfigurationStatus.WARNING: "[yellow]⚠ Configuration has warnings[/yellow]",

493 ConfigurationStatus.ERROR: "[red]✗ Configuration has errors[/red]",

494 }

495 console.print(status_messages.get(result.status, "Unknown status"))

496

497 # Print errors

498 if result.errors:

499 console.print("\n[red]Errors:[/red]")

500 for error in result.errors:

501 console.print(f" • {error}")

502

503 # Print warnings

504 if result.warnings:

505 console.print("\n[yellow]Warnings:[/yellow]")

506 for warning in result.warnings:

507 console.print(f" • {warning}")

508

509

510@config_cli.command()

511@click.argument("config_file")

512@click.option("--format", type=click.Choice(["text", "json"]), default="text")

513def validate(config_file: str, format: str) -> None:

514 """Validate a configuration file."""

515

516 async def _validate() -> None:

517 registry = AdapterRegistry()

518 config_manager = ConfigurationManager(registry)

519 await registry.initialize()

520 await config_manager.initialize()

521

522 try:

523 config = await config_manager.load_configuration(config_file)

524 result = await config_manager.validate_configuration(config)

525

526 # Format output based on requested format

527 if format == "json":

528 _format_json_output(result)

529 else:

530 _format_text_output(result)

531

532 except FileNotFoundError:

533 console.print(

534 f"[red]Error:[/red] Configuration file '{config_file}' not found"

535 )

536 sys.exit(1)

537 except Exception as e:

538 console.print(f"[red]Error:[/red] {e}")

539 sys.exit(1)

540

541 asyncio.run(_validate())

542

543

544@config_cli.command()

545@click.argument("config_file")

546@click.option("--output", "-o", help="Output .env file path")

547def generate_env(config_file: str, output: str | None) -> None:

548 """Generate .env file from configuration."""

549

550 async def _generate_env() -> None:

551 registry = AdapterRegistry()

552 config_manager = ConfigurationManager(registry)

553 await registry.initialize()

554 await config_manager.initialize()

555

556 try:

557 config = await config_manager.load_configuration(config_file)

558 output_path = Path(output) if output else None

559 env_file = await config_manager.generate_environment_file(

560 config, output_path

561 )

562 console.print(f"[green]✓[/green] Environment file created: {env_file}")

563 except Exception as e:

564 console.print(f"[red]Error:[/red] {e}")

565 sys.exit(1)

566

567 asyncio.run(_generate_env())

568

569

570@config_cli.command()

571def list_templates() -> None:

572 """List available configuration templates."""

573

574 async def _list_templates() -> None:

575 registry = AdapterRegistry()

576 config_manager = ConfigurationManager(registry)

577 await config_manager.initialize()

578

579 templates_dir = config_manager.templates_dir

580 if not templates_dir.exists():

581 console.print("[yellow]No templates directory found[/yellow]")

582 return

583

584 templates = list(templates_dir.glob("*.yaml"))

585 if not templates:

586 console.print("[yellow]No templates available[/yellow]")

587 return

588

589 console.print("[bold]Available Templates:[/bold]")

590 for template in sorted(templates):

591 console.print(f" • {template.stem}")

592

593 asyncio.run(_list_templates())

594

595

596@config_cli.command()

597@click.argument("name")

598@click.argument("description")

599def backup(name: str, description: str) -> None:

600 """Create a backup of current configuration."""

601

602 async def _backup() -> None:

603 # This would need to be implemented based on current active configuration

604 console.print(

605 "[yellow]Backup functionality needs active configuration context[/yellow]"

606 )

607

608 asyncio.run(_backup())

609

610

611@config_cli.command()

612def list_backups() -> None:

613 """List all configuration backups."""

614

615 async def _list_backups() -> None:

616 registry = AdapterRegistry()

617 config_manager = ConfigurationManager(registry)

618 await config_manager.initialize()

619

620 backups = await config_manager.list_backups()

621

622 if not backups:

623 console.print("[yellow]No backups found[/yellow]")

624 return

625

626 ServerPanels.backups_table(backups)

627

628 asyncio.run(_list_backups())

629

630

631@config_cli.command()

632@click.argument("backup_id")

633@click.option("--output", "-o", help="Output file path")

634def restore(backup_id: str, output: str | None) -> None:

635 """Restore configuration from backup."""

636

637 async def _restore() -> None:

638 registry = AdapterRegistry()

639 config_manager = ConfigurationManager(registry)

640 await config_manager.initialize()

641

642 try:

643 config = await config_manager.restore_backup(backup_id)

644

645 if output:

646 config_file = await config_manager.save_configuration(config, output)

647 else:

648 config_file = await config_manager.save_configuration(config)

649

650 console.print(f"[green]✓[/green] Configuration restored to: {config_file}")

651 except Exception as e:

652 console.print(f"[red]Error:[/red] {e}")

653 sys.exit(1)

654

655 asyncio.run(_restore())

656

657

658if __name__ == "__main__":

659 config_cli()

Coverage for fastblocks / mcp / config_cli.py: 0%

321 statements