Diagrams MCP Server

Instructions

Implementation Reference

• src/diagrams_mcp/tools/core.py:627-835 (handler)

  Full handler implementation for create_flowchart tool: registers the tool, defines input schema via annotations, validates steps and flows, imports flowchart shapes from diagrams.programming.flowchart, maps string shapes to classes, constructs Diagram with orthogonal edges, generates files in PNG/PDF/JPG/DOT, returns formatted result with metadata and optional base64 embeds.
  @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, }, ) 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)}")
• src/diagrams_mcp/core/validators.py:161-188 (schema)

  Pydantic model defining input schema for flowchart steps: id (alphanumeric+_-), shape (string validated at runtime), 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
• src/diagrams_mcp/core/validators.py:190-208 (schema)

  Pydantic model defining input schema for flowchart connections: from_step, to_step (single or list), optional label and 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", )
• src/diagrams_mcp/tools/core.py:627-639 (registration)

  MCP tool registration decorator defining the tool name, description, input parameters with types, 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:694-724 (helper)

  Mapping dictionary from user-provided shape strings to actual diagrams.programming.flowchart class imports, enabling string-based shape specification.
  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, }