Compare commits

..
10 Commits
Author SHA1 Message Date
moosecrap f05a662414 Rename 2026-07-23 23:54:30 -07:00
moosecrap 05e70a7fc4 README, config file 2026-07-23 23:48:05 -07:00
moosecrap 504794f9a8 get_text_context 2026-07-23 14:38:45 -07:00
moosecrap 03e47bad42 Tool descriptions 2026-07-23 14:33:06 -07:00
moosecrap a2cc56200c Fix token rounding 2026-07-23 02:22:30 -07:00
moosecrap 77a88c093d Exif rotation, description update 2026-07-23 00:58:54 -07:00
moosecrap da5fdc3e4f Image preview 2026-07-23 00:24:29 -07:00
moosecrap d3d0ab9c63 Alignment and JPEG 2026-07-22 23:02:48 -07:00
moosecrap 48132aa21c Refactor 2026-07-22 22:53:15 -07:00
moosecrap d87fcb85be Token size for contact sheet 2026-07-22 22:02:52 -07:00
14 changed files with 661 additions and 258 deletions
+63
View File
@@ -0,0 +1,63 @@
# MooseCP Image Server
A Model Context Protocol (MCP) server designed to provide LLMs with efficient, token-optimized visual access to image directories. Instead of dumping full-resolution images (which waste tokens and cause context overflow), MooseCP provides a hierarchical workflow: **List $\rightarrow$ Scan $\rightarrow$ Preview $\rightarrow$ Inspect**.
## ⚠️ AI SLOP DISCLAIMER
This server is designed to facilitate the interaction between AI models and image datasets. While it provides tools for "previewing" and "inspecting" images, remember that AI models can still hallucinate visual details, especially when working with low-resolution previews. **Always verify critical visual information with the `read_image` tool (full resolution) before drawing final conclusions.**
## Tools Overview
### 📁 `list_directory`
Provides a simplified file-browser view of a directory.
* **Best for**: Getting a sense of the files present and their basic metadata (size, date).
* **Features**: Pagination and sorting by name, size, or modification date.
### 🖼️ `contact_sheet`
Generates a high-density grid of thumbnails.
* **Best for**: Quickly scanning hundreds of images to find a specific one or get a general "vibe" of a folder.
* **Note**: This tool is paginated. You must iterate through pages to see all images in a folder.
* **Workflow**: Use the indices shown on the contact sheet to call `preview_image`.
### 🔍 `preview_image`
Provides medium-detail previews optimized for the model's token budget.
* **Best for**: Comparing a few candidates, inspecting specific details, or selecting a "favorite" image.
* **Efficiency**: Automatically calculates dimensions to fit the model's patch size (e.g., 48px patches for Gemma 4), ensuring maximum detail without wasting tokens on padding.
* **Workflow**: Pass indices from the `contact_sheet` or a direct file path.
### 📖 `read_png_metadata`
Extracts AI generation parameters from PNG files.
* **Best for**: Retrieving prompts, seeds, and model hashes from AI-generated images.
### 📸 `read_image`
Returns the full-resolution image.
* **Best for**: Final confirmation or deep visual analysis where every pixel counts.
---
## Installation & Requirements
### Dependencies
This server requires Python 3.10+ and the following packages:
* `uvicorn`: ASGI server for the SSE transport.
* `starlette`: Lightweight ASGI framework.
* `Pillow`: Image processing and thumbnail generation.
```bash
pip install uvicorn starlette Pillow
```
### Setup
1. Clone this repository to your server.
2. (Optional) Edit `config.py` to adjust the server port, log level, or token budgets if you are using a model other than Gemma 4.
3. Run the server:
```bash
python main.py
```
## Configuration (`config.py`)
You can tune the server's behavior in `config.py`:
- **`PATCH_SIZE`**: Set this to `(clip.vision.patch_size * n_merge)` for your specific model to ensure token-perfect resizing.
- **`PREVIEW_TOKEN_BUDGET`**: Controls how many tokens the `preview_image` tool aims for (default: 70).
- **`CONTACT_SHEET_COLS/ROWS`**: Adjust the grid size to fit within your model's maximum context window.
- **`IMAGE_QUALITY`**: Adjust JPEG compression (1-100).
+47
View File
@@ -0,0 +1,47 @@
import logging
# --- Server & Logs ---
HOST = "127.0.0.1"
PORT = 8000
LOG_LEVEL = "WARNING" # Options: "DEBUG", "INFO", "WARNING", "ERROR"
LOG_FILE = "debug.log"
# --- Model Specific Token Tuning (Tuned for Gemma 4) ---
# Patch size is typically (clip.vision.patch_size * n_merge)
# For Gemma 4, this is 48px.
PATCH_SIZE = 48
# The target token count for a single image preview.
# The actual budget will be this or greater, but as small as possible.
PREVIEW_TOKEN_BUDGET = 70
# Contact sheet grid optimized to fit within Gemma 4's 1120 max token budget
# (10 cols * 7 rows) = 70 images.
# Each thumb (192px) is 4x4 patches.
# 70 images * (4*4) = 1120 tokens.
CONTACT_SHEET_COLS = 10
CONTACT_SHEET_ROWS = 7
CONTACT_SHEET_THUMB_SIZE = 192
# --- Image Quality ---
# JPEG image quality (usually max 95).
# See: https://pillow.readthedocs.io/en/stable/handbook/image-file-formats.html#jpeg-saving
IMAGE_QUALITY = 95
# --- Font Configuration ---
# Generic font names for cross-platform compatibility
# Pillow's truetype() can often find these by name in the system path
SYSTEM_FONT_NAMES = [
"arialbd.ttf",
"DejaVuSans-Bold",
"LiberationSans-Bold",
"Verdana",
"Tahoma",
]
# Fallback absolute paths for Linux systems
FALLBACK_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",
]
+9 -5
View File
@@ -10,15 +10,18 @@ from starlette.middleware.cors import CORSMiddleware
from mcp_logic import MCPServer from mcp_logic import MCPServer
from tools import TOOL_REGISTRY from tools import TOOL_REGISTRY
import config
# --- Configuration & Logging --- # --- Configuration & Logging ---
logging.basicConfig( logging.basicConfig(
level=logging.DEBUG, level=getattr(logging, config.LOG_LEVEL),
filename="debug.log", filename=config.LOG_FILE,
filemode="a", filemode="a",
format="%(asctime)s - %(levelname)s - %(message)s" format="%(asctime)s - %(levelname)s - %(message)s"
) )
logger = logging.getLogger("MattCP") logger = logging.getLogger("MooseCP")
mcp_logic = MCPServer() mcp_logic = MCPServer()
@@ -90,11 +93,12 @@ app.add_middleware(
if __name__ == "__main__": if __name__ == "__main__":
config = uvicorn.Config( config = uvicorn.Config(
app=app, app=app,
host="127.0.0.1", host=config.HOST,
port=8000, port=config.PORT,
log_level="info", log_level="info",
timeout_graceful_shutdown=2 # Reduced from 5 to 2 seconds timeout_graceful_shutdown=2 # Reduced from 5 to 2 seconds
) )
server = uvicorn.Server(config) server = uvicorn.Server(config)
def signal_handler(sig, frame): def signal_handler(sig, frame):
+10 -2
View File
@@ -3,6 +3,7 @@ import uuid
import json import json
from typing import Any, Dict, Union, List from typing import Any, Dict, Union, List
from starlette.responses import Response from starlette.responses import Response
from tools.utils import ToolError
class MCPServer: class MCPServer:
def __init__(self): def __init__(self):
@@ -23,7 +24,7 @@ class MCPServer:
return { return {
"protocolVersion": "2024-11-05", "protocolVersion": "2024-11-05",
"capabilities": {"tools": {}}, "capabilities": {"tools": {}},
"serverInfo": {"name": "MattCP", "version": "0.4.1"} "serverInfo": {"name": "MooseCP", "version": "0.4.1"}
} }
if method == "tools/list": if method == "tools/list":
@@ -40,7 +41,14 @@ class MCPServer:
media_type="application/json" 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) # Handle results that are already lists of content (for multimodal/multi-part responses)
if isinstance(result_data, list): if isinstance(result_data, list):
-5
View File
@@ -1,5 +0,0 @@
#!/bin/bash
# Start the MCP Image Server in the background
nohup /home/matt/code/llama.cpp/build/bin/mcp-image-server/venv/bin/python3 /home/matt/code/llama.cpp/build/bin/mcp-image-server/server.py > /home/matt/code/llama.cpp/build/bin/mcp-image-server/server.log 2>&1 &
echo "MCP Image Server started in background on port 8000"
echo "Log file: /home/matt/code/llama.cpp/build/bin/mcp-image-server/server.log"
+39 -9
View File
@@ -18,14 +18,16 @@ class Tool:
# Import handlers from separate files # Import handlers from separate files
from .read_image import handle as read_image_handler from .read_image import handle as read_image_handler
from .read_metadata import handle as read_png_metadata_handler from .read_metadata import handle as read_png_metadata_handler
from .list_thumbnails import handle as list_thumbnails_handler from .contact_sheet import handle as contact_sheet_handler
from .list_directory import handle as list_directory_details_handler from .list_directory import handle as list_directory_details_handler
from .preview_image import handle as preview_image_handler
from .get_text_context import handle as get_text_context_handler
# Central registry of all available tools # Central registry of all available tools
TOOL_REGISTRY = [ TOOL_REGISTRY = [
Tool( Tool(
name="read_image", name="read_image",
description="Reads an image from the disk and returns it as an image content object.", description="Reads an image at full resolution for maximum detail. Use this for final confirmation of a selected image or when deep visual analysis of a specific file is required.",
schema={ schema={
"type": "object", "type": "object",
"properties": {"path": {"type": "string", "description": "Path to the image file"}}, "properties": {"path": {"type": "string", "description": "Path to the image file"}},
@@ -35,7 +37,7 @@ TOOL_REGISTRY = [
), ),
Tool( Tool(
name="read_png_metadata", name="read_png_metadata",
description="Reads the metadata (tEXt, zTXt, iTXt) from a PNG image to fetch info such as AI prompts.", description="Reads the metadata (tEXt, zTXt, iTXt) from a PNG image to fetch info such as prompts for AI generated images.",
schema={ schema={
"type": "object", "type": "object",
"properties": {"path": {"type": "string", "description": "Path to the PNG file"}}, "properties": {"path": {"type": "string", "description": "Path to the PNG file"}},
@@ -44,23 +46,22 @@ TOOL_REGISTRY = [
handler=read_png_metadata_handler handler=read_png_metadata_handler
), ),
Tool( Tool(
name="list_thumbnails", name="contact_sheet",
description="Generates a thumbnail grid of images in a directory. Supports pagination and sorting.", description="Generates a coarse, high-density overview of a directory. This tool is paginated; a single page may not show all images. To perform a comprehensive scan or find specific images, you MUST iterate through multiple pages. DO NOT use this for selecting specific images or making quality judgments; use preview_image for those tasks.",
schema={ schema={
"type": "object", "type": "object",
"properties": { "properties": {
"path": {"type": "string", "description": "Path to the directory containing images"}, "path": {"type": "string", "description": "Path to the directory containing images"},
"page": {"type": "integer", "description": "The page number to retrieve (1-indexed)", "default": 1}, "page": {"type": "integer", "description": "The page number to retrieve (1-indexed). Increment this value to navigate through the full list of images in the folder.", "default": 1},
"page_size": {"type": "integer", "description": "Number of images per page", "default": 64},
"sort_by": {"type": "string", "description": "Sort order: 'mtime' (newest first, default), 'name' (alphabetical), or 'size' (largest first)", "default": "mtime"}, "sort_by": {"type": "string", "description": "Sort order: 'mtime' (newest first, default), 'name' (alphabetical), or 'size' (largest first)", "default": "mtime"},
}, },
"required": ["path"], "required": ["path"],
}, },
handler=list_thumbnails_handler handler=contact_sheet_handler
), ),
Tool( Tool(
name="list_directory", name="list_directory",
description="Lists the contents of a directory in a file browser format. Directories first, then files. Supports pagination and sorting.", description="Lists the contents of a directory in a simplified format. Directories first, then files. Supports pagination and sorting.",
schema={ schema={
"type": "object", "type": "object",
"properties": { "properties": {
@@ -73,4 +74,33 @@ TOOL_REGISTRY = [
}, },
handler=list_directory_details_handler handler=list_directory_details_handler
), ),
Tool(
name="preview_image",
description="Provides low-resolution previews and detailed file info for specific images. It uses the minimum viable token budget, so some fine details will be missed. This is the PRIMARY tool for inspecting files, comparing candidates, or selecting images before moving to read_image for maximum detail.",
schema={
"type": "object",
"properties": {
"path": {"type": "string", "description": "Path to an image file or a directory containing images"},
"indices": {"type": "string", "description": "1-based indices or ranges (e.g., '1, 3-5, 10') to preview from the directory. Required if path is a directory."},
"sort_by": {"type": "string", "description": "Sort order to resolve indices: 'mtime' (default), 'name', or 'size'."},
},
"required": ["path"],
},
handler=preview_image_handler
),
Tool(
name="get_text_context",
description="Extracts specific sections of a file with absolute line numbers and surrounding context. This tool MUST be called immediately before edit_file to verify the exact line numbers and content of the block being replaced, preventing misalignment errors.",
schema={
"type": "object",
"properties": {
"path": {"type": "string", "description": "Path to the file"},
"pattern": {"type": "string", "description": "Regex pattern to search for"},
"lines": {"type": "string", "description": "1-based indices or ranges (e.g., '10-20, 45')"},
"context": {"type": "integer", "description": "Number of lines of context to show above and below", "default": 1},
},
"required": ["path"],
},
handler=get_text_context_handler
),
] ]
+150
View File
@@ -0,0 +1,150 @@
import base64
import logging
import io
import datetime
from pathlib import Path
from typing import Any, Dict, List
from PIL import Image, ImageDraw, ImageFont, ImageOps
import config
from tools.utils import format_relative_time, ToolError, get_file_info_list, sort_file_list, get_paginated_list
logger = logging.getLogger("MooseCP")
async def handle(args: Dict[str, Any]):
"""
Generates a contact sheet of images.
Grid: 10 columns x 7 rows (70 images total).
Sized for optimal token usage (1120 tokens) on llama.cpp.
"""
dir_path_str = args.get("path")
page = int(args.get("page", 1))
sort_by = args.get("sort_by", "mtime")
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 = get_file_info_list(dir_path, extensions=exts)
if not file_info_list:
return {"text": "No supported images found in the directory."}
# Sorting
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 = config.CONTACT_SHEET_COLS
ROWS = config.CONTACT_SHEET_ROWS
PAGE_SIZE = COLS * ROWS
total_files = len(file_info_list)
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
canvas = Image.new('RGB', (canvas_w, canvas_h), (30, 30, 30))
draw = ImageDraw.Draw(canvas)
# Font loading
font = None
# Try generic system font names first
for font_name in config.SYSTEM_FONT_NAMES:
try:
font = ImageFont.truetype(font_name, 24)
break
except Exception:
continue
# Fallback to absolute paths
if font is None:
for font_path in config.FALLBACK_FONT_PATHS:
try:
font = ImageFont.truetype(font_path, 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
col = i % COLS
x = col * thumb_size
y = row * thumb_size
global_index = start_idx + i + 1
try:
with Image.open(full_path) as img:
# Apply EXIF orientation to fix sideways images
img = ImageOps.exif_transpose(img)
img.thumbnail((thumb_size, thumb_size))
off_x = (thumb_size - img.width) // 2
off_y = (thumb_size - img.height) // 2
canvas.paste(img, (x + off_x, y + off_y))
text = str(global_index)
if font:
bbox = draw.textbbox((0, 0), text, font=font)
text_w, text_h = bbox[2] - bbox[0], bbox[3] - bbox[1]
else:
text_w, text_h = len(text) * 8, 15
# Adjusted black box to align perfectly with the bottom of the image (y + thumb_size)
draw.rectangle([x, y + thumb_size - text_h - 4, x + text_w + 5, y + thumb_size], fill=(0, 0, 0, 180))
draw.text((x + 2, y + thumb_size - text_h - 7), text, fill=(255, 255, 255), font=font)
except Exception as e:
logger.error(f"Failed to process {info['name']}: {e}")
draw.text((x + 5, y + thumb_size // 2), "Error", fill=(255, 0, 0), font=font)
buf = io.BytesIO()
# Save as JPEG to save data
canvas.save(buf, format='JPEG', quality=config.IMAGE_QUALITY)
img_data = base64.b64encode(buf.getvalue()).decode("utf-8")
total_pages = (total_files + PAGE_SIZE - 1) // PAGE_SIZE
first_on_page = paged_files[0]
last_on_page = paged_files[-1]
if sort_by == "mtime":
first_val = f"{format_relative_time(first_on_page['mtime'])} ({datetime.datetime.fromtimestamp(first_on_page['mtime']).strftime('%Y-%m-%d')})"
last_val = f"{format_relative_time(last_on_page['mtime'])} ({datetime.datetime.fromtimestamp(last_on_page['mtime']).strftime('%Y-%m-%d')})"
elif sort_by == "size":
first_val = f"{first_on_page['size']/1024:.1f}KB"
last_val = f"{last_on_page['size']/1024:.1f}KB"
else:
first_val = first_on_page['name']
last_val = last_on_page['name']
header_text = (
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}"
)
return [
{"type": "text", "text": header_text},
{"type": "image", "data": img_data, "mimeType": "image/jpeg"}
]
+86
View File
@@ -0,0 +1,86 @@
import re
import logging
from pathlib import Path
from typing import Any, Dict, List, Set
from tools.utils import ToolError, parse_indices
logger = logging.getLogger("MooseCP")
async def handle(args: Dict[str, Any]):
"""
Extracts specific sections of a file with absolute line numbers and surrounding context.
This tool MUST be called immediately before edit_file to verify the exact line numbers
and content of the block being replaced, preventing misalignment errors.
"""
path_str = args.get("path")
pattern = args.get("pattern")
lines_str = args.get("lines")
context = int(args.get("context", 1))
if not path_str:
raise ToolError("Missing path argument")
path = Path(path_str)
if not path.is_file():
raise ToolError(f"{path_str} is not a file.")
try:
with open(path, "r", encoding="utf-8", errors="replace") as f:
all_lines = f.readlines()
except Exception as e:
raise ToolError(f"Error reading file {path_str}: {e}")
total_lines = len(all_lines)
# Case 1: No pattern and no lines provided -> Dump whole file
if not pattern and not lines_str:
output = [f"{i+1}: {line}" for i, line in enumerate(all_lines)]
return {"text": "".join(output)}
# Determine target line numbers (1-based)
target_lines: Set[int] = set()
if pattern:
try:
regex = re.compile(pattern)
for i, line in enumerate(all_lines):
if regex.search(line):
target_lines.add(i + 1)
except re.error as e:
raise ToolError(f"Invalid regex pattern: {e}")
if lines_str:
try:
indices = parse_indices(lines_str)
for idx in indices:
if 1 <= idx <= total_lines:
target_lines.add(idx)
except ToolError as e:
raise e
except Exception as e:
raise ToolError(f"Error parsing line indices: {e}")
if not target_lines:
return {"text": "No matching lines found."}
# Expand targets with context
lines_to_show: Set[int] = set()
for line in target_lines:
for offset in range(-context, context + 1):
ln = line + offset
if 1 <= ln <= total_lines:
lines_to_show.add(ln)
# Sort and format output
sorted_lines = sorted(list(lines_to_show))
output = []
last_line = None
for ln in sorted_lines:
if last_line is not None and ln > last_line + 1:
output.append("...\n")
output.append(f"{ln}: {all_lines[ln-1]}")
last_line = ln
return {"text": "".join(output)}
+23 -58
View File
@@ -1,10 +1,9 @@
import os
import logging import logging
import datetime from pathlib import Path
from typing import Any, Dict 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") logger = logging.getLogger("MooseCP")
async def handle(args: Dict[str, Any]): async def handle(args: Dict[str, Any]):
""" """
@@ -12,65 +11,24 @@ async def handle(args: Dict[str, Any]):
Directories are listed first, then files. Directories are listed first, then files.
Supports pagination and sorting. Supports pagination and sorting.
""" """
path = args.get("path") path_str = args.get("path")
page = int(args.get("page", 1)) page = int(args.get("page", 1))
page_size = int(args.get("page_size", 64)) page_size = int(args.get("page_size", 64))
sort_by = args.get("sort_by", "mtime") sort_by = args.get("sort_by", "mtime")
if not path: if not path_str:
return {"text": "Error: Missing path argument"} raise ToolError("Missing path argument")
if not os.path.isdir(path):
return {"text": f"Error: {path} is not a directory."}
try: path = Path(path_str)
entries = os.listdir(path) if not path.is_dir():
except Exception as e: raise ToolError(f"{path_str} is not a directory.")
return {"text": f"Error reading directory: {e}"}
all_items = [] all_items = get_file_info_list(path)
for entry in entries: all_items = sort_file_list(all_items, sort_by)
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()))
total_items = len(all_items) 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: if total_items <= 100:
start_idx = 0 start_idx = 0
end_idx = total_items end_idx = total_items
@@ -89,11 +47,18 @@ async def handle(args: Dict[str, Any]):
output = [] output = []
for item in paged_items: for item in paged_items:
if item["type"] == "dir": mtime_rel = format_relative_time(item["mtime"])
output.append(f"[DIR] {item['name']} | {item['info']} | {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: 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)} return {"text": f"{header}\n\n" + "\n".join(output)}
-159
View File
@@ -1,159 +0,0 @@
import os
import base64
import logging
import asyncio
import io
import datetime
import time
from typing import Any, Dict
from PIL import Image, ImageDraw, ImageFont
logger = logging.getLogger("MattCP")
async def handle(args: Dict[str, Any]):
"""
Generates a thumbnail grid of images in a directory.
Returns a combined text list (with global indices) and a visual contact sheet.
"""
dir_path = 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 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."}
exts = ('.png', '.jpg', '.jpeg', '.webp', '.bmp')
file_info_list = []
# 1. Gather all valid images and their stats
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}")
if not file_info_list:
return {"text": "No supported images found in the directory."}
# 2. Sort the list based on requested criteria
if sort_by == "mtime":
file_info_list.sort(key=lambda x: x["mtime"], reverse=True)
elif sort_by == "size":
file_info_list.sort(key=lambda x: x["size"], reverse=True)
else:
file_info_list.sort(key=lambda x: x["name"].lower())
# 3. Apply Pagination
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]
if not paged_files:
return {"text": f"No images found on page {page}."}
# Layout constants
thumb_size = 160
padding = 15
label_height = 35
num_files = len(paged_files)
cols = int(num_files**0.5) if num_files > 0 else 1
if cols == 0: cols = 1
rows = (num_files + cols - 1) // cols
canvas_w = cols * (thumb_size + padding) + padding
canvas_h = rows * (thumb_size + label_height + padding) + padding
# Create dark gray background
canvas = Image.new('RGB', (canvas_w, canvas_h), (30, 30, 30))
draw = ImageDraw.Draw(canvas)
# Attempt to load a bold system font for the indices
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, 20)
break
except Exception:
continue
if font is None:
font = ImageFont.load_default()
file_details = []
# 4. Paste thumbnails and render indices
for i, info in enumerate(paged_files):
filename = info["name"]
full_path = info["path"]
row = i // cols
col = i % cols
x = padding + col * (thumb_size + padding)
y = padding + row * (thumb_size + label_height + padding)
# Continuous indexing across pages
global_index = start_idx + i + 1
try:
with Image.open(full_path) as img:
width, height = img.size
img.thumbnail((thumb_size, thumb_size))
# Center image in its slot
off_x = (thumb_size - img.width) // 2
off_y = (thumb_size - img.height) // 2
canvas.paste(img, (x + off_x, y + off_y))
# Format text metadata
from tools.utils import format_relative_time # Import helper from utils
mod_time_rel = format_relative_time(info["mtime"])
size_kb = info["size"] / 1024
file_details.append(f"{global_index}. {filename} | {width}x{height} | {size_kb:.1f}KB | {mod_time_rel}")
# Draw the index number centered under the image
text = str(global_index)
if font:
bbox = draw.textbbox((0, 0), text, font=font)
text_w = bbox[2] - bbox[0]
else:
text_w = len(text) * 7
text_x = x + (thumb_size - text_w) // 2
draw.text((text_x, y + thumb_size + 2), text, fill=(255, 255, 255), font=font)
except Exception as e:
logger.error(f"Failed to process {filename}: {e}")
draw.text((x, y + thumb_size // 2), "Error", fill=(255, 0, 0), font=font)
file_details.append(f"{global_index}. {filename} | ERROR")
# 5. Encode result to base64 PNG
buf = io.BytesIO()
canvas.save(buf, format='PNG')
img_data = base64.b64encode(buf.getvalue()).decode("utf-8")
total_pages = (total_files + page_size - 1) // page_size
header_text = f"Directory: {dir_path}\nPage {page} of {total_pages} ({total_files} total images). Showing {start_idx+1}-{min(end_idx, total_files)}."
details_text = "\n".join(file_details)
return [
{"type": "text", "text": f"{header_text}\n\nFile List:\n{details_text}"},
{"type": "image", "data": img_data, "mimeType": "image/png"}
]
+132
View File
@@ -0,0 +1,132 @@
import math
import base64
import logging
import io
import datetime
from pathlib import Path
from typing import Any, Dict, List, Tuple
from PIL import Image, ImageOps
import config
from tools.utils import format_relative_time, ToolError, get_file_info_list, sort_file_list, parse_indices
logger = logging.getLogger("MooseCP")
def calculate_patch_dimensions(orig_w: int, orig_h: int, target_tokens: int = None):
"""
Calculates optimal pixel dimensions to hit a token budget of at least target_tokens.
Validates budget based on actual rounded pixel dimensions to avoid float precision errors.
"""
if target_tokens is None:
target_tokens = config.PREVIEW_TOKEN_BUDGET
ar = orig_w / orig_h
# Case A: Width is the anchor (non-padded)
w_anchor = 1
while True:
w_px = w_anchor * config.PATCH_SIZE
h_px = round(w_px / ar)
# Calculate actual tokens based on resulting pixel dimensions
tokens = math.ceil(w_px / config.PATCH_SIZE) * math.ceil(h_px / config.PATCH_SIZE)
if tokens >= target_tokens:
res_a = {"w": w_px, "h": h_px, "tokens": tokens}
break
w_anchor += 1
# Case B: Height is the anchor (non-padded)
h_anchor = 1
while True:
h_px = h_anchor * config.PATCH_SIZE
w_px = round(h_px * ar)
# Calculate actual tokens based on resulting pixel dimensions
tokens = math.ceil(w_px / config.PATCH_SIZE) * math.ceil(h_px / config.PATCH_SIZE)
if tokens >= target_tokens:
res_b = {"w": w_px, "h": h_px, "tokens": tokens}
break
h_anchor += 1
# Pick the one with the smaller token count
best = res_a if res_a["tokens"] <= res_b["tokens"] else res_b
return max(1, best["w"]), max(1, best["h"])
async def handle(args: Dict[str, Any]):
"""
Generates small thumbnails of images with detailed info.
Can take a single file path, or a directory with indices/ranges from contact_sheet.
Thumbnails are optimized for ~70 token usage.
"""
path_str = args.get("path")
indices_str = args.get("indices")
sort_by = args.get("sort_by", "mtime")
if not path_str:
raise ToolError("Missing path argument")
path = Path(path_str)
# Store as (path, index_label) where index_label is None if direct path
target_files: List[Tuple[Path, Any]] = []
if path.is_file():
target_files.append((path, None))
elif path.is_dir():
if not indices_str:
raise ToolError("Indices are required when providing a directory path. Example: '1, 5-10'")
exts = ('.png', '.jpg', '.jpeg', '.webp', '.bmp')
file_info_list = get_file_info_list(path, extensions=exts)
file_info_list = sort_file_list(file_info_list, sort_by)
indices = parse_indices(indices_str)
for idx in indices:
# contact_sheet uses 1-based indexing
if 1 <= idx <= len(file_info_list):
target_files.append((file_info_list[idx-1]["path"], idx))
else:
logger.warning(f"Index {idx} out of range for directory {path_str}")
else:
raise ToolError(f"Path {path_str} is neither a file nor a directory.")
if not target_files:
return {"text": "No valid images found to preview."}
results = []
for file_path, index_label in target_files:
try:
stats = file_path.stat()
with Image.open(file_path) as img:
# Apply EXIF orientation to fix sideways images
img = ImageOps.exif_transpose(img)
orig_w, orig_h = img.size
target_w, target_h = calculate_patch_dimensions(orig_w, orig_h)
# Use thumbnail() to preserve aspect ratio and fit within the target bounding box.
thumb = img.convert("RGB")
thumb.thumbnail((target_w, target_h), Image.Resampling.LANCZOS)
buf = io.BytesIO()
thumb.save(buf, format='JPEG', quality=config.IMAGE_QUALITY)
img_data = base64.b64encode(buf.getvalue()).decode("utf-8")
# Gather info
mtime_rel = format_relative_time(stats.st_mtime)
mtime_abs = datetime.datetime.fromtimestamp(stats.st_mtime).strftime('%Y-%m-%d %H:%M')
size_kb = stats.st_size / 1024
# Prepend index if this image was selected via index/range
idx_prefix = f"[Index: {index_label}] " if index_label is not None else ""
info_text = (
f"{idx_prefix}File: {file_path.name}\n"
f"Resolution: {orig_w}x{orig_h}\n"
f"Size: {size_kb:.1f}KB\n"
f"Modified: {mtime_rel} ({mtime_abs})"
)
results.append({"type": "text", "text": info_text})
results.append({"type": "image", "data": img_data, "mimeType": "image/jpeg"})
except Exception as e:
results.append({"type": "text", "text": f"Error processing {file_path.name}: {e}"})
return results
+13 -12
View File
@@ -1,29 +1,30 @@
import os
import base64 import base64
import mimetypes import mimetypes
import logging import logging
from pathlib import Path
from typing import Any, Dict from typing import Any, Dict
from tools.utils import ToolError
logger = logging.getLogger("MattCP") logger = logging.getLogger("MooseCP")
async def handle(args: Dict[str, Any]): async def handle(args: Dict[str, Any]):
""" """
Reads an image file from disk and returns it as a base64 encoded string Reads an image file from disk and returns it as a base64 encoded string
within an MCP ImageContent object. within an MCP ImageContent object.
""" """
path = args.get("path") path_str = args.get("path")
if not path: if not path_str:
return {"text": "Error: Missing path argument"} raise ToolError("Missing path argument")
path = Path(path_str)
# Validate that the file has an image-like MIME type # 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/"): 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: try:
with open(path, "rb") as f: data = base64.b64encode(path.read_bytes()).decode("utf-8")
# Read raw bytes and encode to base64 for transport return {"type": "image", "data": data, "mimeType": mime_type}
data = base64.b64encode(f.read()).decode("utf-8")
return {"type": "image", "data": data, "mimeType": mime_type}
except Exception as e: except Exception as e:
return {"text": f"Error reading image: {str(e)}"} raise ToolError(f"Error reading image: {str(e)}")
+12 -7
View File
@@ -1,24 +1,27 @@
import os
import logging import logging
from pathlib import Path
from typing import Any, Dict from typing import Any, Dict
from PIL import Image from PIL import Image
from tools.utils import ToolError
logger = logging.getLogger("MattCP") logger = logging.getLogger("MooseCP")
async def handle(args: Dict[str, Any]): async def handle(args: Dict[str, Any]):
""" """
Extracts text-based metadata chunks (tEXt, zTXt, iTXt) from a PNG file. Extracts text-based metadata chunks (tEXt, zTXt, iTXt) from a PNG file.
These often contain AI generation prompts and parameters. These often contain AI generation prompts and parameters.
""" """
path = args.get("path") path_str = args.get("path")
if not path: if not path_str:
return {"text": "Error: Missing path argument"} raise ToolError("Missing path argument")
path = Path(path_str)
try: try:
with Image.open(path) as img: with Image.open(path) as img:
# Metadata extraction is specific to PNG format in this implementation # Metadata extraction is specific to PNG format in this implementation
if img.format != 'PNG': 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 # Pillow parses PNG text chunks into the .info dictionary
metadata = img.info metadata = img.info
@@ -28,5 +31,7 @@ async def handle(args: Dict[str, Any]):
# Format as a simple key: value list # Format as a simple key: value list
output = "\n".join([f"{k}: {v}" for k, v in metadata.items()]) output = "\n".join([f"{k}: {v}" for k, v in metadata.items()])
return {"text": f"PNG Metadata:\n{output}"} return {"text": f"PNG Metadata:\n{output}"}
except ToolError:
raise
except Exception as e: except Exception as e:
return {"text": f"Error reading PNG metadata: {str(e)}"} raise ToolError(f"Error reading PNG metadata: {str(e)}")
+76
View File
@@ -1,5 +1,12 @@
import time import time
import datetime 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: def format_relative_time(timestamp: float) -> str:
"""Converts a timestamp to a human-readable relative format.""" """Converts a timestamp to a human-readable relative format."""
@@ -32,3 +39,72 @@ def format_relative_time(timestamp: float) -> str:
dt = datetime.datetime.fromtimestamp(timestamp) dt = datetime.datetime.fromtimestamp(timestamp)
return dt.strftime('%Y-%m-%d') return dt.strftime('%Y-%m-%d')
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]
def parse_indices(indices_str: str) -> List[int]:
"""
Parses a string like '11, 31-33, 44' into a list of 1-based indices.
Preserves the order specified in the string.
"""
indices = []
parts = [p.strip() for p in indices_str.split(',')]
for part in parts:
if not part:
continue
if '-' in part:
try:
start, end = map(int, part.split('-'))
step = 1 if start <= end else -1
for i in range(start, end + step, step):
indices.append(i)
except ValueError:
raise ToolError(f"Invalid range format: {part}")
else:
try:
indices.append(int(part))
except ValueError:
raise ToolError(f"Invalid index format: {part}")
return indices