Initial commit

.gitignore

@@ -0,0 +1,50 @@
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# Virtual Environments
+venv/
+.venv/
+env/
+.env/
+ENV/
+env.bak/
+venv.bak/
+
+# IDEs / Editors
+.vscode/
+.idea/
+*.swp
+*.swo
+
+# Logs and temporary files
+*.log
+debug.log
+
+# OS generated files
+.DS_Store
+Thumbs.db

main.py

@@ -0,0 +1,105 @@
+import asyncio
+import logging
+import uvicorn
+import signal
+from starlette.applications import Starlette
+from starlette.routing import Route
+from starlette.responses import Response, StreamingResponse
+from starlette.middleware.cors import CORSMiddleware
+
+from mcp_logic import MCPServer
+from tools import TOOL_REGISTRY
+
+# --- Configuration & Logging ---
+logging.basicConfig(
+    level=logging.DEBUG,
+    filename="debug.log",
+    filemode="a",
+    format="%(asctime)s - %(levelname)s - %(message)s"
+)
+logger = logging.getLogger("MattCP")
+
+mcp_logic = MCPServer()
+
+# --- Transport Layer (Starlette/SSE) ---
+
+async def sse_endpoint(request):
+    sid = mcp_logic.create_session()
+    logger.debug(f"Session started: {sid}")
+    
+    async def event_generator():
+        try:
+            yield f"event: endpoint\ndata: /messages?session_id={sid}\n\n"
+            queue = mcp_logic.sessions[sid]
+            while mcp_logic.running:
+                try:
+                    msg = await asyncio.wait_for(queue.get(), timeout=1.0)
+                    yield f"event: message\ndata: {msg}\n\n"
+                    queue.task_done()
+                except asyncio.TimeoutError:
+                    continue
+        except asyncio.CancelledError:
+            logger.debug(f"Session cancelled: {sid}")
+        finally:
+            mcp_logic.sessions.pop(sid, None)
+            logger.debug(f"Session closed: {sid}")
+
+    return StreamingResponse(event_generator(), media_type="text/event-stream")
+
+async def messages_endpoint(request):
+    sid = request.query_params.get("session_id")
+    if not sid or sid not in mcp_logic.sessions:
+        return Response("Session not found", status_code=404)
+
+    try:
+        body = await request.json()
+    except Exception as e:
+        return Response(f"Invalid JSON: {e}", status_code=400)
+
+    method = body.get("method")
+    params = body.get("params", {})
+    req_id = body.get("id")
+    
+    logger.debug(f"Request: {method} (ID: {req_id})")
+    
+    result = await mcp_logic.handle_request(method, params, req_id, TOOL_REGISTRY)
+    
+    if isinstance(result, Response):
+        return result
+    
+    await mcp_logic.send_to_session(sid, {"jsonrpc": "2.0", "id": req_id, "result": result})
+    return Response(status_code=202)
+
+app = Starlette(
+    routes=[
+        Route("/sse", endpoint=sse_endpoint),
+        Route("/messages", endpoint=messages_endpoint, methods=["POST"]),
+    ]
+)
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+if __name__ == "__main__":
+    config = uvicorn.Config(
+        app=app, 
+        host="127.0.0.1", 
+        port=8000, 
+        log_level="info",
+        timeout_graceful_shutdown=5
+    )
+    server = uvicorn.Server(config)
+    
+    def signal_handler(sig, frame):
+        logger.info("Shutdown signal received")
+        mcp_logic.running = False
+    
+    signal.signal(signal.SIGINT, signal_handler)
+    signal.signal(signal.SIGTERM, signal_handler)
+
+    asyncio.run(server.serve())

mcp_logic.py

@@ -0,0 +1,60 @@
+import asyncio
+import uuid
+import json
+from typing import Any, Dict, Union, List
+from starlette.responses import Response
+
+class MCPServer:
+    def __init__(self):
+        self.sessions: Dict[str, asyncio.Queue] = {}
+        self.running = True
+
+    def create_session(self) -> str:
+        sid = str(uuid.uuid4())
+        self.sessions[sid] = asyncio.Queue()
+        return sid
+
+    async def send_to_session(self, sid: str, message: Dict[str, Any]):
+        if sid in self.sessions:
+            await self.sessions[sid].put(json.dumps(message))
+
+    async def handle_request(self, method: str, params: Dict[str, Any], req_id: Any, tool_registry: list) -> Union[Dict[str, Any], Response]:
+        if method == "initialize":
+            return {
+                "protocolVersion": "2024-11-05",
+                "capabilities": {"tools": {}},
+                "serverInfo": {"name": "MattCP", "version": "0.4.1"}
+            }
+        
+        if method == "tools/list":
+            return {"tools": [t.to_mcp_dict() for t in tool_registry]}
+        
+        if method == "tools/call":
+            name = params.get("name")
+            args = params.get("arguments", {})
+            
+            tool = next((t for t in tool_registry if t.name == name), None)
+            if not tool:
+                return Response(
+                    content=json.dumps({"jsonrpc": "2.0", "id": req_id, "error": {"code": -32601, "message": f"Tool {name} not found"}}),
+                    media_type="application/json"
+                )
+            
+            result_data = await tool.handler(args)
+            
+            # Handle results that are already lists of content (for multimodal/multi-part responses)
+            if isinstance(result_data, list):
+                content = result_data
+            elif isinstance(result_data, dict) and "text" in result_data:
+                content = [{"type": "text", "text": result_data["text"]}]
+            elif isinstance(result_data, dict) and result_data.get("type") == "image":
+                content = [result_data]
+            else:
+                content = [result_data]
+                
+            return {"content": content}
+
+        return Response(
+            content=json.dumps({"jsonrpc": "2.0", "id": req_id, "error": {"code": -32601, "message": f"Method {method} not found"}}),
+            media_type="application/json"
+        )

