describe_workflow
Retrieve detailed information about a workflow definition in Kafka Schema Registry. The tool accepts a workflow_id as input to provide specific workflow details for better management and understanding.
Instructions
Get detailed information about a workflow definition
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| workflow_id | Yes |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
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, } )