Refactor

mcp_logic.py

@@ -3,6 +3,7 @@ import uuid
 import json
 from typing import Any, Dict, Union, List
 from starlette.responses import Response
+from tools.utils import ToolError
 
 class MCPServer:
     def __init__(self):
@@ -40,7 +41,14 @@ class MCPServer:
                     media_type="application/json"
                 )
             
-            result_data = await tool.handler(args)
+            try:
+                result_data = await tool.handler(args)
+            except ToolError as e:
+                # Convert tool errors into a text response so the LLM can understand and potentially fix the input
+                result_data = {"text": str(e)}
+            except Exception as e:
+                # Catch-all for unexpected crashes to prevent server death
+                result_data = {"text": f"Unexpected internal error: {str(e)}"}
             
             # Handle results that are already lists of content (for multimodal/multi-part responses)
             if isinstance(result_data, list):

tools/contact_sheet.py

@@ -1,14 +1,11 @@
-import os
 import base64
-import mimetypes
 import logging
-import asyncio
 import io
 import datetime
-import time
-from typing import Any, Callable, Dict, List, Union
+from pathlib import Path
+from typing import Any, Dict, List
 from PIL import Image, ImageDraw, ImageFont
-from tools.utils import format_relative_time
+from tools.utils import format_relative_time, ToolError, COMMON_FONTS, get_file_info_list, sort_file_list, get_paginated_list
 
 logger = logging.getLogger("MattCP")
 
