Merge pull request #9 from meilisearch/add-search-tool

Add search functionnality

Author: tpayet

README.md

@@ -9,6 +9,7 @@ A Model Context Protocol (MCP) server for interacting with Meilisearch through L
 - Task monitoring and API key management
 - Built-in logging and monitoring tools
 - Dynamic connection configuration to switch between Meilisearch instances
+- Smart search across single or multiple indices
 
 ## Installation
 
@@ -62,6 +63,43 @@ Example usage through MCP:
 }
 ```
 
+### Search Functionality
+
+The server provides a flexible search tool that can search across one or all indices:
+
+- `search`: Search through Meilisearch indices with optional parameters
+
+Example usage through MCP:
+```json
+// Search in a specific index
+{
+  "name": "search",
+  "arguments": {
+    "query": "search term",
+    "indexUid": "movies",
+    "limit": 10
+  }
+}
+
+// Search across all indices
+{
+  "name": "search",
+  "arguments": {
+    "query": "search term",
+    "limit": 5,
+    "sort": ["releaseDate:desc"]
+  }
+}
+```
+
+Available search parameters:
+- `query`: The search query (required)
+- `indexUid`: Specific index to search in (optional)
+- `limit`: Maximum number of results per index (optional, default: 20)
+- `offset`: Number of results to skip (optional, default: 0)
+- `filter`: Filter expression (optional)
+- `sort`: Sorting rules (optional)
+
 ### Running the Server
 
 ```bash
@@ -82,6 +120,7 @@ npx @modelcontextprotocol/inspector python -m src.meilisearch_mcp
 - API Key Management: create/update/delete API keys
 - Task Monitoring: track and manage asynchronous tasks
 - System Monitoring: health checks and metrics
+- Search: flexible search across single or multiple indices
 
 ## Contributing
 

src/meilisearch_mcp/client.py

@@ -44,6 +44,63 @@ async def get_stats(self) -> Dict[str, Any]:
         """Get database stats"""
         return self.client.get_all_stats()
 
+    async def search(
+        self,
+        query: str,
+        index_uid: Optional[str] = None,
+        limit: Optional[int] = 20,
+        offset: Optional[int] = 0,
+        filter: Optional[str] = None,
+        sort: Optional[List[str]] = None,
+        **kwargs
+    ) -> Dict[str, Any]:
+        """
+        Search through Meilisearch indices.
+        If index_uid is provided, search in that specific index.
+        If not provided, search across all available indices.
+        """
+        try:
+            # Prepare search parameters, removing None values
+            search_params = {
+                "limit": limit if limit is not None else 20,
+                "offset": offset if offset is not None else 0,
+            }
+            
+            if filter is not None:
+                search_params["filter"] = filter
+            if sort is not None:
+                search_params["sort"] = sort
+                
+            # Add any additional parameters
+            search_params.update({k: v for k, v in kwargs.items() if v is not None})
+
+            if index_uid:
+                # Search in specific index
+                index = self.client.index(index_uid)
+                return index.search(query, search_params)
+            else:
+                # Search across all indices
+                results = {}
+                indexes = self.client.get_indexes()
+                
+                for index in indexes["results"]:
+                    try:
+                        search_result = index.search(query, search_params)
+                        if search_result["hits"]:  # Only include indices with matches
+                            results[index.uid] = search_result
+                    except Exception as e:
+                        logger.warning(f"Failed to search index {index.uid}: {str(e)}")
+                        continue
+                
+                return {
+                    "multi_index": True,
+                    "query": query,
+                    "results": results
+                }
+                
+        except Exception as e:
+            raise Exception(f"Search failed: {str(e)}")
+
     async def get_indexes(self) -> Dict[str, Any]:
         """Get all indexes"""
         indexes = self.client.get_indexes()

src/meilisearch_mcp/server.py

@@ -157,6 +157,22 @@ async def handle_list_tools() -> list[types.Tool]:
                         "required": ["indexUid", "settings"],
                     },
                 ),
+                types.Tool(
+                    name="search",
+                    description="Search through Meilisearch indices. If indexUid is not provided, it will search across all indices.",
+                    inputSchema={
+                        "type": "object",
+                        "properties": {
+                            "query": {"type": "string"},
+                            "indexUid": {"type": "string", "optional": True},
+                            "limit": {"type": "integer", "optional": True},
+                            "offset": {"type": "integer", "optional": True},
+                            "filter": {"type": "string", "optional": True},
+                            "sort": {"type": "array", "items": {"type": "string"}, "optional": True},
+                        },
+                        "required": ["query"],
+                    },
+                ),
                 types.Tool(
                     name="get-task",
                     description="Get information about a specific task",
@@ -354,6 +370,25 @@ async def handle_call_tool(
                         )
                     ]
 
+                elif name == "search":
+                    search_results = await self.meili_client.search(
+                        query=arguments["query"],
+                        index_uid=arguments.get("indexUid"),
+                        limit=arguments.get("limit"),
+                        offset=arguments.get("offset"),
+                        filter=arguments.get("filter"),
+                        sort=arguments.get("sort"),
+                    )
+                    
+                    # Format the results for better readability
+                    formatted_results = json.dumps(search_results, indent=2, default=json_serializer)
+                    return [
+                        types.TextContent(
+                            type="text",
+                            text=f"Search results for '{arguments['query']}':\n{formatted_results}"
+                        )
+                    ]
+
                 elif name == "get-task":
                     task = await self.meili_client.tasks.get_task(arguments["taskUid"])
                     return [