start.sh

@@ -0,0 +1,5 @@
+#!/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"

tools.py

@@ -0,0 +1,185 @@
+import os
+import base64
+import mimetypes
+import logging
+import asyncio
+import io
+import datetime
+from typing import Any, Callable, Dict, List, Union
+from PIL import Image, ImageDraw, ImageFont
+
+logger = logging.getLogger("MattCP")
+
+class Tool:
+    def __init__(self, name: str, description: str, schema: Dict[str, Any], handler: Callable):
+        self.name = name
+        self.description = description
+        self.schema = schema
+        self.handler = handler
+
+    def to_mcp_dict(self):
+        return {
+            "name": self.name,
+            "description": self.description,
+            "inputSchema": self.schema
+        }
+
+async def read_image_handler(args: Dict[str, Any]):
+    path = args.get("path")
+    if not path:
+        return {"text": "Error: Missing path argument"}
+    
+    mime_type, _ = mimetypes.guess_type(path)
+    if not mime_type or not mime_type.startswith("image/"):
+        return {"text": f"Error: File {path} is not a supported image format."}
+    
+    try:
+        with open(path, "rb") as f:
+            data = base64.b64encode(f.read()).decode("utf-8")
+            return {"type": "image", "data": data, "mimeType": mime_type}
+    except Exception as e:
+        return {"text": f"Error reading image: {str(e)}"}
+
+async def read_png_metadata_handler(args: Dict[str, Any]):
+    path = args.get("path")
+    if not path:
+        return {"text": "Error: Missing path argument"}
+    
+    try:
+        with Image.open(path) as img:
+            if img.format != 'PNG':
+                return {"text": "Error: File is not a PNG image."}
+            metadata = img.info
+            if not metadata:
+                return {"text": "No metadata found in this PNG."}
+            output = "\n".join([f"{k}: {v}" for k, v in metadata.items()])
+            return {"text": f"PNG Metadata:\n{output}"}
+    except Exception as e:
+        return {"text": f"Error reading PNG metadata: {str(e)}"}
+
+async def list_thumbnails_handler(args: Dict[str, Any]):
+    dir_path = args.get("path")
+    page = int(args.get("page", 1))
+    page_size = int(args.get("page_size", 64))
+    
+    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')
+    all_files = sorted([f for f in os.listdir(dir_path) if f.lower().endswith(exts)])
+    
+    if not all_files:
+        return {"text": "No supported images found in the directory."}
+
+    total_files = len(all_files)
+    start_idx = (page - 1) * page_size
+    end_idx = start_idx + page_size
+    files = all_files[start_idx:end_idx]
+    
+    if not files:
+        return {"text": f"No images found on page {page}."}
+
+    thumb_size = 160 
+    padding = 15
+    label_height = 30
+    
+    num_files = len(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
+    
+    canvas = Image.new('RGB', (canvas_w, canvas_h), (30, 30, 30))
+    draw = ImageDraw.Draw(canvas)
+    
+    try:
+        font = ImageFont.load_default()
+    except Exception:
+        font = None
+
+    file_details = []
+
+    for i, filename in enumerate(files):
+        full_path = os.path.join(dir_path, filename)
+        row = i // cols
+        col = i % cols
+        
+        x = padding + col * (thumb_size + padding)
+        y = padding + row * (thumb_size + label_height + padding)
+        
+        # Gather metadata for the text list
+        try:
+            stats = os.stat(full_path)
+            with Image.open(full_path) as img:
+                width, height = img.size
+                # Process thumbnail
+                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))
+            
+            mod_time = datetime.datetime.fromtimestamp(stats.st_mtime).strftime('%Y-%m-%d %H:%M')
+            size_kb = stats.st_size / 1024
+            file_details.append(f"{i+1}. {filename} | {width}x{height} | {size_kb:.1f}KB | {mod_time}")
+            
+            # Label the thumbnail with just the index number
+            draw.text((x, y + thumb_size + 2), str(i + 1), fill=(220, 220, 220), 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"{i+1}. {filename} | ERROR")
+
+    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"}
+    ]
+
+TOOL_REGISTRY = [
+    Tool(
+        name="read_image",
+        description="Reads an image from the disk and returns it as an image content object.",
+        schema={
+            "type": "object",
+            "properties": {"path": {"type": "string", "description": "Path to the image file"}},
+            "required": ["path"],
+        },
+        handler=read_image_handler
+    ),
+    Tool(
+        name="read_png_metadata",
+        description="Reads the metadata (tEXt, zTXt, iTXt) from a PNG image to fetch info such as AI prompts.",
+        schema={
+            "type": "object",
+            "properties": {"path": {"type": "string", "description": "Path to the PNG file"}},
+            "required": ["path"],
+        },
+        handler=read_png_metadata_handler
+    ),
+    Tool(
+        name="list_thumbnails",
+        description="Generates a thumbnail grid of images in a directory. Returns a numbered grid and a detailed file list.",
+        schema={
+            "type": "object",
+            "properties": {
+                "path": {"type": "string", "description": "Path to the directory containing images"},
+                "page": {"type": "integer", "description": "The page number to retrieve (1-indexed)", "default": 1},
+                "page_size": {"type": "integer", "description": "Number of images per page", "default": 64},
+            },
+            "required": ["path"],
+        },
+        handler=list_thumbnails_handler
+    ),
+]