list_workflows

Instructions

Implementation Reference

• mcp_server/server.py:244-277 (handler)

  Main handler for the list_workflows MCP tool. Lists built-in templates (from BUILTIN_TEMPLATES) and custom YAML workflows (via list_custom_workflows), plus active workflows from the store.

  @mcp.tool(annotations={"readOnlyHint": True, "destructiveHint": False, "idempotentHint": True, "openWorldHint": True})
  @vmware_tool(risk_level="low")
  def list_workflows() -> dict:
      """[READ] List all available workflow templates (built-in + custom).
  
      Built-in templates are always available. Custom templates are loaded
      from ~/.vmware/workflows/*.yaml — drop a YAML file there to add
      your own workflows.
  
      Returns:
          dict with builtin and custom workflow lists, each with name, description, steps count.
      """
      from vmware_pilot.custom_loader import list_custom_workflows
      from vmware_pilot.templates import BUILTIN_TEMPLATES
  
      builtin = [
          {"name": name, "type": "builtin", "description": (fn.__doc__ or "").split("\n")[0].strip()}
          for name, fn in BUILTIN_TEMPLATES.items()
      ]
      custom = [
          {**c, "type": "custom"}
          for c in list_custom_workflows()
      ]
  
      active = _get_store().list_active()
  
      try:
          return {
              "templates": builtin + custom,
              "active_workflows": active,
          }
      except Exception as e:
          return {"error": str(e), "hint": "Failed to list workflows."}

• mcp_server/server.py:244-246 (registration)

  Registration of list_workflows as an MCP tool via @mcp.tool decorator with read-only annotations and low risk level.

  @mcp.tool(annotations={"readOnlyHint": True, "destructiveHint": False, "idempotentHint": True, "openWorldHint": True})
  @vmware_tool(risk_level="low")
  def list_workflows() -> dict:

• vmware_pilot/custom_loader.py:160-185 (helper)

  Helper function that scans ~/.vmware/workflows/*.yaml and returns a list of custom workflow metadata dicts.

  def list_custom_workflows() -> list[dict[str, str]]:
      """List available custom workflow files (for discovery)."""
      if not _WORKFLOWS_DIR.exists():
          return []
  
      result = []
      try:
          import yaml
      except ImportError:
          return []
  
      for path in sorted(_WORKFLOWS_DIR.glob("*.yaml")):
          try:
              with open(path) as fh:
                  spec = yaml.safe_load(fh)
              if spec and "name" in spec:
                  result.append({
                      "name": spec["name"],
                      "description": spec.get("description", ""),
                      "file": path.name,
                      "steps": len(spec.get("steps", [])),
                  })
          except Exception:
              pass
  
      return result

• vmware_pilot/templates.py:1094-1110 (schema)

  Definition of BUILTIN_TEMPLATES dict containing all built-in workflow template names mapped to their factory functions.

  BUILTIN_TEMPLATES = {
      "clone_and_test": clone_and_test,
      "incident_response": incident_response,
      "investigate_alert": investigate_alert,
      "plan_and_approve": plan_and_approve,
      "compliance_scan": compliance_scan,
      "network_segment_setup": network_segment_setup,
      "vks_cluster_deploy": vks_cluster_deploy,
      "rolling_restart": rolling_restart,
      "capacity_expansion": capacity_expansion,
      "disaster_recovery": disaster_recovery,
      "patch_deployment": patch_deployment,
      "storage_expansion": storage_expansion,
      "baseline_capture": baseline_capture,
      "baseline_audit": baseline_audit,
      "baseline_remediate": baseline_remediate,
  }

• vmware_pilot/models.py:139-149 (helper)

  Helper method on WorkflowStore that queries the SQLite DB for active (non-completed, non-failed) workflows.

  def list_active(self) -> list[dict[str, Any]]:
      conn = self._connect()
      rows = conn.execute(
          "SELECT id, type, state, created_at, updated_at FROM workflows "
          "WHERE state NOT IN ('completed', 'failed') ORDER BY created_at DESC"
      ).fetchall()
      conn.close()
      return [
          {"id": r[0], "type": r[1], "state": r[2], "created_at": r[3], "updated_at": r[4]}
          for r in rows
      ]