@@ -18,44 +15,32 @@ async def handle(args: Dict[str, Any]):
     Grid: 10 columns x 7 rows (70 images total).
     Sized for optimal token usage (1120 tokens) on llama.cpp.
     """
-    dir_path = args.get("path")
+    dir_path_str = args.get("path")
     page = int(args.get("page", 1))
     sort_by = args.get("sort_by", "mtime")
     
-    if not dir_path:
-        return {"text": "Error: Missing path argument"}
-    if not os.path.isdir(dir_path):
-        return {"text": f"Error: {dir_path} is not a directory."}
+    if not dir_path_str:
+        raise ToolError("Missing path argument")
+    
+    dir_path = Path(dir_path_str)
+    if not dir_path.is_dir():
+        raise ToolError(f"{dir_path_str} is not a directory.")
     
     exts = ('.png', '.jpg', '.jpeg', '.webp', '.bmp')
-    file_info_list = []
-    for f in os.listdir(dir_path):
-        if f.lower().endswith(exts):
-            full_path = os.path.join(dir_path, f)
-            try:
-                stats = os.stat(full_path)
-                file_info_list.append({
-                    "name": f,
-                    "path": full_path,
-                    "mtime": stats.st_mtime,
-                    "size": stats.st_size
-                })
-            except Exception as e:
-                logger.error(f"Could not stat {f}: {e}")
+    file_info_list = get_file_info_list(dir_path, extensions=exts)
 
     if not file_info_list:
         return {"text": "No supported images found in the directory."}
 
     # Sorting
-    if sort_by == "mtime":
-        file_info_list.sort(key=lambda x: x["mtime"], reverse=True)
-        sort_label = "Date (newest first)"
-    elif sort_by == "size":
-        file_info_list.sort(key=lambda x: x["size"], reverse=True)
-        sort_label = "Size (largest first)"
-    else:
-        file_info_list.sort(key=lambda x: x["name"].lower())
-        sort_label = "Name (alphabetical)"
+    file_info_list = sort_file_list(file_info_list, sort_by)
+    
+    sort_labels = {
+        "mtime": "Date (newest first)",
+        "size": "Size (largest first)",
+        "name": "Name (alphabetical)"
+    }
+    sort_label = sort_labels.get(sort_by, "Name (alphabetical)")
 
     # Fixed grid dimensions for token optimization
     COLS = 10
@@ -63,15 +48,12 @@ async def handle(args: Dict[str, Any]):
     PAGE_SIZE = COLS * ROWS
     
     total_files = len(file_info_list)
-    start_idx = (page - 1) * PAGE_SIZE
-    end_idx = start_idx + PAGE_SIZE
-    paged_files = file_info_list[start_idx:end_idx]
+    paged_files = get_paginated_list(file_info_list, page, PAGE_SIZE)
     
     if not paged_files:
         return {"text": f"No images found on page {page}."}
 
     thumb_size = 192 # 192 / 48 = 4 patches per side
-    
     canvas_w = COLS * thumb_size
     canvas_h = ROWS * thumb_size
     
@@ -80,21 +62,16 @@ async def handle(args: Dict[str, Any]):
     
     # Font loading
     font = None
-    font_paths = [
-        "/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf",
-        "/usr/share/fonts/truetype/liberation/LiberationSans-Bold.ttf",
-        "/usr/share/fonts/truetype/freefont/FreeSansBold.ttf",
-    ]
-    for path in font_paths:
-        if os.path.exists(path):
-            try:
-                font = ImageFont.truetype(path, 24)
-                break
-            except Exception:
-                continue
+    for font_candidate in COMMON_FONTS:
+        try:
+            font = ImageFont.truetype(font_candidate, 24)
+            break
+        except Exception:
+            continue
     if font is None:
         font = ImageFont.load_default()
 
+    start_idx = (page - 1) * PAGE_SIZE
     for i, info in enumerate(paged_files):
         full_path = info["path"]
         row = i // COLS
@@ -102,7 +79,6 @@ async def handle(args: Dict[str, Any]):
         
         x = col * thumb_size
         y = row * thumb_size
-        
         global_index = start_idx + i + 1
         
         try:
@@ -112,17 +88,13 @@ async def handle(args: Dict[str, Any]):
                 off_y = (thumb_size - img.height) // 2
                 canvas.paste(img, (x + off_x, y + off_y))
             
-            # Index number in lower-left corner
             text = str(global_index)
             if font:
                 bbox = draw.textbbox((0, 0), text, font=font)
-                text_w = bbox[2] - bbox[0]
-                text_h = bbox[3] - bbox[1]
+                text_w, text_h = bbox[2] - bbox[0], bbox[3] - bbox[1]
             else:
-                text_w = len(text) * 8
-                text_h = 15
+                text_w, text_h = len(text) * 8, 15
             
-            # Semi-opaque background rectangle for the number
             draw.rectangle([x, y + thumb_size - text_h - 5, x + text_w + 5, y + thumb_size - 2], fill=(0, 0, 0, 180))
             draw.text((x + 2, y + thumb_size - text_h - 7), text, fill=(255, 255, 255), font=font)
             
@@ -136,7 +108,6 @@ async def handle(args: Dict[str, Any]):
     
     total_pages = (total_files + PAGE_SIZE - 1) // PAGE_SIZE
     
-    # Range shown: first and last of the CURRENT PAGE
     first_on_page = paged_files[0]
     last_on_page = paged_files[-1]
     
@@ -151,7 +122,7 @@ async def handle(args: Dict[str, Any]):
         last_val = last_on_page['name']
         
     header_text = (
-        f"Directory: {dir_path}\n"
+        f"Directory: {dir_path_str}\n"
         f"Page {page} of {total_pages} ({total_files} images total). Sorted by: {sort_label}\n"
         f"Range shown: {first_val} ... {last_val}"
     )

tools/list_directory.py

@@ -1,8 +1,7 @@
-import os
 import logging
-import datetime
+from pathlib import Path
 from typing import Any, Dict
-from tools.utils import format_relative_time
+from tools.utils import format_relative_time, ToolError, get_file_info_list, sort_file_list
 
 logger = logging.getLogger("MattCP")
 
@@ -12,65 +11,24 @@ async def handle(args: Dict[str, Any]):
     Directories are listed first, then files.
     Supports pagination and sorting.
     """
-    path = args.get("path")
+    path_str = args.get("path")
     page = int(args.get("page", 1))
     page_size = int(args.get("page_size", 64))
     sort_by = args.get("sort_by", "mtime")
     
-    if not path:
-        return {"text": "Error: Missing path argument"}
-    if not os.path.isdir(path):
-        return {"text": f"Error: {path} is not a directory."}
+    if not path_str:
+        raise ToolError("Missing path argument")
+    
+    path = Path(path_str)
+    if not path.is_dir():
+        raise ToolError(f"{path_str} is not a directory.")
 
