Megaraptor MCP

Instructions

Implementation Reference

• src/megaraptor_mcp/tools/deployment.py:1045-1169 (handler)

  The tool implementation for `deploy_agents_ssh` is in `src/megaraptor_mcp/tools/deployment.py`. It handles the logic for SSH-based agent deployment.

  async def deploy_agents_ssh(
      deployment_id: str,
      targets: list[str],
      username: str,
      key_path: Optional[str] = None,
      password: Optional[str] = None,
      target_os: str = "linux",
      labels: Optional[list[str]] = None,
      port: int = 22,
  ) -> list[TextContent]:
      """Push Velociraptor agents to Linux/macOS systems via SSH.
  
      Args:
          deployment_id: The deployment to connect agents to
          targets: List of target hostnames or IPs
          username: SSH username
          key_path: Path to SSH private key (preferred)
          password: SSH password (if not using key)
          target_os: Target OS - 'linux' or 'macos'
          labels: Labels to apply to deployed agents
          port: SSH port (default 22)
  
      Returns:
          Deployment results for each target.
      """
      try:
          from ..deployment.agents import SSHDeployer
          from ..deployment.agents.ssh_deployer import SSHCredentials, DeploymentTarget as SSHTarget
          from ..deployment.security import CertificateManager
          from ..deployment.deployers import DockerDeployer
  
          # Get deployment info
          deployer = DockerDeployer()
          info = await deployer.get_status(deployment_id)
  
          if not info:
              return [TextContent(
                  type="text",
                  text=json.dumps({
                      "error": f"Deployment not found: {deployment_id}",
                      "hint": "Use list_deployments tool to see available deployments"
                  }, indent=2)
              )]
  
          # Load certificates
          cert_manager = CertificateManager()
          bundle = cert_manager.load_bundle(deployment_id)
  
          if not bundle:
              return [TextContent(
                  type="text",
                  text=json.dumps({
                      "error": "Certificate bundle not found"
                  }, indent=2)
              )]
  
          # Generate client config
          import yaml
          client_config = yaml.dump({
              "Client": {
                  "server_urls": [info.server_url.replace("/api/", "") + ":8000/"],
                  "ca_certificate": bundle.ca_cert,
                  "nonce": secrets.token_hex(8),
                  "labels": labels or [],
              },
              "version": {"name": "megaraptor-ssh-deploy"},
          })
  
          # Create credentials and targets
          creds = SSHCredentials(
              username=username,
              key_path=key_path,
              password=password,
              port=port,
          )
  
          ssh_targets = [
              SSHTarget(hostname=t, credentials=creds, target_os=target_os)
              for t in targets
          ]
  
          # Deploy
          ssh_deployer = SSHDeployer(default_credentials=creds)
          results = await ssh_deployer.deploy_to_multiple(
              ssh_targets, client_config, labels=labels
          )
  
          return [TextContent(
              type="text",
              text=json.dumps({
                  "total": len(results),
                  "successful": sum(1 for r in results if r.success),
                  "failed": sum(1 for r in results if not r.success),
                  "results": [r.to_dict() for r in results],
              }, indent=2)
          )]
  
      except ImportError:
          return [TextContent(
              type="text",
              text=json.dumps({
                  "error": "paramiko not installed",
                  "suggestion": "pip install paramiko"
              }, indent=2)
          )]
  
      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)
          )]