diff --git a/mcp_logic.py b/mcp_logic.py index 4dd961c..83eeeab 100644 --- a/mcp_logic.py +++ b/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): diff --git a/tools/contact_sheet.py b/tools/contact_sheet.py index 6c09b3b..dbca8cf 100644 --- a/tools/contact_sheet.py +++ b/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}" ) diff --git a/tools/list_directory.py b/tools/list_directory.py index 1bbb2bd..37a567d 100644 --- a/tools/list_directory.py +++ b/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)} diff --git a/tools/read_image.py b/tools/read_image.py index 70a0ba3..a49140a 100644 --- a/tools/read_image.py +++ b/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)}") diff --git a/tools/read_metadata.py b/tools/read_metadata.py index f7bcec4..202512c 100644 --- a/tools/read_metadata.py +++ b/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)}") diff --git a/tools/utils.py b/tools/utils.py index 0709166..46cf9ca 100644 --- a/tools/utils.py +++ b/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]