Compare commits
4
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
8beb7fe2a7 | ||
|
|
689cd1244c | ||
|
|
ca33869007 | ||
|
|
f808f4d599 |
@@ -1,6 +1,6 @@
|
||||
# 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**.
|
||||
A Model Context Protocol (MCP) server designed to provide LLMs with efficient, token-optimized visual access to image directories and AI generation capabilities. 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 entire project was vibe-coded by an AI. It is 100% slop code. Use it at your own risk.
|
||||
@@ -32,6 +32,17 @@ Extracts AI generation parameters from PNG files.
|
||||
Returns the full-resolution image.
|
||||
* **Best for**: Final confirmation or deep visual analysis where every pixel counts.
|
||||
|
||||
### 🎨 `generate_image`
|
||||
Triggers an image generation on a local Stable Diffusion WebUI Forge instance.
|
||||
* **Workflow**: Always call `get_model_info` first to determine the correct prompting style (e.g., tag-based vs. natural language).
|
||||
* **Features**: Supports model-specific presets, resolution presets, and standard parameter overrides.
|
||||
* **Output**: Returns a Base64 image for AI analysis and a proxy URL for direct embedding in the chat.
|
||||
|
||||
### ℹ️ `get_model_info`
|
||||
Provides the "manual" for available generation models.
|
||||
* **Best for**: Learning the prompting style, recommended settings, and available resolution presets for a specific model.
|
||||
* **Workflow**: Call without arguments to see the catalog; call with `model_name` for the detailed guide.
|
||||
|
||||
### 🌐 `browse_wikipedia`
|
||||
Allows the model to browse Wikipedia using its API.
|
||||
* **Best for**: Quickly retrieving summaries, structural maps (ToC), or specific section content from Wikipedia without dumping the entire page.
|
||||
@@ -47,9 +58,10 @@ 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.
|
||||
* `requests`: For communicating with the Stable Diffusion API.
|
||||
|
||||
```bash
|
||||
pip install uvicorn starlette Pillow
|
||||
pip install uvicorn starlette Pillow requests
|
||||
```
|
||||
|
||||
### Setup
|
||||
|
||||
@@ -1,12 +1,22 @@
|
||||
import logging
|
||||
from pathlib import Path
|
||||
|
||||
# --- Project Root ---
|
||||
# Get the directory where config.py is located
|
||||
ROOT_DIR = Path(__file__).parent.resolve()
|
||||
|
||||
# --- Server & Logs ---
|
||||
HOST = "127.0.0.1"
|
||||
PORT = 8000
|
||||
LOG_LEVEL = "WARNING" # Options: "DEBUG", "INFO", "WARNING", "ERROR"
|
||||
LOG_FILE = "debug.log"
|
||||
LOG_FILE = str(ROOT_DIR / "debug.log")
|
||||
USER_AGENT = "Mozilla/5.0 (X11; Linux x86_64; rv:151.0) Gecko/20100101 Firefox/151.0" # Stealth User-Agent to bypass Wikipedia's bot detection
|
||||
|
||||
# --- Stable Diffusion Config ---
|
||||
SD_URL = "http://127.0.0.1:7860"
|
||||
MODEL_PRESETS_PATH = str(ROOT_DIR / "model_presets.toml")
|
||||
RES_PRESETS_PATH = str(ROOT_DIR / "resolution_presets.toml")
|
||||
|
||||
# --- Model Specific Token Tuning (Tuned for Gemma 4) ---
|
||||
# Patch size is typically (clip.vision.patch_size * n_merge)
|
||||
# For Gemma 4, this is 48px.
|
||||
|
||||
@@ -5,7 +5,7 @@ import signal
|
||||
import os
|
||||
from starlette.applications import Starlette
|
||||
from starlette.routing import Route
|
||||
from starlette.responses import Response, StreamingResponse
|
||||
from starlette.responses import Response, StreamingResponse, FileResponse
|
||||
from starlette.middleware.cors import CORSMiddleware
|
||||
|
||||
from mcp_logic import MCPServer
|
||||
@@ -75,10 +75,35 @@ async def messages_endpoint(request):
|
||||
await mcp_logic.send_to_session(sid, {"jsonrpc": "2.0", "id": req_id, "result": result})
|
||||
return Response(status_code=202)
|
||||
|
||||
async def file_endpoint(request):
|
||||
"""
|
||||
Proxy endpoint to serve files from disk with CORP/CORS headers to bypass browser restrictions.
|
||||
"""
|
||||
path = request.path_params.get("path")
|
||||
if not path:
|
||||
return Response("Path not provided", status_code=400)
|
||||
|
||||
# Ensure the path is absolute (SDFiles are usually in /tmp/gradio/...)
|
||||
# If the path doesn't start with /, we assume it's relative to root for simplicity in this context
|
||||
full_path = path if path.startswith("/") else f"/{path}"
|
||||
|
||||
if not os.path.exists(full_path):
|
||||
return Response(f"File not found: {full_path}", status_code=404)
|
||||
|
||||
return FileResponse(
|
||||
full_path,
|
||||
headers={
|
||||
"Cross-Origin-Resource-Policy": "cross-origin",
|
||||
"Access-Control-Allow-Origin": "*",
|
||||
"Access-Control-Allow-Methods": "GET"
|
||||
}
|
||||
)
|
||||
|
||||
app = Starlette(
|
||||
routes=[
|
||||
Route("/sse", endpoint=sse_endpoint),
|
||||
Route("/messages", endpoint=messages_endpoint, methods=["POST"]),
|
||||
Route("/file/{path:path}", endpoint=file_endpoint),
|
||||
]
|
||||
)
|
||||
|
||||
|
||||
@@ -0,0 +1,27 @@
|
||||
# Model Presets
|
||||
# Format: ["Model Name"]
|
||||
# Description and Guide are mandatory.
|
||||
# Other keys should be the Labels found in the /info endpoint.
|
||||
|
||||
["noobaiXLNAIXL_vPred10Version"]
|
||||
description = "High-quality anime model, requires tag-based prompting."
|
||||
guide = """
|
||||
Detailed Prompting Guide:
|
||||
- Style: Tag-based (Danbooru). Use comma-separated tags rather than natural language.
|
||||
- Quality: Use 'masterpiece, best quality' for high fidelity.
|
||||
- Negatives: Use 'lowres, bad anatomy, bad hands' or a specific inverse style.
|
||||
- Settings: Keep CFG between 5.0 and 7.0; higher values can cause color burn.
|
||||
- Sampler: Works well with DPM++ 2M SDE.
|
||||
"""
|
||||
"Resolution Set" = "sdxl"
|
||||
"CFG Scale" = 4.0
|
||||
"Hires. fix" = true
|
||||
"Denoising strength" = 0.5
|
||||
"Upscale by" = 1.5
|
||||
"Upscaler" = "Lanczos"
|
||||
"Hires steps" = 16
|
||||
"Hires CFG Scale" = 4.0
|
||||
"Sampling Steps" = 32
|
||||
"Sampling Method" = "Euler a"
|
||||
"Schedule Type" = "Uniform"
|
||||
"Rescale CFG" = 0.3
|
||||
@@ -0,0 +1,9 @@
|
||||
# Resolution Presets
|
||||
# Format: [preset_name]
|
||||
# Values: "WidthxHeight"
|
||||
|
||||
[sdxl]
|
||||
widescreen = "1344x768"
|
||||
landscape = "1152x896"
|
||||
square = "1024x1024"
|
||||
portrait = "896x1152"
|
||||
@@ -23,6 +23,8 @@ 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
|
||||
from .wikipedia import handle as wikipedia_handler
|
||||
from .get_model_info import handle as get_model_info_handler
|
||||
from .generate_image import handle as generate_image_handler
|
||||
|
||||
# Central registry of all available tools
|
||||
TOOL_REGISTRY = [
|
||||
@@ -124,4 +126,32 @@ TOOL_REGISTRY = [
|
||||
},
|
||||
handler=wikipedia_handler
|
||||
),
|
||||
Tool(
|
||||
name="get_model_info",
|
||||
description="Returns a list of available models and their short descriptions. If a specific model name is provided, it returns the comprehensive prompting guide, available resolution presets, and active configuration tips for that model. CRITICAL: You must call this for any model you are unfamiliar with, as prompting styles vary wildly (e.g., tag-based vs. natural language) and using the wrong style will result in poor image quality.",
|
||||
schema={
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"model_name": {"type": "string", "description": "The name of the model to get detailed info for. Leave empty to list all available models."}
|
||||
},
|
||||
"required": [],
|
||||
},
|
||||
handler=get_model_info_handler
|
||||
),
|
||||
Tool(
|
||||
name="generate_image",
|
||||
description="Generates an image using the specified model and parameters. Note: To ensure high quality, verify the model's prompting requirements via get_model_info before calling this tool.",
|
||||
schema={
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"model_name": {"type": "string", "description": "The name of the model to use. Required."},
|
||||
"prompt": {"type": "string", "description": "The prompt for the image. Required."},
|
||||
"negative_prompt": {"type": "string", "description": "The negative prompt to exclude unwanted elements."},
|
||||
"resolution_preset": {"type": "string", "description": "A named resolution preset (e.g., 'square', 'portrait'). Available options depend on the model."},
|
||||
"cfg_scale": {"type": "number", "description": "CFG scale for prompt adherence."},
|
||||
},
|
||||
"required": ["model_name", "prompt"],
|
||||
},
|
||||
handler=generate_image_handler
|
||||
),
|
||||
]
|
||||
|
||||
@@ -0,0 +1,156 @@
|
||||
import requests
|
||||
import base64
|
||||
import config
|
||||
from typing import Any, Dict, List
|
||||
from tools.utils import ToolError, load_toml
|
||||
|
||||
async def handle(args: Dict[str, Any]) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Generates an image using the specified model and parameters.
|
||||
"""
|
||||
# 1. REQUIRED ARGUMENTS
|
||||
prompt = args.get("prompt")
|
||||
if not prompt:
|
||||
raise ToolError("The 'prompt' argument is required.")
|
||||
|
||||
model_name = args.get("model_name")
|
||||
if not model_name:
|
||||
raise ToolError("The 'model_name' argument is required. Use get_model_info to see available models.")
|
||||
|
||||
# 2. FETCH CURRENT SERVER DEFAULTS & MAP LABELS
|
||||
try:
|
||||
info_resp = requests.get(f"{config.SD_URL}/info", timeout=10)
|
||||
info_resp.raise_for_status()
|
||||
info_data = info_resp.json()
|
||||
params_info = info_data["named_endpoints"]["/txt2img"]["parameters"]
|
||||
except Exception as e:
|
||||
raise ToolError(f"Failed to connect to Stable Diffusion server: {str(e)}")
|
||||
|
||||
# Build the label-to-index map
|
||||
label_map = {}
|
||||
label_counts = {}
|
||||
|
||||
payload = [p["parameter_default"] for p in params_info]
|
||||
|
||||
for idx, p in enumerate(params_info):
|
||||
label = p["label"]
|
||||
if not label or label.startswith("parameter_"):
|
||||
label = p["parameter_name"]
|
||||
|
||||
if label in label_counts:
|
||||
label_counts[label] += 1
|
||||
mapped_label = f"{label} [{label_counts[label]}]"
|
||||
else:
|
||||
label_counts[label] = 0
|
||||
mapped_label = label
|
||||
|
||||
label_map[mapped_label] = idx
|
||||
|
||||
# 3. LOAD CONFIGS
|
||||
models_cfg = load_toml(config.MODEL_PRESETS_PATH)
|
||||
res_cfg = load_toml(config.RES_PRESETS_PATH)
|
||||
|
||||
if model_name not in models_cfg:
|
||||
raise ToolError(f"Model '{model_name}' not found in presets. Available: {', '.join(models_cfg.keys())}")
|
||||
|
||||
model_preset = models_cfg[model_name]
|
||||
|
||||
# 4. MERGE PIPELINE
|
||||
for key, value in model_preset.items():
|
||||
if key in ["description", "guide", "Resolution Set"]:
|
||||
continue
|
||||
if key in label_map:
|
||||
payload[label_map[key]] = value
|
||||
elif key.startswith("param_"):
|
||||
try:
|
||||
idx = int(key.replace("param_", ""))
|
||||
if 0 <= idx < len(payload):
|
||||
payload[idx] = value
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
res_preset_name = args.get("resolution_preset")
|
||||
if res_preset_name:
|
||||
res_set_name = model_preset.get("Resolution Set")
|
||||
if res_set_name and res_set_name in res_cfg:
|
||||
res_set = res_cfg[res_set_name]
|
||||
if res_preset_name in res_set:
|
||||
res_val_str = res_set[res_preset_name]
|
||||
try:
|
||||
w, h = map(int, res_val_str.split('x'))
|
||||
if "Width" in label_map:
|
||||
payload[label_map["Width"]] = w
|
||||
if "Height" in label_map:
|
||||
payload[label_map["Height"]] = h
|
||||
except (ValueError, AttributeError):
|
||||
raise ToolError(f"Invalid resolution format for preset '{res_preset_name}': {res_val_str}. Expected 'WidthxHeight'.")
|
||||
else:
|
||||
raise ToolError(f"Resolution preset '{res_preset_name}' not found for this model. Available: {', '.join(res_set.keys())}")
|
||||
else:
|
||||
raise ToolError(f"No resolution set configured for model '{model_name}'.")
|
||||
|
||||
overrides = {
|
||||
"prompt": "Prompt",
|
||||
"negative_prompt": "Negative Prompt",
|
||||
"cfg_scale": "CFG Scale",
|
||||
}
|
||||
|
||||
for arg_key, label in overrides.items():
|
||||
if arg_key in args and label in label_map:
|
||||
payload[label_map[label]] = args[arg_key]
|
||||
|
||||
if "Prompt" in label_map:
|
||||
payload[label_map["Prompt"]] = prompt
|
||||
|
||||
if "Seed" in label_map:
|
||||
payload[label_map["Seed"]] = -1
|
||||
|
||||
for _ in range(7):
|
||||
payload.insert(39, None)
|
||||
|
||||
# 6. EXECUTE GENERATION
|
||||
try:
|
||||
gen_payload = {"data": payload}
|
||||
gen_resp = requests.post(f"{config.SD_URL}/api/txt2img", json=gen_payload, timeout=300)
|
||||
gen_resp.raise_for_status()
|
||||
res_data = gen_resp.json()
|
||||
|
||||
gallery_data = res_data["data"][0]["value"]
|
||||
if not gallery_data:
|
||||
raise ToolError("Server returned successfully but no image was generated.")
|
||||
|
||||
sd_img_url = gallery_data[0]["image"]["url"]
|
||||
|
||||
# Extract raw path from SD URL (e.g. http://.../file=/tmp/gradio/abc.png -> /tmp/gradio/abc.png)
|
||||
if "/file=" in sd_img_url:
|
||||
raw_path = sd_img_url.split("/file=")[1]
|
||||
else:
|
||||
# Fallback if the URL format changes
|
||||
raise ToolError(f"Could not extract file path from SD URL: {sd_img_url}")
|
||||
|
||||
# Construct proxy URL through our MCP server
|
||||
# We strip leading slash from raw_path to avoid double slashes in the proxy URL if we want,
|
||||
# but the current endpoint handles it.
|
||||
proxy_url = f"http://localhost:{config.PORT}/file{raw_path}"
|
||||
|
||||
# Download the image and convert to base64 for the AI's vision
|
||||
img_resp = requests.get(sd_img_url, timeout=30)
|
||||
img_resp.raise_for_status()
|
||||
b64_data = base64.b64encode(img_resp.content).decode('utf-8')
|
||||
|
||||
return [
|
||||
{
|
||||
"type": "text",
|
||||
"text": f"Image successfully generated using model '{model_name}'.\n\nTo show the image to the user, you can include this markdown link in your response: "
|
||||
},
|
||||
{
|
||||
"type": "image",
|
||||
"data": b64_data,
|
||||
"mimeType": "image/png"
|
||||
}
|
||||
]
|
||||
|
||||
except requests.exceptions.HTTPError as e:
|
||||
raise ToolError(f"Server error during generation: {str(e)}")
|
||||
except Exception as e:
|
||||
raise ToolError(f"Unexpected error during generation: {str(e)}")
|
||||
@@ -0,0 +1,59 @@
|
||||
import requests
|
||||
from typing import Any, Dict, List
|
||||
import config
|
||||
from tools.utils import ToolError, load_toml
|
||||
|
||||
async def handle(args: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""
|
||||
Returns information about available models or detailed guides for a specific model.
|
||||
"""
|
||||
model_name = args.get("model_name")
|
||||
|
||||
models = load_toml(config.MODEL_PRESETS_PATH)
|
||||
res_presets = load_toml(config.RES_PRESETS_PATH)
|
||||
|
||||
if not model_name:
|
||||
# Return a catalog of all models
|
||||
catalog = []
|
||||
for name, data in models.items():
|
||||
catalog.append({
|
||||
"name": name,
|
||||
"description": data.get("description", "No description provided.")
|
||||
})
|
||||
return {
|
||||
"text": "Available models:\n\n" + "\n".join([f"- {m['name']}: {m['description']}" for m in catalog]) +
|
||||
"\n\nTo get a detailed prompting guide and available resolutions for a specific model, call this tool again with the 'model_name' argument."
|
||||
}
|
||||
|
||||
if model_name not in models:
|
||||
raise ToolError(f"Model '{model_name}' not found in presets. Available models: {', '.join(models.keys())}")
|
||||
|
||||
model_data = models[model_name]
|
||||
res_set_name = model_data.get("Resolution Set")
|
||||
|
||||
# Get available resolution presets for this model
|
||||
res_options = []
|
||||
if res_set_name and res_set_name in res_presets:
|
||||
res_options = list(res_presets[res_set_name].keys())
|
||||
else:
|
||||
res_options = ["Default (1024x1024)"]
|
||||
|
||||
guide = model_data.get("guide", "No detailed guide available.")
|
||||
description = model_data.get("description", "")
|
||||
|
||||
# Filter out internal config keys to show only the human-friendly presets
|
||||
presets_to_show = {k: v for k, v in model_data.items() if k not in ["description", "guide", "Resolution Set"]}
|
||||
|
||||
preset_text = "\n".join([f"- {k}: {v}" for k, v in presets_to_show.items()])
|
||||
res_text = ", ".join(res_options)
|
||||
|
||||
return {
|
||||
"text": (
|
||||
f"Model: {model_name}\n"
|
||||
f"Description: {description}\n\n"
|
||||
f"--- Prompting Guide ---\n{guide}\n\n"
|
||||
f"--- Active Presets ---\n{preset_text}\n\n"
|
||||
f"--- Available Resolution Presets ---\n{res_text}\n\n"
|
||||
f"Use 'resolution_preset' in generate_image to choose one of these."
|
||||
)
|
||||
}
|
||||
@@ -1,5 +1,6 @@
|
||||
import time
|
||||
import datetime
|
||||
import tomllib
|
||||
from pathlib import Path
|
||||
from typing import List, Dict, Any, Optional
|
||||
|
||||
@@ -8,6 +9,14 @@ class ToolError(Exception):
|
||||
"""Custom exception for tool-related errors to be caught by the MCP server."""
|
||||
pass
|
||||
|
||||
def load_toml(path: str) -> Dict[str, Any]:
|
||||
"""Loads a TOML file into a dictionary."""
|
||||
try:
|
||||
with open(path, "rb") as f:
|
||||
return tomllib.load(f)
|
||||
except Exception as e:
|
||||
raise ToolError(f"Failed to load config file {path}: {str(e)}")
|
||||
|
||||
def format_relative_time(timestamp: float) -> str:
|
||||
"""Converts a timestamp to a human-readable relative format."""
|
||||
now = time.time()
|
||||
|
||||
Reference in New Issue
Block a user