From f0e526782879fccd63b2378378baf4482d744daf Mon Sep 17 00:00:00 2001 From: moosecrap Date: Mon, 27 Jul 2026 00:35:20 -0700 Subject: [PATCH] Return type standardization --- mcp_logic.py | 16 +++---------- tools/get_model_info.py | 49 +++++++++++++++++++++++---------------- tools/get_text_context.py | 6 ++--- tools/list_directory.py | 4 ++-- tools/read_image.py | 2 +- tools/read_metadata.py | 4 ++-- tools/search_tags.py | 8 +++---- tools/wikipedia.py | 6 ++--- 8 files changed, 47 insertions(+), 48 deletions(-) diff --git a/mcp_logic.py b/mcp_logic.py index 7795499..8aae1e1 100644 --- a/mcp_logic.py +++ b/mcp_logic.py @@ -42,24 +42,14 @@ class MCPServer: ) try: - result_data = await tool.handler(args) + content = await tool.handler(args) except ToolError as e: # Convert tool errors into a text response so the LLM can understand and potentially fix the input - result_data = {"text": str(e)} + content = [{"type": "text", "text": str(e)}] except Exception as e: # Catch-all for unexpected crashes to prevent server death - result_data = {"text": f"Unexpected internal error: {str(e)}"} + 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 Response( diff --git a/tools/get_model_info.py b/tools/get_model_info.py index 6e606c6..7668ce4 100644 --- a/tools/get_model_info.py +++ b/tools/get_model_info.py @@ -3,7 +3,7 @@ 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]: +async def handle(args: Dict[str, Any]) -> List[Dict[str, Any]]: """ 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, "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." - } + return [ + { + "type": "text", + "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: # 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, "description": data.get("description", "No description provided.") }) - return { - "text": f"Model '{model_name}' not found.\n\nAvailable 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." - } + return [ + { + "type": "text", + "text": f"Model '{model_name}' not found.\n\nAvailable 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." + } + ] model_data = models[model_name] res_set_name = model_data.get("Resolution Set") @@ -58,13 +64,16 @@ 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()]) 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." - ) - } + return [ + { + "type": "text", + "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." + ) + } + ] diff --git a/tools/get_text_context.py b/tools/get_text_context.py index 8bb3c44..ae2df49 100644 --- a/tools/get_text_context.py +++ b/tools/get_text_context.py @@ -35,7 +35,7 @@ async def handle(args: Dict[str, Any]): # Case 1: No pattern and no lines provided -> Dump whole file if not pattern and not lines_str: output = [f"{i+1}: {line}" for i, line in enumerate(all_lines)] - return {"text": "".join(output)} + return [{"type": "text", "text": "".join(output)}] # Determine target line numbers (1-based) target_lines: Set[int] = set() @@ -61,7 +61,7 @@ async def handle(args: Dict[str, Any]): raise ToolError(f"Error parsing line indices: {e}") if not target_lines: - return {"text": "No matching lines found."} + return [{"type": "text", "text": "No matching lines found."}] # Expand targets with context 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]}") last_line = ln - return {"text": "".join(output)} + return [{"type": "text", "text": "".join(output)}] diff --git a/tools/list_directory.py b/tools/list_directory.py index 573e33c..bcd3a97 100644 --- a/tools/list_directory.py +++ b/tools/list_directory.py @@ -43,7 +43,7 @@ async def handle(args: Dict[str, Any]): paged_items = all_items[start_idx:end_idx] 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 = [] 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)}." - return {"text": f"{header}\n\n" + "\n".join(output)} + return [{"type": "text", "text": f"{header}\n\n" + "\n".join(output)}] diff --git a/tools/read_image.py b/tools/read_image.py index 7db1f30..afb50c4 100644 --- a/tools/read_image.py +++ b/tools/read_image.py @@ -25,6 +25,6 @@ async def handle(args: Dict[str, Any]): try: 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: raise ToolError(f"Error reading image: {str(e)}") diff --git a/tools/read_metadata.py b/tools/read_metadata.py index 41c5775..29a5fe3 100644 --- a/tools/read_metadata.py +++ b/tools/read_metadata.py @@ -26,11 +26,11 @@ async def handle(args: Dict[str, Any]): # Pillow parses PNG text chunks into the .info dictionary metadata = img.info 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 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: raise except Exception as e: diff --git a/tools/search_tags.py b/tools/search_tags.py index c94f535..0ffb0f3 100644 --- a/tools/search_tags.py +++ b/tools/search_tags.py @@ -41,7 +41,7 @@ def _load_tags() -> List[Dict[str, Any]]: return tags -async def handle(args: Dict[str, Any]) -> Dict[str, Any]: +async def handle(args: Dict[str, Any]) -> List[Dict[str, Any]]: """ Searches the Danbooru tag database for tags matching a query. Returns the most popular tags including alias matches. @@ -93,7 +93,7 @@ async def handle(args: Dict[str, Any]) -> Dict[str, Any]: suggestions_lower = difflib.get_close_matches(query, all_names_lower, n=10, cutoff=0.5) if not suggestions_lower: - return {"text": f"No tags found matching '{query}'."} + return [{"type": "text", "text": f"No tags found matching '{query}'."}] # Map lowercased suggestions back to original tag objects suggestions = [] @@ -105,6 +105,6 @@ async def handle(args: Dict[str, Any]) -> Dict[str, Any]: 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)} + return [{"type": "text", "text": f"No exact matches for '{query}'. Did you mean:\n" + "\n".join(suggestions)}] - return {"text": "Top matches:\n" + "\n".join(results)} + return [{"type": "text", "text": "Top matches:\n" + "\n".join(results)}] diff --git a/tools/wikipedia.py b/tools/wikipedia.py index eedfd5a..0bc79a7 100644 --- a/tools/wikipedia.py +++ b/tools/wikipedia.py @@ -3,7 +3,7 @@ import urllib.parse import json import re import asyncio -from typing import Any, Dict, Optional +from typing import Any, Dict, List, Optional from .utils import ToolError import config @@ -130,7 +130,7 @@ async def _fetch_toc(title: str) -> Optional[str]: return "\n".join(lines) -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. Supports modes: summary, toc, section, and search. @@ -170,4 +170,4 @@ async def handle(args: Dict[str, Any]) -> Dict[str, Any]: else: raise ToolError(f"Invalid mode '{mode}'. Supported modes: summary, toc, section, search") - return {"text": result} + return [{"type": "text", "text": result}]