design_workflow

Instructions

Implementation Reference

• mcp_server/server.py:513-571 (handler)

  The 'design_workflow' tool handler function. Registered as an MCP tool and annotated with vmware_tool(risk_level='low'). Takes a 'goal' (str) and optional 'constraints' (str), creates a Workflow with state=DRAFT and empty steps, saves it, and returns a dict with workflow_id, state, goal, constraints, available_skills, and next_step instructions telling the AI to call update_draft() to fill in steps.

  @mcp.tool(annotations={"readOnlyHint": False, "destructiveHint": False, "idempotentHint": False, "openWorldHint": True})
  @vmware_tool(risk_level="low")
  def design_workflow(
      goal: str,
      constraints: str = "",
  ) -> dict:
      """[WRITE] Start designing a workflow from a natural language description.
  
      Call this when the user describes a complex operation and you need to
      design a multi-step workflow. Returns a DRAFT workflow with proposed steps
      for the user to review and edit before execution.
  
      Design flow:
        1. AI calls design_workflow(goal="...") → returns draft with proposed steps
        2. User reviews: "step 3 should use vm_power_off instead" or "add an approval before step 4"
        3. AI calls update_draft(workflow_id, ...) to modify
        4. User confirms: "looks good"
        5. AI calls confirm_draft(workflow_id) → state changes to PENDING
        6. AI calls run_workflow(workflow_id) → execute
  
      The AI should use get_skill_catalog() first to understand available tools,
      then propose steps based on the user's goal.
  
      Args:
          goal: Natural language description of what the user wants to accomplish.
          constraints: Optional constraints (e.g. "must have approval before any destructive step",
                       "use NSX for networking", "target is vcenter-prod").
  
      Returns:
          dict with workflow_id (state=DRAFT), proposed steps placeholder,
          and instructions for the AI to fill in steps via update_draft.
      """
      from datetime import datetime, timezone
      from vmware_pilot.models import Workflow, WorkflowState, new_workflow_id
  
      now = datetime.now(tz=timezone.utc).isoformat()
      wf = Workflow(
          id=new_workflow_id(),
          workflow_type="custom_draft",
          state=WorkflowState.DRAFT,
          steps=[],
          params={"goal": goal, "constraints": constraints, "custom": True},
          created_at=now,
          updated_at=now,
      )
      wf.log("draft_created", f"Goal: {goal}")
      _get_store().save(wf)
  
      return {
          "workflow_id": wf.id,
          "state": "draft",
          "goal": goal,
          "constraints": constraints,
          "available_skills": list(SKILL_CATALOG.keys()),
          "next_step": (
              "Now design the workflow steps. Use get_skill_catalog() to see available tools, "
              "then call update_draft() to add steps. When done, call confirm_draft() to finalize."
          ),
      }

• mcp_server/server.py:535-548 (schema)

  Docstring defining the schema for design_workflow: 'goal' (str) is the natural language description of what the user wants, 'constraints' (str, optional) are constraints like approvals or target environment. Returns a dict with workflow_id, state=DRAFT, proposed steps placeholder, and instructions.

  Args:
      goal: Natural language description of what the user wants to accomplish.
      constraints: Optional constraints (e.g. "must have approval before any destructive step",
                   "use NSX for networking", "target is vcenter-prod").
  
  Returns:
      dict with workflow_id (state=DRAFT), proposed steps placeholder,
      and instructions for the AI to fill in steps via update_draft.
  """
  from datetime import datetime, timezone
  from vmware_pilot.models import Workflow, WorkflowState, new_workflow_id
  
  now = datetime.now(tz=timezone.utc).isoformat()

• mcp_server/server.py:513-514 (registration)

  Registration of design_workflow as an MCP tool via @mcp.tool(annotations={...}) and @vmware_tool(risk_level='low') decorators. The @mcp.tool decorator registers it with the FastMCP server, making it available to AI agents.

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

• mcp_server/server.py:545-571 (helper)

  Internal helper logic: imports datetime, Workflow, WorkflowState, new_workflow_id from vmware_pilot.models; creates a new Workflow with id, type='custom_draft', state=DRAFT, empty steps, params containing goal/constraints; logs 'draft_created' event; saves via _get_store().save().

  from datetime import datetime, timezone
  from vmware_pilot.models import Workflow, WorkflowState, new_workflow_id
  
  now = datetime.now(tz=timezone.utc).isoformat()
  wf = Workflow(
      id=new_workflow_id(),
      workflow_type="custom_draft",
      state=WorkflowState.DRAFT,
      steps=[],
      params={"goal": goal, "constraints": constraints, "custom": True},
      created_at=now,
      updated_at=now,
  )
  wf.log("draft_created", f"Goal: {goal}")
  _get_store().save(wf)
  
  return {
      "workflow_id": wf.id,
      "state": "draft",
      "goal": goal,
      "constraints": constraints,
      "available_skills": list(SKILL_CATALOG.keys()),
      "next_step": (
          "Now design the workflow steps. Use get_skill_catalog() to see available tools, "
          "then call update_draft() to add steps. When done, call confirm_draft() to finalize."
      ),
  }

• mcp_server/server.py:576-639 (helper)

  The update_draft tool (lines 576-639) and confirm_draft tool (lines 644-691) are companion helpers that complete the design workflow flow: update_draft fills in/modifies steps, confirm_draft transitions from DRAFT to PENDING so the workflow can be run.

  def update_draft(
      workflow_id: str,
      name: str = "",
      description: str = "",
      steps: list[dict[str, Any]] | None = None,
  ) -> dict:
      """[WRITE] Update a DRAFT workflow's name, description, or steps.
  
      Call this after design_workflow() to fill in the actual steps,
      or to modify steps based on user feedback.
  
      Each step dict: {action, skill, tool, params, rollback_tool?, rollback_params?}
      Use action="require_approval" for approval gates.
  
      Args:
          workflow_id: The draft workflow ID.
          name: Workflow name (optional, updates workflow_type).
          description: Human-readable description.
          steps: Complete list of steps (replaces all existing steps).
  
      Returns:
          Updated workflow summary for user review.
      """
      from vmware_pilot.models import WorkflowStep
  
      wf = _get_store().load(workflow_id)
      if not wf:
          return {"error": f"Workflow '{workflow_id}' not found"}
      if wf.state != WorkflowState.DRAFT:
          return {"error": f"Workflow '{workflow_id}' is not a draft (state: {wf.state.value})"}
  
      if name:
          wf.workflow_type = name
      if description:
          wf.params["description"] = description
  
      if steps is not None:
          wf.steps = [
              WorkflowStep(
                  index=i,
                  action=s.get("action", f"step_{i}"),
                  skill=s.get("skill", "unknown"),
                  tool=s.get("tool", "unknown"),
                  params=s.get("params", {}),
                  rollback_tool=s.get("rollback_tool", ""),
                  rollback_params=s.get("rollback_params", {}),
              )
              for i, s in enumerate(steps)
          ]
          wf.log("steps_updated", f"{len(steps)} steps")
  
      _get_store().save(wf)
  
      return {
          "workflow_id": wf.id,
          "workflow_type": wf.workflow_type,
          "state": "draft",
          "steps": [
              {"index": s.index, "action": s.action, "skill": s.skill, "tool": s.tool,
               "has_rollback": bool(s.rollback_tool)}
              for s in wf.steps
          ],
          "message": "Draft updated. Show to user for review. Call confirm_draft() when approved.",
      }