Workflows MCP Server

Instructions

Implementation Reference

• src/server.py:186-227 (handler)

  Primary handler for the 'list_workflows' MCP tool, decorated with @mcp.tool() for FastMCP registration and execution.
  @mcp.tool() def list_workflows() -> dict: """ List all available workflow scripts. Returns: dict: List of workflows with their metadata """ try: workflows = [] workflows_path = Path(WORKFLOWS_DIR) for file in workflows_path.glob("*.py"): # Read the file to extract metadata content = file.read_text() # Extract description from docstring description = "" lines = content.split("\n") for line in lines: if line.startswith("Description:"): description = line.replace("Description:", "").strip() break workflows.append({ "name": file.stem, "path": str(file), "description": description, "modified": datetime.fromtimestamp(file.stat().st_mtime).isoformat() }) return { "status": "success", "count": len(workflows), "workflows": workflows } except Exception as e: return { "status": "error", "message": f"Failed to list workflows: {str(e)}" }
• src/mcp_http_server.py:127-134 (schema)

  Input schema definition for the 'list_workflows' tool in the MCP_TOOLS configuration list.
  "name": "list_workflows", "description": "List all available workflow scripts with their metadata.", "inputSchema": { "type": "object", "properties": {} } }, {
• src/mcp_http_server.py:266-289 (handler)

  Handler function for 'list_workflows' tool in the MCP HTTP server implementation.
  def tool_list_workflows(arguments: dict) -> dict: """List all workflows.""" try: workflows = [] workflows_path = Path(WORKFLOWS_DIR) for file in workflows_path.glob("*.py"): content = file.read_text() description = "" for line in content.split("\n"): if line.startswith("Description:"): description = line.replace("Description:", "").strip() break workflows.append({ "name": file.stem, "description": description, "modified": datetime.fromtimestamp(file.stat().st_mtime).isoformat() }) return {"workflows": workflows, "count": len(workflows)} except Exception as e: return {"error": str(e)}
• src/mcp_http_server.py:378-385 (registration)

  Registration of the 'list_workflows' handler in the TOOL_HANDLERS dictionary for the MCP HTTP server.
  TOOL_HANDLERS = { "create_workflow": tool_create_workflow, "execute_workflow": tool_execute_workflow, "list_workflows": tool_list_workflows, "read_workflow": tool_read_workflow, "update_workflow": tool_update_workflow, "delete_workflow": tool_delete_workflow, }
• src/http_server.py:175-199 (handler)

  HTTP endpoint handler for listing workflows in the simple HTTP server.
  async def list_workflows_handler(): """List all available workflow scripts.""" try: workflows = [] workflows_path = Path(WORKFLOWS_DIR) for file in workflows_path.glob("*.py"): content = file.read_text() description = "" for line in content.split("\n"): if line.startswith("Description:"): description = line.replace("Description:", "").strip() break workflows.append({ "name": file.stem, "path": str(file), "description": description, "modified": datetime.fromtimestamp(file.stat().st_mtime).isoformat() }) return {"status": "success", "count": len(workflows), "workflows": workflows} except Exception as e: return {"status": "error", "message": str(e)}