import httpx
from meilisearch import Client
from typing import Optional, Dict, Any, List
from .indexes import IndexManager
from .documents import DocumentManager
from .tasks import TaskManager
from .settings import SettingsManager
from .keys import KeyManager
from .logging import MCPLogger
from .monitoring import MonitoringManager
from .__version__ import __version__
logger = MCPLogger()
class MeilisearchClient:
def __init__(
self, url: str = "http://localhost:7700", api_key: Optional[str] = None
):
"""Initialize Meilisearch client"""
self.url = url
self.api_key = api_key
# Add custom user agent to identify this as Meilisearch MCP
self.client = Client(
url, api_key, client_agents=("meilisearch-mcp", f"v{__version__}")
)
self.indexes = IndexManager(self.client)
self.documents = DocumentManager(self.client)
self.settings = SettingsManager(self.client)
self.tasks = TaskManager(self.client)
self.keys = KeyManager(self.client)
self.monitoring = MonitoringManager(self.client)
def health_check(self) -> bool:
"""Check if Meilisearch is healthy"""
try:
response = self.client.health()
return response.get("status") == "available"
except Exception:
return False
def get_version(self) -> Dict[str, Any]:
"""Get Meilisearch version information"""
return self.client.get_version()
def get_stats(self) -> Dict[str, Any]:
"""Get database stats"""
return self.client.get_all_stats()
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)}")
def get_indexes(self) -> Dict[str, Any]:
"""Get all indexes"""
indexes = self.client.get_indexes()
# Convert Index objects to serializable dictionaries
serialized_indexes = []
for index in indexes["results"]:
serialized_indexes.append(
{
"uid": index.uid,
"primaryKey": index.primary_key,
"createdAt": index.created_at,
"updatedAt": index.updated_at,
}
)
return {
"results": serialized_indexes,
"offset": indexes["offset"],
"limit": indexes["limit"],
"total": indexes["total"],
}