MCP Kafka Schema Reg

Instructions

Implementation Reference

• workflow_mcp_integration.py:149-189 (handler)

  The main handler function for the 'describe_workflow' tool. It fetches the workflow definition using get_workflow_by_id, constructs a detailed step graph with field information, and returns a comprehensive JSON description of the workflow structure.

  @self.mcp.tool(description="Get detailed information about a workflow definition")
  async def describe_workflow(workflow_id: str) -> str:
      """Get detailed workflow information."""
      workflow = get_workflow_by_id(workflow_id)
  
      if not workflow:
          return json.dumps({"error": f"Workflow '{workflow_id}' not found"})
  
      # Build step graph
      step_graph = {}
      for step_id, step in workflow.steps.items():
          step_info = {
              "id": step.id,
              "title": step.title,
              "description": step.description,
              "type": step.elicitation_type.value,
              "fields": [
                  {
                      "name": field.name,
                      "type": field.type,
                      "description": field.description,
                      "required": field.required,
                      "options": field.options if field.type == "choice" else None,
                  }
                  for field in step.fields
              ],
              "next_steps": step.next_steps,
          }
          step_graph[step_id] = step_info
  
      return json.dumps(
          {
              "workflow_id": workflow.id,
              "name": workflow.name,
              "description": workflow.description,
              "initial_step_id": workflow.initial_step_id,
              "total_steps": len(workflow.steps),
              "steps": step_graph,
              "metadata": workflow.metadata,
          }
      )