Megaraptor MCP

Instructions

Implementation Reference

• src/megaraptor_mcp/tools/deployment.py:1569-1698 (handler)

  Implementation of the `validate_deployment` tool, which runs security and health checks on a specific deployment.

  async def validate_deployment(
      deployment_id: str,
  ) -> list[TextContent]:
      """Run comprehensive security and health validation on a deployment.
  
      Checks:
      - Server accessibility
      - Certificate validity
      - Service health
      - Security configuration
  
      Args:
          deployment_id: The deployment to validate
  
      Returns:
          Detailed validation report with any issues found.
      """
      try:
          from ..deployment.security import CertificateManager
          from ..deployment.deployers import DockerDeployer
  
          checks = []
  
          # Check deployment exists
          deployer = DockerDeployer()
          info = await deployer.get_status(deployment_id)
  
          if not info:
              return [TextContent(
                  type="text",
                  text=json.dumps({
                      "valid": False,
                      "error": f"Deployment not found: {deployment_id}"
                  }, indent=2)
              )]
  
          checks.append({
              "check": "deployment_exists",
              "status": "pass",
              "message": f"Deployment found: {info.target}",
          })
  
          # Check certificates
          cert_manager = CertificateManager()
          bundle = cert_manager.load_bundle(deployment_id)
  
          if bundle:
              checks.append({
                  "check": "certificates_exist",
                  "status": "pass",
                  "message": f"CA fingerprint: {bundle.ca_fingerprint[:16]}...",
              })
          else:
              checks.append({
                  "check": "certificates_exist",
                  "status": "fail",
                  "message": "Certificate bundle not found",
              })
  
          # Check health
          health = await deployer.health_check(deployment_id)
          health_status = "pass" if health.get("healthy") else "fail"
          checks.append({
              "check": "health_check",
              "status": health_status,
              "message": health.get("checks", []),
          })
  
          # Check auto-destroy schedule
          if info.auto_destroy_at:
              destroy_time = datetime.fromisoformat(info.auto_destroy_at.replace("Z", "+00:00"))
              hours_remaining = (destroy_time - datetime.now(timezone.utc)).total_seconds() / 3600
  
              if hours_remaining < 0:
                  checks.append({
                      "check": "auto_destroy",
                      "status": "warn",
                      "message": f"Deployment scheduled for destruction has passed",
                  })
              elif hours_remaining < 24:
                  checks.append({
                      "check": "auto_destroy",
                      "status": "warn",
                      "message": f"Deployment will be destroyed in {hours_remaining:.1f} hours",
                  })
              else:
                  checks.append({
                      "check": "auto_destroy",
                      "status": "info",
                      "message": f"Deployment will be destroyed in {hours_remaining:.1f} hours",
                  })
  
          # Overall status
          failed_checks = [c for c in checks if c["status"] == "fail"]
          warn_checks = [c for c in checks if c["status"] == "warn"]
  
          return [TextContent(
              type="text",
              text=json.dumps({
                  "valid": len(failed_checks) == 0,
                  "deployment_id": deployment_id,
                  "state": info.state.value,
                  "summary": {
                      "passed": len([c for c in checks if c["status"] == "pass"]),
                      "warnings": len(warn_checks),
                      "failed": len(failed_checks),
                  },
                  "checks": checks,
              }, indent=2, default=str)
          )]
  
      except ImportError as e:
          return [TextContent(
              type="text",
              text=json.dumps({
                  "error": f"Missing dependency: {str(e)}",
                  "hint": "Install required packages with: pip install megaraptor-mcp[deployment]"
              }, indent=2)
          )]
  
      except Exception:
          # Generic errors - don't expose internals
          return [TextContent(
              type="text",
              text=json.dumps({
                  "error": "Operation failed",
                  "hint": "Check deployment configuration and try again"
              }, indent=2)
          )]