-    try:
-        entries = os.listdir(path)
-    except Exception as e:
-        return {"text": f"Error reading directory: {e}"}
-
-    all_items = []
-    for entry in entries:
-        full_path = os.path.join(path, entry)
-        try:
-            stats = os.stat(full_path)
-            mtime_rel = format_relative_time(stats.st_mtime)
-            
-            if os.path.isdir(full_path):
-                try:
-                    count = len(os.listdir(full_path))
-                except:
-                    count = 0
-                all_items.append({
-                    "type": "dir",
-                    "name": entry,
-                    "info": f"{count} items",
-                    "mtime": mtime_rel,
-                    "mtime_raw": stats.st_mtime,
-                    "size": stats.st_size
-                })
-            else:
-                size_kb = stats.st_size / 1024
-                all_items.append({
-                    "type": "file",
-                    "name": entry,
-                    "info": f"{size_kb:.1f}KB",
-                    "mtime": mtime_rel,
-                    "mtime_raw": stats.st_mtime,
-                    "size": stats.st_size
-                })
-        except Exception as e:
-            logger.error(f"Could not stat {entry}: {e}")
-
-    # Sort Logic: Directories always come first, then apply secondary sort
-    if sort_by == "mtime":
-        all_items.sort(key=lambda x: (x["type"] == "file", -x["mtime_raw"]))
-    elif sort_by == "size":
-        all_items.sort(key=lambda x: (x["type"] == "file", -x["size"]))
-    else:
-        all_items.sort(key=lambda x: (x["type"] == "file", x["name"].lower()))
+    all_items = get_file_info_list(path)
+    all_items = sort_file_list(all_items, sort_by)
 
     total_items = len(all_items)
     
-    # Custom Pagination: show all if <= 100, otherwise paginate by 64
+    # Custom Pagination: show all if <= 100, otherwise paginate by page_size
     if total_items <= 100:
         start_idx = 0
         end_idx = total_items
@@ -89,11 +47,18 @@ async def handle(args: Dict[str, Any]):
 
     output = []
     for item in paged_items:
-        if item["type"] == "dir":
-            output.append(f"[DIR] {item['name']} | {item['info']} | {item['mtime']}")
+        mtime_rel = format_relative_time(item["mtime"])
+        if item["is_dir"]:
+            try:
+                count = len(list(item["path"].iterdir()))
+                info = f"{count} items"
+            except:
+                info = "0 items"
+            output.append(f"[DIR] {item['name']} | {info} | {mtime_rel}")
         else:
-            output.append(f"      {item['name']} | {item['info']} | {item['mtime']}")
+            size_kb = item["size"] / 1024
+            output.append(f"      {item['name']} | {size_kb:.1f}KB | {mtime_rel}")
 
-    header = f"Listing of {path}\nPage {effective_page} of {total_pages} ({total_items} total items). Showing {start_idx+1}-{min(end_idx, total_items)}."
+    header = f"Listing of {path_str}\nPage {effective_page} of {total_pages} ({total_items} total items). Showing {start_idx+1}-{min(end_idx, total_items)}."
     
     return {"text": f"{header}\n\n" + "\n".join(output)}

tools/read_image.py

@@ -1,8 +1,9 @@
-import os
 import base64
 import mimetypes
 import logging
+from pathlib import Path
 from typing import Any, Dict
+from tools.utils import ToolError
 
 logger = logging.getLogger("MattCP")
 
@@ -11,19 +12,19 @@ async def handle(args: Dict[str, Any]):
     Reads an image file from disk and returns it as a base64 encoded string 
     within an MCP ImageContent object.
     """
-    path = args.get("path")
-    if not path:
-        return {"text": "Error: Missing path argument"}
+    path_str = args.get("path")
+    if not path_str:
+        raise ToolError("Missing path argument")
+    
+    path = Path(path_str)
     
     # Validate that the file has an image-like MIME type
-    mime_type, _ = mimetypes.guess_type(path)
+    mime_type, _ = mimetypes.guess_type(path_str)
     if not mime_type or not mime_type.startswith("image/"):
-        return {"text": f"Error: File {path} is not a supported image format."}
+        raise ToolError(f"File {path_str} is not a supported image format.")
     
     try:
-        with open(path, "rb") as f:
-            # Read raw bytes and encode to base64 for transport
-            data = base64.b64encode(f.read()).decode("utf-8")
-            return {"type": "image", "data": data, "mimeType": mime_type}
+        data = base64.b64encode(path.read_bytes()).decode("utf-8")
+        return {"type": "image", "data": data, "mimeType": mime_type}
     except Exception as e:
-        return {"text": f"Error reading image: {str(e)}"}
+        raise ToolError(f"Error reading image: {str(e)}")

tools/read_metadata.py

@@ -1,7 +1,8 @@
-import os
 import logging
+from pathlib import Path
 from typing import Any, Dict
 from PIL import Image
+from tools.utils import ToolError
 
 logger = logging.getLogger("MattCP")
 
@@ -10,15 +11,17 @@ async def handle(args: Dict[str, Any]):
     Extracts text-based metadata chunks (tEXt, zTXt, iTXt) from a PNG file.
     These often contain AI generation prompts and parameters.
     """
