MCP Kafka Schema Reg

Instructions

Implementation Reference

• workflow_mcp_integration.py:308-338 (handler)

  The primary MCP tool handler for 'guided_disaster_recovery'. This async function is decorated with @self.mcp.tool and initiates the multi-step 'disaster_recovery_setup' workflow, returning a JSON response with workflow status and first step details.

  @self.mcp.tool(description="Start the Disaster Recovery Setup workflow for configuring DR strategies")
  async def guided_disaster_recovery() -> str:
      """Convenience method to start Disaster Recovery workflow."""
      workflow_id = "disaster_recovery_setup"
  
      try:
          request = await self.multi_step_manager.start_workflow(workflow_id=workflow_id, initial_context={})
  
          if request:
              return json.dumps(
                  {
                      "status": "started",
                      "workflow_id": workflow_id,
                      "workflow_name": "Disaster Recovery Setup",
                      "request_id": request.id,
                      "first_step": request.title,
                      "description": request.description,
                      "message": (
                          "Disaster Recovery Setup workflow started. This workflow will guide you through:\n"
                          "1. Current infrastructure assessment\n"
                          "2. Backup strategy configuration\n"
                          "3. Recovery procedures planning\n"
                          "4. Testing and validation setup"
                      ),
                  }
              )
          else:
              return json.dumps({"error": "Failed to start Disaster Recovery workflow"})
      except Exception as e:
          logger.error(f"Error starting Disaster Recovery workflow: {str(e)}")
          return json.dumps({"error": f"Failed to start workflow: {str(e)}"})

• workflow_mcp_integration.py:443-486 (helper)

  Supporting helper function that takes the completed workflow responses and constructs a detailed disaster recovery execution plan/strategy based on user inputs during the guided workflow.

  def execute_disaster_recovery_setup(responses: Dict[str, Any]) -> Dict[str, Any]:
      """Execute disaster recovery setup based on workflow responses."""
      dr_strategy = responses.get("dr_strategy")
  
      result = {"operation": "disaster_recovery_setup", "strategy": dr_strategy, "status": "pending"}
  
      # Add strategy-specific configuration
      if dr_strategy == "active_passive":
          result["config"] = {
              "primary_registry": responses.get("primary_registry"),
              "standby_registry": responses.get("standby_registry"),
              "replication_interval": responses.get("replication_interval"),
              "failover_mode": responses.get("failover_mode"),
          }
      elif dr_strategy == "active_active":
          result["config"] = {
              "active_registries": [reg.strip() for reg in responses.get("active_registries", "").split(",")],
              "conflict_resolution": responses.get("conflict_resolution"),
              "sync_topology": responses.get("sync_topology"),
          }
      elif dr_strategy == "backup_restore":
          result["config"] = {
              "backup_schedule": responses.get("backup_schedule"),
              "backup_location": responses.get("backup_location"),
              "retention_policy": responses.get("retention_policy"),
              "encryption": responses.get("encryption") == "true",
          }
      elif dr_strategy == "multi_region":
          result["config"] = {
              "regions": [region.strip() for region in responses.get("regions", "").split(",")],
              "primary_region": responses.get("primary_region"),
              "data_residency": responses.get("data_residency") == "true",
              "cross_region_replication": responses.get("cross_region_replication"),
          }
  
      # Add common DR options
      result["options"] = {
          "enable_monitoring": responses.get("enable_monitoring") == "true",
          "run_dr_drill": responses.get("run_dr_drill") == "true",
          "generate_runbook": responses.get("generate_runbook") == "true",
          "initial_sync": responses.get("initial_sync") == "true",
      }
  
      return result