"""MCP tool definitions for tool search functionality."""
import logging
from typing import Any
from mcp.server import Server
from mcp.types import Tool, TextContent
from .catalog import catalog, ToolReference
from .search import BM25Search, RegexSearch, EmbeddingsSearch
from .search.regex import RegexSearchError
from .config import config
logger = logging.getLogger(__name__)
# Initialize search engines
bm25_search = BM25Search(catalog)
regex_search = RegexSearch(catalog)
embeddings_search = EmbeddingsSearch(catalog)
# Create MCP server
mcp_server = Server("tool-search-server")
def _create_tool_references(tool_names: list[str]) -> list[dict[str, str]]:
"""Create tool_reference blocks from tool names.
Args:
tool_names: List of tool names to reference
Returns:
List of tool_reference dictionaries
"""
return [
ToolReference(tool_name=name).to_dict()
for name in tool_names
]
@mcp_server.list_tools()
async def list_tools() -> list[Tool]:
"""List available MCP tools."""
return [
Tool(
name="tool_search_bm25",
description=(
"Search for tools using BM25 keyword matching. "
"Uses natural language queries to find tools based on keyword relevance. "
"Best for finding tools when you know specific terms that might appear in "
"the tool name or description."
),
inputSchema={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Natural language search query with keywords"
},
"top_k": {
"type": "integer",
"description": f"Maximum number of results (default: {config.MAX_SEARCH_RESULTS})",
"default": config.MAX_SEARCH_RESULTS,
"minimum": 1,
"maximum": 20,
}
},
"required": ["query"]
}
),
Tool(
name="tool_search_regex",
description=(
"Search for tools using regex pattern matching. "
"Uses Python regex syntax (re.search()) to match against tool names and descriptions. "
"Pattern limit: 200 characters. "
"Common patterns: 'weather' matches tools containing 'weather', "
"'get_.*_data' matches tools like get_user_data or get_weather_data, "
"'(?i)slack' for case-insensitive matching."
),
inputSchema={
"type": "object",
"properties": {
"pattern": {
"type": "string",
"description": "Python regex pattern (re.search() syntax)",
"maxLength": config.REGEX_PATTERN_MAX_LENGTH,
},
"top_k": {
"type": "integer",
"description": f"Maximum number of results (default: {config.MAX_SEARCH_RESULTS})",
"default": config.MAX_SEARCH_RESULTS,
"minimum": 1,
"maximum": 20,
}
},
"required": ["pattern"]
}
),
Tool(
name="tool_search_semantic",
description=(
"Search for tools using semantic similarity. "
"Uses embeddings to understand the meaning of your query and find tools "
"that are conceptually related, even if they don't share exact keywords. "
"Best for natural language queries like 'tools for checking weather' or "
"'something to send messages'."
),
inputSchema={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Natural language description of what you're looking for"
},
"top_k": {
"type": "integer",
"description": f"Maximum number of results (default: {config.MAX_SEARCH_RESULTS})",
"default": config.MAX_SEARCH_RESULTS,
"minimum": 1,
"maximum": 20,
}
},
"required": ["query"]
}
),
Tool(
name="list_all_tools",
description=(
"List all available tools in the catalog. "
"Returns tool names only, not full definitions. "
"Use this to get an overview of available tools."
),
inputSchema={
"type": "object",
"properties": {},
"required": []
}
),
]
@mcp_server.call_tool()
async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]:
"""Handle tool calls."""
if name == "tool_search_bm25":
return await _handle_bm25_search(arguments)
elif name == "tool_search_regex":
return await _handle_regex_search(arguments)
elif name == "tool_search_semantic":
return await _handle_semantic_search(arguments)
elif name == "list_all_tools":
return await _handle_list_all_tools()
else:
return [TextContent(
type="text",
text=f"Unknown tool: {name}"
)]
async def _handle_bm25_search(arguments: dict[str, Any]) -> list[TextContent]:
"""Handle BM25 search tool call."""
query = arguments.get("query", "")
top_k = arguments.get("top_k", config.MAX_SEARCH_RESULTS)
if not query:
return [TextContent(type="text", text="Error: query is required")]
try:
tool_names = bm25_search.search_names(query, top_k)
references = _create_tool_references(tool_names)
if not references:
return [TextContent(
type="text",
text=f"No tools found matching query: {query}"
)]
# Return tool references as JSON
import json
return [TextContent(
type="text",
text=json.dumps({"tool_references": references}, indent=2)
)]
except Exception as e:
logger.error(f"BM25 search error: {e}")
return [TextContent(
type="text",
text=f"Search error: {str(e)}"
)]
async def _handle_regex_search(arguments: dict[str, Any]) -> list[TextContent]:
"""Handle regex search tool call."""
pattern = arguments.get("pattern", "")
top_k = arguments.get("top_k", config.MAX_SEARCH_RESULTS)
if not pattern:
return [TextContent(type="text", text="Error: pattern is required")]
try:
tool_names = regex_search.search_names(pattern, top_k)
references = _create_tool_references(tool_names)
if not references:
return [TextContent(
type="text",
text=f"No tools found matching pattern: {pattern}"
)]
import json
return [TextContent(
type="text",
text=json.dumps({"tool_references": references}, indent=2)
)]
except RegexSearchError as e:
return [TextContent(
type="text",
text=f"Regex error ({e.error_code}): {str(e)}"
)]
except Exception as e:
logger.error(f"Regex search error: {e}")
return [TextContent(
type="text",
text=f"Search error: {str(e)}"
)]
async def _handle_semantic_search(arguments: dict[str, Any]) -> list[TextContent]:
"""Handle semantic search tool call."""
query = arguments.get("query", "")
top_k = arguments.get("top_k", config.MAX_SEARCH_RESULTS)
if not query:
return [TextContent(type="text", text="Error: query is required")]
try:
tool_names = embeddings_search.search_names(query, top_k)
references = _create_tool_references(tool_names)
if not references:
return [TextContent(
type="text",
text=f"No tools found semantically matching: {query}"
)]
import json
return [TextContent(
type="text",
text=json.dumps({"tool_references": references}, indent=2)
)]
except Exception as e:
logger.error(f"Semantic search error: {e}")
return [TextContent(
type="text",
text=f"Search error: {str(e)}"
)]
async def _handle_list_all_tools() -> list[TextContent]:
"""Handle list all tools call."""
tool_names = catalog.get_tool_names()
import json
return [TextContent(
type="text",
text=json.dumps({
"total_tools": len(tool_names),
"tools": tool_names
}, indent=2)
)]
