Meilisearch MCP Server

from typing import Dict, Any, List, Optional from meilisearch import Client from datetime import datetime def serialize_task_results(obj: Any) -> Any: """Serialize task results into JSON-compatible format""" if hasattr(obj, "__dict__"): return {k: serialize_task_results(v) for k, v in obj.__dict__.items()} elif isinstance(obj, (list, tuple)): return [serialize_task_results(item) for item in obj] elif isinstance(obj, datetime): return obj.isoformat() return obj class TaskManager: def __init__(self, client: Client): """Initialize TaskManager with Meilisearch client""" self.client = client def get_task(self, task_uid: int) -> Dict[str, Any]: """Get information about a specific task""" try: task = self.client.get_task(task_uid) return serialize_task_results(task) except Exception as e: raise Exception(f"Failed to get task: {str(e)}") def get_tasks(self, parameters: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: """Get list of tasks with optional filters""" try: tasks = self.client.get_tasks(parameters) return serialize_task_results(tasks) except Exception as e: raise Exception(f"Failed to get tasks: {str(e)}") def cancel_tasks(self, query_parameters: Dict[str, Any]) -> Dict[str, Any]: """Cancel tasks based on query parameters""" try: result = self.client.cancel_tasks(query_parameters) return serialize_task_results(result) except Exception as e: raise Exception(f"Failed to cancel tasks: {str(e)}") def delete_tasks(self, query_parameters: Dict[str, Any]) -> Dict[str, Any]: """Delete tasks based on query parameters""" try: result = self.client.delete_tasks(query_parameters) return serialize_task_results(result) except Exception as e: raise Exception(f"Failed to delete tasks: {str(e)}")