Compare commits
16
Commits
81fef17af2
..
master
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
627ab3aec6 | ||
|
|
74eaa2a3e2 | ||
|
|
747bf5e02b | ||
|
|
8c18d0d40e | ||
|
|
bad4a24426 | ||
|
|
3e523e9725 | ||
|
|
b107aff150 | ||
|
|
1109a33edb | ||
|
|
e67aebbd3d | ||
|
|
2d30bd2155 | ||
|
|
f6f021f2a3 | ||
|
|
05a2a5010c | ||
|
|
f423764645 | ||
|
|
c0c3a4c405 | ||
|
|
f0e5267828 | ||
|
|
baa48d7cec |
@@ -5,6 +5,11 @@ A Model Context Protocol (MCP) server designed to provide LLMs with efficient, t
|
|||||||
## ⚠️ 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.
|
||||||
|
|
||||||
|
## 🚨 SECURITY WARNING
|
||||||
|
**This server is designed to be run on `localhost` ONLY.**
|
||||||
|
It contains tools (such as `read_image` and `list_directory`) that allow the LLM to read arbitrary files from your filesystem. If you expose this server to the network or a public IP, any user or compromised AI could potentially read sensitive system files (e.g., SSH keys, `/etc/passwd`).
|
||||||
|
**NEVER run this server on a public-facing IP without implementing strict path validation.**
|
||||||
|
|
||||||
## Tools Overview
|
## Tools Overview
|
||||||
|
|
||||||
### 📁 `list_directory`
|
### 📁 `list_directory`
|
||||||
@@ -64,9 +69,10 @@ This server requires Python 3.10+ and the following packages:
|
|||||||
* `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.
|
* `requests`: For communicating with the Stable Diffusion API.
|
||||||
|
* `httpx`: For asynchronous API requests (e.g., Wikipedia).
|
||||||
|
|
||||||
```bash
|
```bash
|
||||||
pip install uvicorn starlette Pillow requests
|
pip install -r requirements.txt
|
||||||
```
|
```
|
||||||
|
|
||||||
### Setup
|
### Setup
|
||||||
@@ -79,21 +85,38 @@ pip install uvicorn starlette Pillow requests
|
|||||||
|
|
||||||
## Configuration (`config.py`)
|
## Configuration (`config.py`)
|
||||||
|
|
||||||
You can tune the server's behavior in `config.py`:
|
Tuning the server's behavior is done via `config.py`.
|
||||||
|
|
||||||
### Server & Logging
|
### 🌐 Server & Logging
|
||||||
- **`HOST` / `PORT`**: The network address and port the server binds to.
|
| Parameter | Description | Default |
|
||||||
- **`LOG_LEVEL`**: Logging verbosity (`DEBUG`, `INFO`, `WARNING`, `ERROR`).
|
| :--- | :--- | :--- |
|
||||||
- **`LOG_FILE`**: Path to the server log file.
|
| `HOST` | The network address the server binds to. | `"127.0.0.1"` |
|
||||||
- **`USER_AGENT`**: The User-Agent string used for API requests (e.g., Wikipedia). Use a browser-like string to avoid 403 Forbidden errors.
|
| `PORT` | The port the server listens on. | `8000` |
|
||||||
|
| `LOG_LEVEL` | Logging verbosity (`DEBUG`, `INFO`, `WARNING`, `ERROR`). | `"WARNING"` |
|
||||||
|
| `LOG_FILE` | Absolute path to the server log file. | `ROOT_DIR / "debug.log"` |
|
||||||
|
| `USER_AGENT` | User-Agent string for API requests (e.g., Wikipedia). | Browser-like string |
|
||||||
|
|
||||||
### Model & Token Tuning
|
### 🎨 Stable Diffusion Integration
|
||||||
- **`PATCH_SIZE`**: Set this to `(clip.vision.patch_size * n_merge)` for your specific model to ensure token-perfect resizing.
|
| Parameter | Description | Default |
|
||||||
- **`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.
|
| `SD_URL` | Base URL of the SD WebUI/Forge instance. | `"http://127.0.0.1:7860"` |
|
||||||
- **`CONTACT_SHEET_THUMB_SIZE`**: The pixel size of thumbnails in the contact sheet.
|
| `MODEL_PRESETS_PATH` | Path to the `model_presets.toml` file. | `ROOT_DIR / "model_presets.toml"` |
|
||||||
|
| `RES_PRESETS_PATH` | Path to the `resolution_presets.toml` file. | `ROOT_DIR / "resolution_presets.toml"` |
|
||||||
|
| `TAG_DATABASE_PATH` | Path to the Danbooru `tags.csv` file. | (Path to extension folder) |
|
||||||
|
| `TAG_SEARCH_LIMIT` | Number of results returned by `search_tags` (direct or similar). | `20` |
|
||||||
|
|
||||||
### Image & Font Settings
|
### 🧠 Model & Token Tuning (Optimized for Gemma 4)
|
||||||
- **`IMAGE_QUALITY`**: Adjust JPEG compression (1-100).
|
| Parameter | Description | Default |
|
||||||
- **`SYSTEM_FONT_NAMES`**: A list of font names Pillow should attempt to find in the system path.
|
| :--- | :--- | :--- |
|
||||||
- **`FALLBACK_FONT_PATHS`**: Absolute paths to `.ttf` files used if no system fonts are found.
|
| `PATCH_SIZE` | Model's vision patch size in pixels. | `48` |
|
||||||
|
| `PREVIEW_TOKEN_BUDGET` | Target token count for `preview_image` thumbnails. | `70` |
|
||||||
|
| `CONTACT_SHEET_COLS` | Number of columns in the contact sheet grid. | `10` |
|
||||||
|
| `CONTACT_SHEET_ROWS` | Number of rows in the contact sheet grid. | `7` |
|
||||||
|
| `CONTACT_SHEET_THUMB_SIZE` | Pixel size of thumbnails in the contact sheet. | `192` |
|
||||||
|
|
||||||
|
### 🖼️ Image & Font Settings
|
||||||
|
| Parameter | Description | Default |
|
||||||
|
| :--- | :--- | :--- |
|
||||||
|
| `IMAGE_QUALITY` | JPEG compression quality (1-100). | `95` |
|
||||||
|
| `SYSTEM_FONT_NAMES` | List of font names for Pillow to try in system paths. | Arial, DejaVu, etc. |
|
||||||
|
| `FALLBACK_FONT_PATHS` | List of absolute paths to `.ttf` files. | Linux-specific paths |
|
||||||
|
|||||||
@@ -1,5 +1,6 @@
|
|||||||
import logging
|
import logging
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
from contextvars import ContextVar
|
||||||
|
|
||||||
# --- Project Root ---
|
# --- Project Root ---
|
||||||
# Get the directory where config.py is located
|
# Get the directory where config.py is located
|
||||||
@@ -8,15 +9,20 @@ 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"
|
request_host = ContextVar("request_host", default=f"{HOST}:{PORT}")
|
||||||
|
LOG_LEVEL = "DEBUG" # Options: "DEBUG", "INFO", "WARNING", "ERROR"
|
||||||
LOG_FILE = str(ROOT_DIR / "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 = "MooseCP/1.0 Local MCP Server (https://long-cat.net/)"
|
||||||
|
|
||||||
# --- Stable Diffusion Config ---
|
# --- Stable Diffusion Config ---
|
||||||
SD_URL = "http://127.0.0.1:7860"
|
SD_URL = "http://127.0.0.1:7860"
|
||||||
MODEL_PRESETS_PATH = str(ROOT_DIR / "model_presets.toml")
|
MODEL_PRESETS_PATH = str(ROOT_DIR / "model_presets.toml")
|
||||||
RES_PRESETS_PATH = str(ROOT_DIR / "resolution_presets.toml")
|
RES_PRESETS_PATH = str(ROOT_DIR / "resolution_presets.toml")
|
||||||
TAG_DATABASE_PATH = "/home/matt/stable-diffusion-webui/extensions/a1111-sd-webui-tagcomplete/tags/danbooru.csv"
|
TAG_DATABASE_PATH = str(ROOT_DIR / "danbooru.csv")
|
||||||
|
TAG_SEARCH_LIMIT = 20
|
||||||
|
ENABLE_TAG_WIKI = False
|
||||||
|
DANBOORU_LOGIN = "" # Your Danbooru username (Optional)
|
||||||
|
DANBOORU_API_KEY = "" # Your Danbooru API key (Optional)
|
||||||
|
|
||||||
# --- 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)
|
||||||
|
|||||||
+140782
File diff suppressed because it is too large
Load Diff
@@ -3,6 +3,7 @@ import logging
|
|||||||
import uvicorn
|
import uvicorn
|
||||||
import signal
|
import signal
|
||||||
import os
|
import os
|
||||||
|
import sys
|
||||||
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, FileResponse
|
from starlette.responses import Response, StreamingResponse, FileResponse
|
||||||
@@ -56,6 +57,10 @@ async def messages_endpoint(request):
|
|||||||
if not sid or sid not in mcp_logic.sessions:
|
if not sid or sid not in mcp_logic.sessions:
|
||||||
return Response("Session not found", status_code=404)
|
return Response("Session not found", status_code=404)
|
||||||
|
|
||||||
|
# Capture the host header to ensure image links work across the network
|
||||||
|
host_header = request.headers.get("host", f"{config.HOST}:{config.PORT}")
|
||||||
|
config.request_host.set(host_header)
|
||||||
|
|
||||||
try:
|
try:
|
||||||
body = await request.json()
|
body = await request.json()
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
@@ -116,7 +121,8 @@ app.add_middleware(
|
|||||||
)
|
)
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
config = uvicorn.Config(
|
# Use a separate variable for uvicorn config to avoid shadowing the config module
|
||||||
|
uvicorn_config = uvicorn.Config(
|
||||||
app=app,
|
app=app,
|
||||||
host=config.HOST,
|
host=config.HOST,
|
||||||
port=config.PORT,
|
port=config.PORT,
|
||||||
@@ -124,7 +130,7 @@ if __name__ == "__main__":
|
|||||||
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(uvicorn_config)
|
||||||
|
|
||||||
def signal_handler(sig, frame):
|
def signal_handler(sig, frame):
|
||||||
logger.info("Shutdown signal received")
|
logger.info("Shutdown signal received")
|
||||||
@@ -140,6 +146,5 @@ if __name__ == "__main__":
|
|||||||
except KeyboardInterrupt:
|
except KeyboardInterrupt:
|
||||||
pass
|
pass
|
||||||
finally:
|
finally:
|
||||||
# Final fallback to ensure the process actually dies
|
logger.info("Server process exiting.")
|
||||||
# if the event loop is still hanging on a connection
|
sys.exit(0)
|
||||||
os._exit(0)
|
|
||||||
|
|||||||
+3
-13
@@ -42,23 +42,13 @@ class MCPServer:
|
|||||||
)
|
)
|
||||||
|
|
||||||
try:
|
try:
|
||||||
result_data = await tool.handler(args)
|
content = await tool.handler(args)
|
||||||
except ToolError as e:
|
except ToolError as e:
|
||||||
# Convert tool errors into a text response so the LLM can understand and potentially fix the input
|
# Convert tool errors into a text response so the LLM can understand and potentially fix the input
|
||||||
result_data = {"text": str(e)}
|
content = [{"type": "text", "text": str(e)}]
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
# Catch-all for unexpected crashes to prevent server death
|
# Catch-all for unexpected crashes to prevent server death
|
||||||
result_data = {"text": f"Unexpected internal error: {str(e)}"}
|
content = [{"type": "text", "text": f"Unexpected internal error: {str(e)}"}]
|
||||||
|
|
||||||
# Handle results that are already lists of content (for multimodal/multi-part responses)
|
|
||||||
if isinstance(result_data, list):
|
|
||||||
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 {"content": content}
|
||||||
|
|
||||||
|
|||||||
+8
-4
@@ -10,9 +10,9 @@ This model is not aesthetic tuned, it must be given explicit tags for everything
|
|||||||
Has very excellent understanding of characters and artists down to extremely niche. Unless prompting an original character, their name is enough to decribe their appearance completely except for clothing.
|
Has very excellent understanding of characters and artists down to extremely niche. Unless prompting an original character, their name is enough to decribe their appearance completely except for clothing.
|
||||||
Accepts a list of comma-separated booru-style tags. Use spaces, not underscores, for tags. **Use the `search_tags` tool to verify your tags**.
|
Accepts a list of comma-separated booru-style tags. Use spaces, not underscores, for tags. **Use the `search_tags` tool to verify your tags**.
|
||||||
Prompts MUST follow this format: <1girl/1boy/1other/solo/couple/(can use multiple)>, <character(s)>, <series>, <artist>, <tags>
|
Prompts MUST follow this format: <1girl/1boy/1other/solo/couple/(can use multiple)>, <character(s)>, <series>, <artist>, <tags>
|
||||||
Every prompt MUST include every one of the above sections.
|
Every prompt MUST include every one of the above sections (the angle brackets are not part of the prompt).
|
||||||
Quality tags such as "masterpiece", "best quality", "very awa", are a LAST RESORT, they override the artist tags. If absolutely required they should be prepended.
|
Quality tags such as "masterpiece", "best quality", "very awa", are a LAST RESORT, they override the artist tags. If absolutely required they should be prepended.
|
||||||
It understands 'implicit' artist tags such as "official art" or "game cg" for the 'artist' immediately following the series.
|
It understands every artist, so pick one appropriate for the image. If desired, it also knows 'implicit' artist tags such as "official art" or "game cg" for the 'artist' immediately following the series name.
|
||||||
Natural language understanding is very limited, but can do things like "dark blue skirt" or natural language order of tags such as "lying, on bed".
|
Natural language understanding is very limited, but can do things like "dark blue skirt" or natural language order of tags such as "lying, on bed".
|
||||||
Do not use a negative prompt unless explicitly required to exclude something, your first prompt should have a blank negative prompt.
|
Do not use a negative prompt unless explicitly required to exclude something, your first prompt should have a blank negative prompt.
|
||||||
Has the SDXL problem with hands, works best if hand posture is explicitly prompted.
|
Has the SDXL problem with hands, works best if hand posture is explicitly prompted.
|
||||||
@@ -35,10 +35,14 @@ preset = "xl"
|
|||||||
"Rescale CFG" = 0.3
|
"Rescale CFG" = 0.3
|
||||||
|
|
||||||
["Z-Image-Turbo"]
|
["Z-Image-Turbo"]
|
||||||
description = "General image generation model. Aesthetic tuned, gets good results first try. NSFW is quite limited."
|
description = "General image generation model. Aesthetic tuned, gets good results first try. Can do softcore NSFW, e.g. underwear, breasts, asses. No full-frontal nudity."
|
||||||
guide = """
|
guide = """
|
||||||
This model is aesthetic tuned, regenerating with the same prompt will yield essentially the same image. Change the prompt before resubmitting.
|
This model is aesthetic tuned, regenerating with the same prompt will yield essentially the same image. Change the prompt before resubmitting.
|
||||||
Understands natural language very well. Characters can be described by naming them and using this name later in the prompt. Longer, detailed prompts work better.
|
It's a CFG 1.0 turbo model, the negative prompt has no effect.
|
||||||
|
Understands natural language very well, uses Qwen 3 4B as the text encoder. The longer and more detailed prompt the better, take advantage of line breaks and formatting.
|
||||||
|
If any part of the image is left out of the prompt, it will default to generic AI slop which is most NOT what you want. Be very explicit about each character's details. Appearance, ethnicity, age, individual outfit components, facial expression, pose, action, where they are looking, position in the image, etc. should ALL be included in the prompt. Characters can be referenced by naming them and using this name later in the prompt. This also helps the model avoid mixing traits between them.
|
||||||
|
This also applies to the image itself. Composition, framing, lighting, image style, setting, background, etc. should all be explicitly specified in the prompt.
|
||||||
|
Has some idiosyncrasies so you may need to iterate the prompt a few times to get around some weird artifacts. Think things like "green eyes" making them glow green, or "blush" making the entire face glow. Also really wants to make shirts tucked in for some reason. Examine the generated image closely and edit the prompt if needed.
|
||||||
"""
|
"""
|
||||||
preset = "zit"
|
preset = "zit"
|
||||||
filename = "z_image_turbo_bf16.safetensors"
|
filename = "z_image_turbo_bf16.safetensors"
|
||||||
|
|||||||
@@ -0,0 +1,5 @@
|
|||||||
|
uvicorn
|
||||||
|
starlette
|
||||||
|
requests
|
||||||
|
httpx
|
||||||
|
Pillow
|
||||||
+3
-3
@@ -22,7 +22,7 @@ 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 .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 .browse_wikipedia import handle as wikipedia_handler
|
||||||
from .get_model_info import handle as get_model_info_handler
|
from .get_model_info import handle as get_model_info_handler
|
||||||
from .generate_image import handle as generate_image_handler
|
from .generate_image import handle as generate_image_handler
|
||||||
from .search_tags import handle as search_tags_handler
|
from .search_tags import handle as search_tags_handler
|
||||||
@@ -148,7 +148,7 @@ TOOL_REGISTRY = [
|
|||||||
"model_name": {"type": "string", "description": "The name of the model to use. Required."},
|
"model_name": {"type": "string", "description": "The name of the model to use. Required."},
|
||||||
"prompt": {"type": "string", "description": "The prompt for the image. Required."},
|
"prompt": {"type": "string", "description": "The prompt for the image. Required."},
|
||||||
"negative_prompt": {"type": "string", "description": "The negative prompt to exclude unwanted elements."},
|
"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."},
|
"resolution_preset": {"type": "string", "description": "A named resolution preset from the model info"},
|
||||||
"cfg_scale": {"type": "number", "description": "CFG scale for prompt adherence. Usually should be left omitted to select the default."},
|
"cfg_scale": {"type": "number", "description": "CFG scale for prompt adherence. Usually should be left omitted to select the default."},
|
||||||
},
|
},
|
||||||
"required": ["model_name", "prompt", "resolution_preset"],
|
"required": ["model_name", "prompt", "resolution_preset"],
|
||||||
@@ -157,7 +157,7 @@ TOOL_REGISTRY = [
|
|||||||
),
|
),
|
||||||
Tool(
|
Tool(
|
||||||
name="search_tags",
|
name="search_tags",
|
||||||
description="Searches the Danbooru tag database for tags matching a query. Returns the most popular tags including alias matches. Useful for finding the correct booru-style tags for anime models.",
|
description="Searches the Danbooru tag database for tags matching a query. Returns the Danbooru wiki page on an exact match. On any query, returns a list of similar tags as well as aliases. Useful for finding the correct booru-style tags for anime models or getting more information.",
|
||||||
schema={
|
schema={
|
||||||
"type": "object",
|
"type": "object",
|
||||||
"properties": {
|
"properties": {
|
||||||
|
|||||||
@@ -1,9 +1,8 @@
|
|||||||
import urllib.request
|
import httpx
|
||||||
import urllib.parse
|
|
||||||
import json
|
import json
|
||||||
import re
|
import re
|
||||||
import asyncio
|
import asyncio
|
||||||
from typing import Any, Dict, Optional
|
from typing import Any, Dict, List, Optional
|
||||||
from .utils import ToolError
|
from .utils import ToolError
|
||||||
import config
|
import config
|
||||||
|
|
||||||
@@ -13,22 +12,33 @@ def strip_html(text: str) -> str:
|
|||||||
"""Removes HTML tags from a string using regex to provide clean text to the LLM."""
|
"""Removes HTML tags from a string using regex to provide clean text to the LLM."""
|
||||||
return re.sub(r'<[^>]*>', '', text)
|
return re.sub(r'<[^>]*>', '', text)
|
||||||
|
|
||||||
def _make_request(params: Dict[str, Any]) -> Dict[str, Any]:
|
def clean_wikitext(text: str) -> str:
|
||||||
"""
|
"""
|
||||||
Synchronous helper to make the API request using urllib.
|
Removes the most distracting elements of raw Wikitext:
|
||||||
urllib is used instead of httpx to avoid TLS/HTTP fingerprinting
|
1. HTML comments (<!-- ... -->)
|
||||||
that triggers 403 Forbidden responses from Wikipedia.
|
2. Citations (<ref /> and <ref>...</ref>)
|
||||||
"""
|
"""
|
||||||
query_string = urllib.parse.urlencode(params)
|
# Remove HTML comments
|
||||||
url = f"{API_URL}?{query_string}"
|
text = re.sub(r'<!--.*?-->', '', text, flags=re.DOTALL)
|
||||||
|
# Remove self-closing citations FIRST to prevent them being seen as opening tags
|
||||||
|
text = re.sub(r'<ref[^>]*/>', '', text)
|
||||||
|
# Remove paired citations
|
||||||
|
text = re.sub(r'<ref[^>]*>.*?</ref>', '', text, flags=re.DOTALL)
|
||||||
|
return text
|
||||||
|
|
||||||
|
async def _make_request(params: Dict[str, Any]) -> Dict[str, Any]:
|
||||||
|
"""
|
||||||
|
Asynchronous helper to make the API request using httpx.
|
||||||
|
A custom User-Agent is used to avoid bot detection.
|
||||||
|
"""
|
||||||
headers = {
|
headers = {
|
||||||
"User-Agent": config.USER_AGENT
|
"User-Agent": config.USER_AGENT
|
||||||
}
|
}
|
||||||
|
|
||||||
req = urllib.request.Request(url, headers=headers)
|
async with httpx.AsyncClient(headers=headers, timeout=15.0) as client:
|
||||||
with urllib.request.urlopen(req) as response:
|
response = await client.get(API_URL, params=params)
|
||||||
return json.loads(response.read().decode('utf-8'))
|
response.raise_for_status()
|
||||||
|
return response.json()
|
||||||
|
|
||||||
async def _search(title: str, limit: int = 5, fallback: bool = False) -> str:
|
async def _search(title: str, limit: int = 5, fallback: bool = False) -> str:
|
||||||
"""
|
"""
|
||||||
@@ -42,8 +52,8 @@ async def _search(title: str, limit: int = 5, fallback: bool = False) -> str:
|
|||||||
"format": "json",
|
"format": "json",
|
||||||
"srlimit": limit
|
"srlimit": limit
|
||||||
}
|
}
|
||||||
# Run synchronous urllib call in a thread to avoid blocking the event loop
|
# Directly await the async request
|
||||||
data = await asyncio.to_thread(_make_request, params)
|
data = await _make_request(params)
|
||||||
|
|
||||||
search_results = data.get("query", {}).get("search", [])
|
search_results = data.get("query", {}).get("search", [])
|
||||||
if not search_results:
|
if not search_results:
|
||||||
@@ -66,10 +76,11 @@ async def _search(title: str, limit: int = 5, fallback: bool = False) -> str:
|
|||||||
|
|
||||||
return "\n".join(lines)
|
return "\n".join(lines)
|
||||||
|
|
||||||
async def _fetch_content(title: str, section_index: int) -> Optional[str]:
|
async def _fetch_content(title: str, section_index: int) -> tuple[Optional[str], Optional[str]]:
|
||||||
"""
|
"""
|
||||||
Retrieves raw Wikitext for a specific section.
|
Retrieves raw Wikitext for a specific section.
|
||||||
section_index=0 returns the lead section.
|
section_index=0 returns the lead section.
|
||||||
|
Returns a tuple of (content, final_title).
|
||||||
"""
|
"""
|
||||||
params = {
|
params = {
|
||||||
"action": "query",
|
"action": "query",
|
||||||
@@ -80,27 +91,30 @@ async def _fetch_content(title: str, section_index: int) -> Optional[str]:
|
|||||||
"redirects": 1,
|
"redirects": 1,
|
||||||
"format": "json"
|
"format": "json"
|
||||||
}
|
}
|
||||||
data = await asyncio.to_thread(_make_request, params)
|
data = await _make_request(params)
|
||||||
|
|
||||||
pages = data.get("query", {}).get("pages", {})
|
pages = data.get("query", {}).get("pages", {})
|
||||||
if not pages:
|
if not pages:
|
||||||
return None
|
return None, None
|
||||||
|
|
||||||
page_id = next(iter(pages))
|
page_id = next(iter(pages))
|
||||||
page = pages[page_id]
|
page = pages[page_id]
|
||||||
|
final_title = page.get("title")
|
||||||
|
|
||||||
if "missing" in page:
|
if "missing" in page:
|
||||||
return None
|
return None, None
|
||||||
|
|
||||||
revisions = page.get("revisions", [])
|
revisions = page.get("revisions", [])
|
||||||
if not revisions:
|
if not revisions:
|
||||||
return None
|
return None, final_title
|
||||||
|
|
||||||
return revisions[0].get("*")
|
content = revisions[0].get("*")
|
||||||
|
return (clean_wikitext(content) if content else None), final_title
|
||||||
|
|
||||||
async def _fetch_toc(title: str) -> Optional[str]:
|
async def _fetch_toc(title: str) -> tuple[Optional[str], Optional[str]]:
|
||||||
"""
|
"""
|
||||||
Retrieves the Table of Contents data and formats it hierarchically.
|
Retrieves the Table of Contents data and formats it hierarchically.
|
||||||
|
Returns a tuple of (toc, final_title).
|
||||||
"""
|
"""
|
||||||
params = {
|
params = {
|
||||||
"action": "parse",
|
"action": "parse",
|
||||||
@@ -109,18 +123,19 @@ async def _fetch_toc(title: str) -> Optional[str]:
|
|||||||
"format": "json",
|
"format": "json",
|
||||||
"redirects": 1
|
"redirects": 1
|
||||||
}
|
}
|
||||||
data = await asyncio.to_thread(_make_request, params)
|
data = await _make_request(params)
|
||||||
|
|
||||||
parse_data = data.get("parse")
|
parse_data = data.get("parse")
|
||||||
if not parse_data:
|
if not parse_data:
|
||||||
return None
|
return None, None
|
||||||
|
|
||||||
|
final_title = parse_data.get("title")
|
||||||
toc_data = parse_data.get("tocdata", {})
|
toc_data = parse_data.get("tocdata", {})
|
||||||
sections = toc_data.get("sections", [])
|
sections = toc_data.get("sections", [])
|
||||||
if not sections:
|
if not sections:
|
||||||
return "No table of contents found for this page."
|
return "No table of contents found for this page.", final_title
|
||||||
|
|
||||||
lines = [f"Table of Contents for \"{parse_data.get('title', title)}\":"]
|
lines = [f"Table of Contents for \"{final_title}\":"]
|
||||||
for s in sections:
|
for s in sections:
|
||||||
level = s.get("tocLevel", 1)
|
level = s.get("tocLevel", 1)
|
||||||
index = s.get("index")
|
index = s.get("index")
|
||||||
@@ -128,9 +143,9 @@ async def _fetch_toc(title: str) -> Optional[str]:
|
|||||||
indent = " " * (level - 1)
|
indent = " " * (level - 1)
|
||||||
lines.append(f"{indent}[{index}] {line}")
|
lines.append(f"{indent}[{index}] {line}")
|
||||||
|
|
||||||
return "\n".join(lines)
|
return "\n".join(lines), final_title
|
||||||
|
|
||||||
async def handle(args: Dict[str, Any]) -> Dict[str, Any]:
|
async def handle(args: Dict[str, Any]) -> List[Dict[str, Any]]:
|
||||||
"""
|
"""
|
||||||
Main handler for the browse_wikipedia tool.
|
Main handler for the browse_wikipedia tool.
|
||||||
Supports modes: summary, toc, section, and search.
|
Supports modes: summary, toc, section, and search.
|
||||||
@@ -143,18 +158,24 @@ async def handle(args: Dict[str, Any]) -> Dict[str, Any]:
|
|||||||
section_index = args.get("section_index")
|
section_index = args.get("section_index")
|
||||||
search_limit = args.get("search_limit", 5)
|
search_limit = args.get("search_limit", 5)
|
||||||
|
|
||||||
|
final_title = None
|
||||||
|
|
||||||
if mode == "search":
|
if mode == "search":
|
||||||
result = await _search(title, search_limit, fallback=False)
|
result = await _search(title, search_limit, fallback=False)
|
||||||
elif mode == "summary":
|
elif mode == "summary":
|
||||||
# Lead section + ToC is the default 'summary' to guide the AI's next steps
|
# Lead section + ToC is the default 'summary' to guide the AI's next steps
|
||||||
content = await _fetch_content(title, 0)
|
content, title_from_content = await _fetch_content(title, 0)
|
||||||
|
final_title = title_from_content
|
||||||
if content is None:
|
if content is None:
|
||||||
result = await _search(title, search_limit, fallback=True)
|
result = await _search(title, search_limit, fallback=True)
|
||||||
else:
|
else:
|
||||||
toc = await _fetch_toc(title)
|
toc, title_from_toc = await _fetch_toc(title)
|
||||||
|
if title_from_toc:
|
||||||
|
final_title = title_from_toc
|
||||||
result = f"{content}\n\n---\n\n{toc}"
|
result = f"{content}\n\n---\n\n{toc}"
|
||||||
elif mode == "toc":
|
elif mode == "toc":
|
||||||
toc = await _fetch_toc(title)
|
toc, title_from_toc = await _fetch_toc(title)
|
||||||
|
final_title = title_from_toc
|
||||||
if toc is None:
|
if toc is None:
|
||||||
result = await _search(title, search_limit, fallback=True)
|
result = await _search(title, search_limit, fallback=True)
|
||||||
else:
|
else:
|
||||||
@@ -162,7 +183,8 @@ async def handle(args: Dict[str, Any]) -> Dict[str, Any]:
|
|||||||
elif mode == "section":
|
elif mode == "section":
|
||||||
if section_index is None:
|
if section_index is None:
|
||||||
raise ToolError("Missing required parameter 'section_index' for mode='section'")
|
raise ToolError("Missing required parameter 'section_index' for mode='section'")
|
||||||
content = await _fetch_content(title, int(section_index))
|
content, title_from_content = await _fetch_content(title, int(section_index))
|
||||||
|
final_title = title_from_content
|
||||||
if content is None:
|
if content is None:
|
||||||
result = await _search(title, search_limit, fallback=True)
|
result = await _search(title, search_limit, fallback=True)
|
||||||
else:
|
else:
|
||||||
@@ -170,4 +192,7 @@ async def handle(args: Dict[str, Any]) -> Dict[str, Any]:
|
|||||||
else:
|
else:
|
||||||
raise ToolError(f"Invalid mode '{mode}'. Supported modes: summary, toc, section, search")
|
raise ToolError(f"Invalid mode '{mode}'. Supported modes: summary, toc, section, search")
|
||||||
|
|
||||||
return {"text": result}
|
if final_title and final_title != title:
|
||||||
|
result = f"Redirected to \"{final_title}\"\n\n{result}"
|
||||||
|
|
||||||
|
return [{"type": "text", "text": result}]
|
||||||
+46
-15
@@ -32,7 +32,7 @@ async def ensure_model_state(model_name: str, model_preset: Dict[str, Any]):
|
|||||||
preset = model_preset.get("preset", "xl")
|
preset = model_preset.get("preset", "xl")
|
||||||
await asyncio.to_thread(
|
await asyncio.to_thread(
|
||||||
requests.post,
|
requests.post,
|
||||||
f"{config.SD_URL}/api/predict/checkpoint_change",
|
f"{config.SD_URL}/api/checkpoint_change",
|
||||||
json={"data": [target_ckpt, preset]},
|
json={"data": [target_ckpt, preset]},
|
||||||
timeout=30
|
timeout=30
|
||||||
)
|
)
|
||||||
@@ -45,7 +45,7 @@ async def ensure_model_state(model_name: str, model_preset: Dict[str, Any]):
|
|||||||
preset = model_preset.get("preset", "xl")
|
preset = model_preset.get("preset", "xl")
|
||||||
await asyncio.to_thread(
|
await asyncio.to_thread(
|
||||||
requests.post,
|
requests.post,
|
||||||
f"{config.SD_URL}/api/predict/modules_change",
|
f"{config.SD_URL}/api/modules_change",
|
||||||
json={"data": [target_modules, preset]},
|
json={"data": [target_modules, preset]},
|
||||||
timeout=30
|
timeout=30
|
||||||
)
|
)
|
||||||
@@ -81,16 +81,47 @@ async def handle(args: Dict[str, Any]) -> List[Dict[str, Any]]:
|
|||||||
except Exception as e:
|
except Exception as e:
|
||||||
raise ToolError(f"Failed to connect to Stable Diffusion server: {str(e)}")
|
raise ToolError(f"Failed to connect to Stable Diffusion server: {str(e)}")
|
||||||
|
|
||||||
# Build the label-to-index map
|
# Build the label-to-index map and a sparse payload
|
||||||
label_map = {}
|
label_map = {}
|
||||||
label_counts = {}
|
label_counts = {}
|
||||||
|
|
||||||
payload = [p["parameter_default"] for p in params_info]
|
# Find the maximum param index to determine payload size
|
||||||
|
max_param_idx = 0
|
||||||
|
for p in params_info:
|
||||||
|
name = p.get("parameter_name", "")
|
||||||
|
if name.startswith("param_"):
|
||||||
|
try:
|
||||||
|
idx = int(name.replace("param_", ""))
|
||||||
|
max_param_idx = max(max_param_idx, idx)
|
||||||
|
except ValueError:
|
||||||
|
pass
|
||||||
|
|
||||||
for idx, p in enumerate(params_info):
|
# Initialize payload with Nones (size is max_idx + 1)
|
||||||
label = p["label"]
|
payload = [None] * (max_param_idx + 1)
|
||||||
|
|
||||||
|
for idx_in_list, p in enumerate(params_info):
|
||||||
|
name = p.get("parameter_name", "")
|
||||||
|
label = p.get("label")
|
||||||
|
default = p.get("parameter_default")
|
||||||
|
|
||||||
|
# Determine the absolute index in the payload
|
||||||
|
if name == "id_task":
|
||||||
|
abs_idx = 0
|
||||||
|
elif name.startswith("param_"):
|
||||||
|
try:
|
||||||
|
abs_idx = int(name.replace("param_", ""))
|
||||||
|
except ValueError:
|
||||||
|
continue
|
||||||
|
else:
|
||||||
|
# Fallback for unexpected names, though unlikely
|
||||||
|
continue
|
||||||
|
|
||||||
|
# Set the default value at the absolute index
|
||||||
|
payload[abs_idx] = default
|
||||||
|
|
||||||
|
# Handle label mapping for AI overrides
|
||||||
if not label or label.startswith("parameter_"):
|
if not label or label.startswith("parameter_"):
|
||||||
label = p["parameter_name"]
|
label = name
|
||||||
|
|
||||||
if label in label_counts:
|
if label in label_counts:
|
||||||
label_counts[label] += 1
|
label_counts[label] += 1
|
||||||
@@ -99,14 +130,14 @@ async def handle(args: Dict[str, Any]) -> List[Dict[str, Any]]:
|
|||||||
label_counts[label] = 0
|
label_counts[label] = 0
|
||||||
mapped_label = label
|
mapped_label = label
|
||||||
|
|
||||||
label_map[mapped_label] = idx
|
label_map[mapped_label] = abs_idx
|
||||||
|
|
||||||
# 3. LOAD CONFIGS
|
# 3. LOAD CONFIGS
|
||||||
models_cfg = load_toml(config.MODEL_PRESETS_PATH)
|
models_cfg = load_toml(config.MODEL_PRESETS_PATH)
|
||||||
res_cfg = load_toml(config.RES_PRESETS_PATH)
|
res_cfg = load_toml(config.RES_PRESETS_PATH)
|
||||||
|
|
||||||
if model_name not in models_cfg:
|
if model_name not in models_cfg:
|
||||||
raise ToolError(f"Model '{model_name}' not found in presets. Available: {', '.join(models_cfg.keys())}")
|
raise ToolError(f"Model '{model_name}' not found in presets. Please call get_model_info to see available models and their correct names.")
|
||||||
|
|
||||||
model_preset = models_cfg[model_name]
|
model_preset = models_cfg[model_name]
|
||||||
|
|
||||||
@@ -143,7 +174,7 @@ async def handle(args: Dict[str, Any]) -> List[Dict[str, Any]]:
|
|||||||
except (ValueError, AttributeError):
|
except (ValueError, AttributeError):
|
||||||
raise ToolError(f"Invalid resolution format for preset '{res_preset_name}': {res_val_str}. Expected 'WidthxHeight'.")
|
raise ToolError(f"Invalid resolution format for preset '{res_preset_name}': {res_val_str}. Expected 'WidthxHeight'.")
|
||||||
else:
|
else:
|
||||||
raise ToolError(f"Resolution preset '{res_preset_name}' not found for this model. Available: {', '.join(res_set.keys())}")
|
raise ToolError(f"Resolution preset '{res_preset_name}' not found for this model. You MUST call get_model_info with this model_name to see the available resolution presets.")
|
||||||
else:
|
else:
|
||||||
raise ToolError(f"No resolution set configured for model '{model_name}'.")
|
raise ToolError(f"No resolution set configured for model '{model_name}'.")
|
||||||
|
|
||||||
@@ -163,8 +194,8 @@ async def handle(args: Dict[str, Any]) -> List[Dict[str, Any]]:
|
|||||||
if "Seed" in label_map:
|
if "Seed" in label_map:
|
||||||
payload[label_map["Seed"]] = -1
|
payload[label_map["Seed"]] = -1
|
||||||
|
|
||||||
for _ in range(7):
|
# The "Magic Number" splice is no longer needed as gaps are
|
||||||
payload.insert(39, None)
|
# automatically filled by the sparse-to-dense mapping.
|
||||||
|
|
||||||
# 6. EXECUTE GENERATION
|
# 6. EXECUTE GENERATION
|
||||||
try:
|
try:
|
||||||
@@ -187,9 +218,9 @@ async def handle(args: Dict[str, Any]) -> List[Dict[str, Any]]:
|
|||||||
raise ToolError(f"Could not extract file path from SD URL: {sd_img_url}")
|
raise ToolError(f"Could not extract file path from SD URL: {sd_img_url}")
|
||||||
|
|
||||||
# Construct proxy URL through our MCP server
|
# 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,
|
# Use the captured request host to ensure links work across the network
|
||||||
# but the current endpoint handles it.
|
host = config.request_host.get()
|
||||||
proxy_url = f"http://localhost:{config.PORT}/file{raw_path}"
|
proxy_url = f"http://{host}/file{raw_path}"
|
||||||
|
|
||||||
# Download the image and convert to base64 for the AI's vision
|
# Download the image and convert to base64 for the AI's vision
|
||||||
img_resp = await asyncio.to_thread(requests.get, sd_img_url, timeout=30)
|
img_resp = await asyncio.to_thread(requests.get, sd_img_url, timeout=30)
|
||||||
|
|||||||
+15
-6
@@ -3,7 +3,7 @@ from typing import Any, Dict, List
|
|||||||
import config
|
import config
|
||||||
from tools.utils import ToolError, load_toml
|
from tools.utils import ToolError, load_toml
|
||||||
|
|
||||||
async def handle(args: Dict[str, Any]) -> Dict[str, Any]:
|
async def handle(args: Dict[str, Any]) -> List[Dict[str, Any]]:
|
||||||
"""
|
"""
|
||||||
Returns information about available models or detailed guides for a specific model.
|
Returns information about available models or detailed guides for a specific model.
|
||||||
"""
|
"""
|
||||||
@@ -20,10 +20,13 @@ async def handle(args: Dict[str, Any]) -> Dict[str, Any]:
|
|||||||
"name": name,
|
"name": name,
|
||||||
"description": data.get("description", "No description provided.")
|
"description": data.get("description", "No description provided.")
|
||||||
})
|
})
|
||||||
return {
|
return [
|
||||||
|
{
|
||||||
|
"type": "text",
|
||||||
"text": "Available models:\n\n" + "\n".join([f"- {m['name']}: {m['description']}" for m in catalog]) +
|
"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."
|
"\n\nBefore generating an image, you MUST call this tool again with your selected model as the 'model_name' argument."
|
||||||
}
|
}
|
||||||
|
]
|
||||||
|
|
||||||
if model_name not in models:
|
if model_name not in models:
|
||||||
# Return the full catalog if the specific model isn't found
|
# Return the full catalog if the specific model isn't found
|
||||||
@@ -33,11 +36,14 @@ async def handle(args: Dict[str, Any]) -> Dict[str, Any]:
|
|||||||
"name": name,
|
"name": name,
|
||||||
"description": data.get("description", "No description provided.")
|
"description": data.get("description", "No description provided.")
|
||||||
})
|
})
|
||||||
return {
|
return [
|
||||||
|
{
|
||||||
|
"type": "text",
|
||||||
"text": f"Model '{model_name}' not found.\n\nAvailable models:\n\n" +
|
"text": f"Model '{model_name}' not found.\n\nAvailable models:\n\n" +
|
||||||
"\n".join([f"- {m['name']}: {m['description']}" for m in catalog]) +
|
"\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."
|
"\n\nBefore generating an image, you MUST call this tool again with selected model as the 'model_name' argument."
|
||||||
}
|
}
|
||||||
|
]
|
||||||
|
|
||||||
model_data = models[model_name]
|
model_data = models[model_name]
|
||||||
res_set_name = model_data.get("Resolution Set")
|
res_set_name = model_data.get("Resolution Set")
|
||||||
@@ -58,7 +64,9 @@ async def handle(args: Dict[str, Any]) -> Dict[str, Any]:
|
|||||||
preset_text = "\n".join([f"- {k}: {v}" for k, v in presets_to_show.items()])
|
preset_text = "\n".join([f"- {k}: {v}" for k, v in presets_to_show.items()])
|
||||||
res_text = ", ".join(res_options)
|
res_text = ", ".join(res_options)
|
||||||
|
|
||||||
return {
|
return [
|
||||||
|
{
|
||||||
|
"type": "text",
|
||||||
"text": (
|
"text": (
|
||||||
f"Model: {model_name}\n"
|
f"Model: {model_name}\n"
|
||||||
f"Description: {description}\n\n"
|
f"Description: {description}\n\n"
|
||||||
@@ -68,3 +76,4 @@ async def handle(args: Dict[str, Any]) -> Dict[str, Any]:
|
|||||||
f"Use 'resolution_preset' in generate_image to choose one of these."
|
f"Use 'resolution_preset' in generate_image to choose one of these."
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
]
|
||||||
|
|||||||
@@ -35,7 +35,7 @@ async def handle(args: Dict[str, Any]):
|
|||||||
# Case 1: No pattern and no lines provided -> Dump whole file
|
# Case 1: No pattern and no lines provided -> Dump whole file
|
||||||
if not pattern and not lines_str:
|
if not pattern and not lines_str:
|
||||||
output = [f"{i+1}: {line}" for i, line in enumerate(all_lines)]
|
output = [f"{i+1}: {line}" for i, line in enumerate(all_lines)]
|
||||||
return {"text": "".join(output)}
|
return [{"type": "text", "text": "".join(output)}]
|
||||||
|
|
||||||
# Determine target line numbers (1-based)
|
# Determine target line numbers (1-based)
|
||||||
target_lines: Set[int] = set()
|
target_lines: Set[int] = set()
|
||||||
@@ -61,7 +61,7 @@ async def handle(args: Dict[str, Any]):
|
|||||||
raise ToolError(f"Error parsing line indices: {e}")
|
raise ToolError(f"Error parsing line indices: {e}")
|
||||||
|
|
||||||
if not target_lines:
|
if not target_lines:
|
||||||
return {"text": "No matching lines found."}
|
return [{"type": "text", "text": "No matching lines found."}]
|
||||||
|
|
||||||
# Expand targets with context
|
# Expand targets with context
|
||||||
lines_to_show: Set[int] = set()
|
lines_to_show: Set[int] = set()
|
||||||
@@ -83,4 +83,4 @@ async def handle(args: Dict[str, Any]):
|
|||||||
output.append(f"{ln}: {all_lines[ln-1]}")
|
output.append(f"{ln}: {all_lines[ln-1]}")
|
||||||
last_line = ln
|
last_line = ln
|
||||||
|
|
||||||
return {"text": "".join(output)}
|
return [{"type": "text", "text": "".join(output)}]
|
||||||
|
|||||||
@@ -43,7 +43,7 @@ async def handle(args: Dict[str, Any]):
|
|||||||
paged_items = all_items[start_idx:end_idx]
|
paged_items = all_items[start_idx:end_idx]
|
||||||
|
|
||||||
if not paged_items:
|
if not paged_items:
|
||||||
return {"text": f"No items found on page {page}."}
|
return [{"type": "text", "text": f"No items found on page {page}."}]
|
||||||
|
|
||||||
output = []
|
output = []
|
||||||
for item in paged_items:
|
for item in paged_items:
|
||||||
@@ -61,4 +61,4 @@ async def handle(args: Dict[str, Any]):
|
|||||||
|
|
||||||
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)}."
|
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 [{"type": "text", "text": f"{header}\n\n" + "\n".join(output)}]
|
||||||
|
|||||||
+1
-1
@@ -25,6 +25,6 @@ async def handle(args: Dict[str, Any]):
|
|||||||
|
|
||||||
try:
|
try:
|
||||||
data = base64.b64encode(path.read_bytes()).decode("utf-8")
|
data = base64.b64encode(path.read_bytes()).decode("utf-8")
|
||||||
return {"type": "image", "data": data, "mimeType": mime_type}
|
return [{"type": "image", "data": data, "mimeType": mime_type}]
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
raise ToolError(f"Error reading image: {str(e)}")
|
raise ToolError(f"Error reading image: {str(e)}")
|
||||||
|
|||||||
@@ -26,11 +26,11 @@ async def handle(args: Dict[str, Any]):
|
|||||||
# Pillow parses PNG text chunks into the .info dictionary
|
# Pillow parses PNG text chunks into the .info dictionary
|
||||||
metadata = img.info
|
metadata = img.info
|
||||||
if not metadata:
|
if not metadata:
|
||||||
return {"text": "No metadata found in this PNG."}
|
return [{"type": "text", "text": "No metadata found in this PNG."}]
|
||||||
|
|
||||||
# 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 [{"type": "text", "text": f"PNG Metadata:\n{output}"}]
|
||||||
except ToolError:
|
except ToolError:
|
||||||
raise
|
raise
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
|
|||||||
+109
-39
@@ -2,6 +2,8 @@ import csv
|
|||||||
import os
|
import os
|
||||||
import config
|
import config
|
||||||
import difflib
|
import difflib
|
||||||
|
import httpx
|
||||||
|
import re
|
||||||
from typing import List, Dict, Any, Optional
|
from typing import List, Dict, Any, Optional
|
||||||
from tools.utils import ToolError, format_count, get_type_suffix
|
from tools.utils import ToolError, format_count, get_type_suffix
|
||||||
|
|
||||||
@@ -41,14 +43,47 @@ def _load_tags() -> List[Dict[str, Any]]:
|
|||||||
|
|
||||||
return tags
|
return tags
|
||||||
|
|
||||||
async def handle(args: Dict[str, Any]) -> Dict[str, Any]:
|
return tags
|
||||||
|
|
||||||
|
async def _fetch_wiki_info(tag_name: str) -> Optional[str]:
|
||||||
|
"""Fetches the wiki description for a tag from Danbooru."""
|
||||||
|
if not config.ENABLE_TAG_WIKI:
|
||||||
|
return None
|
||||||
|
|
||||||
|
url = f"https://danbooru.donmai.us/wiki_pages/{tag_name}.json"
|
||||||
|
try:
|
||||||
|
# Use Basic Auth if credentials are provided
|
||||||
|
auth = None
|
||||||
|
if config.DANBOORU_LOGIN and config.DANBOORU_API_KEY:
|
||||||
|
auth = (config.DANBOORU_LOGIN, config.DANBOORU_API_KEY)
|
||||||
|
|
||||||
|
async with httpx.AsyncClient(timeout=5.0) as client:
|
||||||
|
# Using a browser-like UA to avoid potential blocks
|
||||||
|
headers = {"User-Agent": config.USER_AGENT}
|
||||||
|
resp = await client.get(url, headers=headers, auth=auth)
|
||||||
|
|
||||||
|
if resp.status_code == 403:
|
||||||
|
return " (Wiki access blocked by Danbooru/Cloudflare)"
|
||||||
|
|
||||||
|
if resp.status_code == 200:
|
||||||
|
data = resp.json()
|
||||||
|
wiki_body = data.get("wiki_page", {}).get("body")
|
||||||
|
if wiki_body:
|
||||||
|
# Strip HTML tags for the LLM
|
||||||
|
return re.sub(r'<[^>]*>', '', wiki_body).strip()
|
||||||
|
except Exception as e:
|
||||||
|
return f" (Error fetching wiki: {str(e)})"
|
||||||
|
return None
|
||||||
|
|
||||||
|
async def handle(args: Dict[str, Any]) -> List[Dict[str, Any]]:
|
||||||
"""
|
"""
|
||||||
Searches the Danbooru tag database for tags matching a query.
|
Searches the Danbooru tag database for tags matching a query.
|
||||||
Returns the most popular tags including alias matches.
|
Returns a unified list prioritized by substring matches then similarity.
|
||||||
"""
|
"""
|
||||||
global _TAG_CACHE
|
global _TAG_CACHE
|
||||||
|
|
||||||
query = args.get("query", "").lower()
|
# Normalize query: treat spaces and underscores as identical
|
||||||
|
query = args.get("query", "").lower().replace(" ", "_")
|
||||||
if not query:
|
if not query:
|
||||||
raise ToolError("The 'query' argument is required.")
|
raise ToolError("The 'query' argument is required.")
|
||||||
|
|
||||||
@@ -56,55 +91,90 @@ async def handle(args: Dict[str, Any]) -> Dict[str, Any]:
|
|||||||
if _TAG_CACHE is None:
|
if _TAG_CACHE is None:
|
||||||
_TAG_CACHE = _load_tags()
|
_TAG_CACHE = _load_tags()
|
||||||
|
|
||||||
matches = []
|
# 1. Find Substring/Alias Matches (High Priority)
|
||||||
|
substring_matches = []
|
||||||
for tag in _TAG_CACHE:
|
for tag in _TAG_CACHE:
|
||||||
# Check for match in name or aliases
|
|
||||||
is_direct = query in tag["name_lower"]
|
is_direct = query in tag["name_lower"]
|
||||||
is_alias = any(query in alias.strip() for alias in tag["aliases"])
|
is_alias = any(query in alias.strip() for alias in tag["aliases"])
|
||||||
|
|
||||||
if is_direct or is_alias:
|
if is_direct or is_alias:
|
||||||
matches.append({
|
# Find which alias actually matched for reporting
|
||||||
"name": tag["name"],
|
matched_alias = ""
|
||||||
"type": tag["type"],
|
if not is_direct:
|
||||||
"count": tag["count"],
|
matched_alias = next((a.strip() for a in tag["aliases"] if query in a.strip()), query)
|
||||||
|
|
||||||
|
substring_matches.append({
|
||||||
|
"tag": tag,
|
||||||
"matched_via": "name" if is_direct else "alias",
|
"matched_via": "name" if is_direct else "alias",
|
||||||
"alias_match": "" if is_direct else next((a.strip() for a in tag["aliases"] if query in a.strip()), query)
|
"alias_match": matched_alias
|
||||||
})
|
})
|
||||||
|
|
||||||
# Sort by count descending
|
# Sort substring matches by count descending
|
||||||
matches.sort(key=lambda x: x["count"], reverse=True)
|
substring_matches.sort(key=lambda x: x["tag"]["count"], reverse=True)
|
||||||
|
|
||||||
# Format top 20 results
|
# 2. Find Similarity Matches (Low Priority)
|
||||||
results = []
|
|
||||||
for m in matches[:20]:
|
|
||||||
count_fmt = format_count(str(m["count"]))
|
|
||||||
type_sfx = get_type_suffix(m["type"])
|
|
||||||
|
|
||||||
if m["matched_via"] == "alias":
|
|
||||||
line = f"{m['alias_match']} → {m['name']} ({count_fmt}){type_sfx}"
|
|
||||||
else:
|
|
||||||
line = f"{m['name']} ({count_fmt}){type_sfx}"
|
|
||||||
|
|
||||||
results.append(line)
|
|
||||||
|
|
||||||
if not results:
|
|
||||||
# Attempt to find similar tags using difflib
|
|
||||||
all_names_lower = [t["name_lower"] for t in _TAG_CACHE]
|
all_names_lower = [t["name_lower"] for t in _TAG_CACHE]
|
||||||
suggestions_lower = difflib.get_close_matches(query, all_names_lower, n=10, cutoff=0.5)
|
similar_names_lower = difflib.get_close_matches(query, all_names_lower, n=config.TAG_SEARCH_LIMIT, cutoff=0.5)
|
||||||
|
similar_matches = []
|
||||||
if not suggestions_lower:
|
for s_lower in similar_names_lower:
|
||||||
return {"text": f"No tags found matching '{query}'."}
|
|
||||||
|
|
||||||
# Map lowercased suggestions back to original tag objects
|
|
||||||
suggestions = []
|
|
||||||
for s_lower in suggestions_lower:
|
|
||||||
# Find the first tag that matches this lowercased name
|
|
||||||
tag = next((t for t in _TAG_CACHE if t["name_lower"] == s_lower), None)
|
tag = next((t for t in _TAG_CACHE if t["name_lower"] == s_lower), None)
|
||||||
if tag:
|
if tag:
|
||||||
|
similar_matches.append(tag)
|
||||||
|
|
||||||
|
# Build Unified List
|
||||||
|
final_results = []
|
||||||
|
|
||||||
|
# Add substring matches first
|
||||||
|
for m in substring_matches:
|
||||||
|
final_results.append(m)
|
||||||
|
if len(final_results) >= config.TAG_SEARCH_LIMIT:
|
||||||
|
break
|
||||||
|
|
||||||
|
# Fill remaining slots with similar matches
|
||||||
|
if len(final_results) < config.TAG_SEARCH_LIMIT:
|
||||||
|
mentioned_names = {m["tag"]["name_lower"] for m in final_results}
|
||||||
|
for tag in similar_matches:
|
||||||
|
if tag["name_lower"] not in mentioned_names:
|
||||||
|
final_results.append({"tag": tag, "matched_via": "similarity", "alias_match": ""})
|
||||||
|
if len(final_results) >= config.TAG_SEARCH_LIMIT:
|
||||||
|
break
|
||||||
|
|
||||||
|
if not final_results:
|
||||||
|
return [{"type": "text", "text": f"No tags found matching '{query}'."}]
|
||||||
|
|
||||||
|
# Format output lines
|
||||||
|
output_lines = []
|
||||||
|
|
||||||
|
# Special Case: Exact Match Wiki Header
|
||||||
|
# Check if the very first result is an exact match
|
||||||
|
first_res = final_results[0]
|
||||||
|
if first_res["tag"]["name_lower"] == query:
|
||||||
|
exact_tag = first_res["tag"]
|
||||||
|
count_fmt = format_count(str(exact_tag["count"]))
|
||||||
|
type_sfx = get_type_suffix(exact_tag["type"])
|
||||||
|
output_lines.append(f"Exact Match: {exact_tag['name']} ({count_fmt}){type_sfx}")
|
||||||
|
|
||||||
|
wiki_info = await _fetch_wiki_info(exact_tag["name"])
|
||||||
|
if wiki_info:
|
||||||
|
output_lines.append(f"Wiki: {wiki_info}\n")
|
||||||
|
else:
|
||||||
|
output_lines.append("") # spacer
|
||||||
|
|
||||||
|
# List the tags
|
||||||
|
for res in final_results:
|
||||||
|
tag = res["tag"]
|
||||||
count_fmt = format_count(str(tag["count"]))
|
count_fmt = format_count(str(tag["count"]))
|
||||||
type_sfx = get_type_suffix(tag["type"])
|
type_sfx = get_type_suffix(tag["type"])
|
||||||
suggestions.append(f"{tag['name']} ({count_fmt}){type_sfx}")
|
|
||||||
|
|
||||||
return {"text": f"No exact matches for '{query}'. Did you mean:\n" + "\n".join(suggestions)}
|
# If this is the exact match we already listed in the header, skip it
|
||||||
|
if first_res["tag"]["name_lower"] == query and tag["name_lower"] == query:
|
||||||
|
continue
|
||||||
|
|
||||||
return {"text": "Top matches:\n" + "\n".join(results)}
|
if res["matched_via"] == "alias":
|
||||||
|
line = f"- {res['alias_match']} → {tag['name']} ({count_fmt}){type_sfx}"
|
||||||
|
else:
|
||||||
|
line = f"- {tag['name']} ({count_fmt}){type_sfx}"
|
||||||
|
|
||||||
|
output_lines.append(line)
|
||||||
|
|
||||||
|
return [{"type": "text", "text": "\n".join(output_lines)}]
|
||||||
|
|||||||
Reference in New Issue
Block a user