Compare commits
4
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
8beb7fe2a7 | ||
|
|
689cd1244c | ||
|
|
ca33869007 | ||
|
|
f808f4d599 |
@@ -1,6 +1,6 @@
|
|||||||
# MooseCP Image Server
|
# 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
|
## ⚠️ AI SLOP DISCLAIMER
|
||||||
This entire project was vibe-coded by an AI. It is 100% slop code. Use it at your own risk.
|
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.
|
Returns the full-resolution image.
|
||||||
* **Best for**: Final confirmation or deep visual analysis where every pixel counts.
|
* **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`
|
### 🌐 `browse_wikipedia`
|
||||||
Allows the model to browse Wikipedia using its API.
|
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.
|
* **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.
|
* `uvicorn`: ASGI server for the SSE transport.
|
||||||
* `starlette`: Lightweight ASGI framework.
|
* `starlette`: Lightweight ASGI framework.
|
||||||
* `Pillow`: Image processing and thumbnail generation.
|
* `Pillow`: Image processing and thumbnail generation.
|
||||||
|
* `requests`: For communicating with the Stable Diffusion API.
|
||||||
|
|
||||||
```bash
|
```bash
|
||||||
pip install uvicorn starlette Pillow
|
pip install uvicorn starlette Pillow requests
|
||||||
```
|
```
|
||||||
|
|
||||||
### Setup
|
### Setup
|
||||||
|
|||||||
@@ -1,12 +1,22 @@
|
|||||||
import logging
|
import logging
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
# --- Project Root ---
|
||||||
|
# Get the directory where config.py is located
|
||||||
|
ROOT_DIR = Path(__file__).parent.resolve()
|
||||||
|
|
||||||
# --- Server & Logs ---
|
# --- Server & Logs ---
|
||||||
HOST = "127.0.0.1"
|
HOST = "127.0.0.1"
|
||||||
PORT = 8000
|
PORT = 8000
|
||||||
LOG_LEVEL = "WARNING" # Options: "DEBUG", "INFO", "WARNING", "ERROR"
|
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
|
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) ---
|
# --- Model Specific Token Tuning (Tuned for Gemma 4) ---
|
||||||
# Patch size is typically (clip.vision.patch_size * n_merge)
|
# Patch size is typically (clip.vision.patch_size * n_merge)
|
||||||
# For Gemma 4, this is 48px.
|
# For Gemma 4, this is 48px.
|
||||||
|
|||||||
@@ -5,7 +5,7 @@ import signal
|
|||||||
import os
|
import os
|
||||||
from starlette.applications import Starlette
|
from starlette.applications import Starlette
|
||||||
from starlette.routing import Route
|
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 starlette.middleware.cors import CORSMiddleware
|
||||||
|
|
||||||
from mcp_logic import MCPServer
|
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})
|
await mcp_logic.send_to_session(sid, {"jsonrpc": "2.0", "id": req_id, "result": result})
|
||||||
return Response(status_code=202)
|
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(
|
app = Starlette(
|
||||||
routes=[
|
routes=[
|
||||||
Route("/sse", endpoint=sse_endpoint),
|
Route("/sse", endpoint=sse_endpoint),
|
||||||
Route("/messages", endpoint=messages_endpoint, methods=["POST"]),
|
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 .preview_image import handle as preview_image_handler
|
||||||
from .get_text_context import handle as get_text_context_handler
|
from .get_text_context import handle as get_text_context_handler
|
||||||
from .wikipedia import handle as wikipedia_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
|
# Central registry of all available tools
|
||||||
TOOL_REGISTRY = [
|
TOOL_REGISTRY = [
|
||||||
@@ -124,4 +126,32 @@ TOOL_REGISTRY = [
|
|||||||
},
|
},
|
||||||
handler=wikipedia_handler
|
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 time
|
||||||
import datetime
|
import datetime
|
||||||
|
import tomllib
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import List, Dict, Any, Optional
|
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."""
|
"""Custom exception for tool-related errors to be caught by the MCP server."""
|
||||||
pass
|
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:
|
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."""
|
||||||
now = time.time()
|
now = time.time()
|
||||||
|
|||||||
Reference in New Issue
Block a user