-    path = args.get("path")
-    if not path:
-        return {"text": "Error: Missing path argument"}
+    path_str = args.get("path")
+    if not path_str:
+        raise ToolError("Missing path argument")
+    
+    path = Path(path_str)
     
     try:
         with Image.open(path) as img:
             # Metadata extraction is specific to PNG format in this implementation
             if img.format != 'PNG':
-                return {"text": "Error: File is not a PNG image."}
+                raise ToolError("File is not a PNG image.")
             
             # Pillow parses PNG text chunks into the .info dictionary
             metadata = img.info
@@ -28,5 +31,7 @@ async def handle(args: Dict[str, Any]):
             # Format as a simple key: value list
             output = "\n".join([f"{k}: {v}" for k, v in metadata.items()])
             return {"text": f"PNG Metadata:\n{output}"}
+    except ToolError:
+        raise
     except Exception as e:
-        return {"text": f"Error reading PNG metadata: {str(e)}"}
+        raise ToolError(f"Error reading PNG metadata: {str(e)}")

tools/utils.py

@@ -1,5 +1,11 @@
 import time
 import datetime
+from pathlib import Path
+from typing import List, Dict, Any, Optional
+
+class ToolError(Exception):
+    """Custom exception for tool-related errors to be caught by the MCP server."""
+    pass
 
 def format_relative_time(timestamp: float) -> str:
     """Converts a timestamp to a human-readable relative format."""
@@ -32,3 +38,61 @@ def format_relative_time(timestamp: float) -> str:
     
     dt = datetime.datetime.fromtimestamp(timestamp)
     return dt.strftime('%Y-%m-%d')
+
+# Cross-platform font candidates
+# ImageFont.truetype can often find these by name on Windows, 
+# or we can provide paths for Linux.
+COMMON_FONTS = [
+    # Linux paths
+    "/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf",
+    "/usr/share/fonts/truetype/liberation/LiberationSans-Bold.ttf",
+    "/usr/share/fonts/truetype/freefont/FreeSansBold.ttf",
+    # Windows filenames (Pillow often finds these in the system path)
+    "arialbd.ttf", 
+    "calibrib.ttf",
+    "verdanab.ttf",
+    "tahomabd.ttf",
+]
+
+def get_file_info_list(path: Path, extensions: Optional[tuple] = None) -> List[Dict[str, Any]]:
+    """
+    Gathers basic information about files in a directory.
+    Optional extensions filter (e.g. ('.png', '.jpg')).
+    """
+    items = []
+    try:
+        for entry in path.iterdir():
+            if extensions and not entry.name.lower().endswith(extensions):
+                continue
+                
+            stats = entry.stat()
+            items.append({
+                "name": entry.name,
+                "path": entry,
+                "mtime": stats.st_mtime,
+                "size": stats.st_size,
+                "is_dir": entry.is_dir()
+            })
+    except Exception as e:
+        raise ToolError(f"Error accessing directory {path}: {e}")
+    
+    return items
+
+def sort_file_list(items: List[Dict[str, Any]], sort_by: str) -> List[Dict[str, Any]]:
+    """Sorts items based on the provided key. Directories always come first."""
+    if sort_by == "mtime":
+        # Newest first
+        items.sort(key=lambda x: (not x["is_dir"], -x["mtime"]))
+    elif sort_by == "size":
+        # Largest first
+        items.sort(key=lambda x: (not x["is_dir"], -x["size"]))
+    else:
+        # Alphabetical
+        items.sort(key=lambda x: (not x["is_dir"], x["name"].lower()))
+    return items
+
+def get_paginated_list(items: List[Any], page: int, page_size: int) -> List[Any]:
+    """Returns a slice of the list based on page and page_size."""
+    start_idx = (page - 1) * page_size
+    end_idx = start_idx + page_size
+    return items[start_idx:end_idx]