create_flowchart
Idempotent
Generate process flowcharts using 24 shapes like StartEnd, Process, and Decision to visualize workflows and decision trees.
Instructions
Create process flowcharts with 24 shapes (StartEnd, Process, Decision, etc.).
Example: steps=[{"id":"start","shape":"StartEnd","label":"Start"},...] flows=[{"from_step":"start","to_step":"check"},...]
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| name | Yes | Flowchart title | |
| steps | Yes | Flowchart steps | |
| flows | Yes | Connections between steps | |
| direction | No | Layout direction | TB |
| output_format | No | Output format(s): png, pdf, jpg, dot | png |
| output_dir | No | Output directory (default: current directory). Auto-created if missing. | |
| graph_attr | No | Graphviz graph attributes (overrides defaults) | |
| return_base64 | No | Return base64 images |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- src/diagrams_mcp/tools/core.py:627-639 (registration)Registers the create_flowchart tool with MCP using the @mcp.tool decorator, including name, description, and annotations.
@mcp.tool( name="create_flowchart", description="""Create process flowcharts with 24 shapes (StartEnd, Process, Decision, etc.). Example: steps=[{"id":"start","shape":"StartEnd","label":"Start"},...] flows=[{"from_step":"start","to_step":"check"},...]""", annotations={ "readOnlyHint": False, "destructiveHint": False, "idempotentHint": True, }, ) - src/diagrams_mcp/tools/core.py:640-835 (handler)The main handler function that validates inputs, maps shapes to diagrams.programming.flowchart classes, builds the diagram with connections, generates output files in specified formats, and returns formatted results.
async def create_flowchart( name: Annotated[str, Field(description="Flowchart title")], steps: Annotated[List[FlowStepDef], Field(description="Flowchart steps")], flows: Annotated[List[FlowConnectionDef], Field(description="Connections between steps")], direction: Annotated[ Literal["LR", "RL", "TB", "BT"], Field(description="Layout direction") ] = "TB", output_format: Annotated[ str | List[str], Field(description="Output format(s): png, pdf, jpg, dot") ] = "png", output_dir: Annotated[ Optional[str], Field( description="Output directory (default: current directory). Auto-created if missing." ), ] = None, graph_attr: Annotated[ Optional[Dict[str, Any]], Field(description="Graphviz graph attributes (overrides defaults)"), ] = None, return_base64: Annotated[bool, Field(description="Return base64 images")] = False, ) -> str: """Create flowchart diagram.""" start_time = time.time() try: from diagrams.programming.flowchart import ( Action, Collate, Database, Decision, Delay, Display, Document, InputOutput, Inspection, InternalStorage, LoopLimit, ManualInput, ManualLoop, Merge, MultipleDocuments, OffPageConnectorLeft, OffPageConnectorRight, Or, PredefinedProcess, Preparation, Sort, StartEnd, StoredData, SummingJunction, ) # Map shape names to classes (including user-friendly aliases) shape_map = { # User-friendly aliases "Process": PredefinedProcess, "Data": InputOutput, # Standard flowchart shapes "StartEnd": StartEnd, "Decision": Decision, "Document": Document, "Database": Database, "Delay": Delay, # All other available shapes "Action": Action, "Collate": Collate, "Display": Display, "Inspection": Inspection, "InternalStorage": InternalStorage, "InputOutput": InputOutput, "LoopLimit": LoopLimit, "ManualInput": ManualInput, "ManualLoop": ManualLoop, "Merge": Merge, "MultipleDocuments": MultipleDocuments, "OffPageConnectorLeft": OffPageConnectorLeft, "OffPageConnectorRight": OffPageConnectorRight, "Or": Or, "PredefinedProcess": PredefinedProcess, "Preparation": Preparation, "Sort": Sort, "StoredData": StoredData, "SummingJunction": SummingJunction, } # Validate steps step_ids = {step.id for step in steps} for step in steps: if step.shape not in shape_map: available = ", ".join(sorted(shape_map.keys())) raise ValueError(f"Unknown flowchart shape '{step.shape}'. Available: {available}") # Validate flows for flow in flows: if flow.from_step not in step_ids: raise ValueError(f"Flow references unknown step '{flow.from_step}'") targets = [flow.to_step] if isinstance(flow.to_step, str) else flow.to_step for target in targets: if target not in step_ids: raise ValueError(f"Flow references unknown step '{target}'") # Generate flowchart formats = [output_format] if isinstance(output_format, str) else output_format # Reject SVG - it's buggy and unsupported if any("svg" in fmt.lower() for fmt in formats): raise ValueError("SVG output is not supported. Use png, pdf, jpg, or dot instead.") original_dir = os.getcwd() if output_dir: os.makedirs(output_dir, exist_ok=True) os.chdir(output_dir) try: # Hide diagram title and set better flowchart layout attributes default_graph_attr = { "label": "", "splines": "ortho", # Orthogonal edges with 90-degree angles "nodesep": "0.8", # Horizontal spacing between nodes (inches) "ranksep": "0.75", # Vertical spacing between ranks (inches) } merged_graph_attr = {**default_graph_attr, **(graph_attr or {})} with Diagram( name=name, show=False, direction=direction, outformat=formats, graph_attr=merged_graph_attr, ) as _: # Create step objects step_objects = {} for step in steps: ShapeClass = shape_map[step.shape] step_obj = ShapeClass(step.label) step_objects[step.id] = step_obj # Create flows edge_count = 0 for flow in flows: from_obj = step_objects[flow.from_step] targets = [flow.to_step] if isinstance(flow.to_step, str) else flow.to_step for target in targets: to_obj = step_objects[target] if flow.label: edge = Edge(label=flow.label) _ = from_obj >> edge >> to_obj else: _ = from_obj >> to_obj edge_count += 1 # Get file paths diagram_filename = name.replace(" ", "_").replace("-", "_").lower() file_paths = [] for fmt in formats: file_path = f"{diagram_filename}.{fmt}" if output_dir: file_path = os.path.join(output_dir, file_path) file_paths.append(os.path.abspath(file_path)) # Build metadata generation_time_ms = (time.time() - start_time) * 1000 metadata = build_diagram_metadata( file_paths, node_count=len(steps), edge_count=edge_count, cluster_count=0, generation_time_ms=generation_time_ms, ) # Base64 if requested base64_images = None if return_base64: base64_images = {} for path in file_paths: ext = Path(path).suffix[1:] if ext != "dot": try: base64_images[ext] = encode_file_base64(path) except Exception: pass return format_diagram_result(file_paths, metadata, base64_images) finally: if output_dir: os.chdir(original_dir) except Exception as e: return format_error(f"Failed to create flowchart: {str(e)}") - Pydantic model defining the structure and validation for flowchart steps (id, shape, label).
class FlowStepDef(BaseModel): """Definition of a flowchart step.""" id: str = Field( description="Unique step ID", min_length=1, max_length=200, ) shape: str = Field( description="Shape (StartEnd, Process, Decision, Data...)", min_length=1, ) label: str = Field( description="Display label", min_length=1, ) @field_validator("id") @classmethod def validate_id_format(cls, v: str) -> str: """Validate ID format.""" if not re.match(r"^[a-zA-Z0-9_-]+$", v): raise ValueError( f"Invalid step ID '{v}': only alphanumeric characters, " "underscores, and hyphens allowed" ) return v - Pydantic model defining the structure and validation for flowchart connections (from_step, to_step, label, condition).
class FlowConnectionDef(BaseModel): """Definition of a flowchart connection.""" from_step: str = Field( description="Source step ID", min_length=1, ) to_step: str | List[str] = Field( description="Target step ID(s)", ) label: Optional[str] = Field( default=None, description="Label (e.g. 'Yes', 'No')", ) condition: Optional[str] = Field( default=None, description="Condition description", )