MAGG

Instructions

Implementation Reference

• magg/server/server.py:786-876 (handler)

  The handler function that implements the core logic of the 'magg_check' tool. It checks the health of all mounted backend servers by attempting to list their tools within a timeout period. Servers that timeout or error are marked unresponsive. Depending on the 'action' parameter, it can report only or take remedial actions like remounting, unmounting, or disabling unresponsive servers.
  async def check( self, action: Annotated[Literal["report", "remount", "unmount", "disable"], Field( description="Action to take for unresponsive servers: 'report' (default), 'remount', 'unmount', or 'disable'" )] = "report", timeout: Annotated[float, Field( description="Timeout in seconds for health check per server" )] = 2.5, ) -> MaggResponse: """Check health of all mounted servers and handle unresponsive ones.""" try: results = {} unresponsive_servers = [] for server_name, server_info in self.server_manager.mounted_servers.items(): client = server_info.get('client') if not client: results[server_name] = {"status": "error", "reason": "No client found"} unresponsive_servers.append(server_name) continue try: async with asyncio.timeout(timeout): async with client: tools = await client.list_tools() results[server_name] = { "status": "healthy", "tools_count": len(tools) } except asyncio.TimeoutError: results[server_name] = {"status": "timeout", "reason": f"No response within {timeout}s"} unresponsive_servers.append(server_name) except Exception as e: results[server_name] = {"status": "error", "reason": str(e)} unresponsive_servers.append(server_name) actions_taken = [] if unresponsive_servers and action != "report": if action == "disable": config = self.config any_changes = False for server_name in unresponsive_servers: if server_name in config.servers: server = config.servers[server_name] if server.enabled: server.enabled = False any_changes = True await self.server_manager.unmount_server(server_name) actions_taken.append(f"Disabled {server_name}") results[server_name]["action"] = "disabled" else: actions_taken.append(f"{server_name} already disabled") results[server_name]["action"] = "already_disabled" else: actions_taken.append(f"Failed to disable {server_name}") results[server_name]["action"] = "disable_failed" if any_changes: if not self.save_config(config): logger.error("Failed to save config after disabling servers") else: for server_name in unresponsive_servers: if action == "remount": await self.server_manager.unmount_server(server_name) server = self.config.servers.get(server_name) if server and server.enabled: mount_success = await self.server_manager.mount_server(server) if mount_success: actions_taken.append(f"Remounted {server_name}") results[server_name]["action"] = "remounted" else: actions_taken.append(f"Failed to remount {server_name}") results[server_name]["action"] = "remount_failed" elif action == "unmount": await self.server_manager.unmount_server(server_name) actions_taken.append(f"Unmounted {server_name}") results[server_name]["action"] = "unmounted" return MaggResponse.success({ "servers_checked": len(results), "healthy": len([r for r in results.values() if r["status"] == "healthy"]), "unresponsive": len(unresponsive_servers), "results": results, "actions_taken": actions_taken if actions_taken else None }) except Exception as e: return MaggResponse.error(f"Failed to check servers: {str(e)}")
• magg/server/server.py:48-80 (registration)

  Registers the 'magg_check' tool by including (self.check, f\"{self_prefix_}check\", None) in the tools list and applying the FastMCP tool decorator via the loop. self_prefix_ resolves to 'magg_' making the tool name 'magg_check'. Also wraps handlers to convert MaggResponse to JSON text content.
  tools = [ (self.add_server, f"{self_prefix_}add_server", None), (self.remove_server, f"{self_prefix_}remove_server", None), (self.list_servers, f"{self_prefix_}list_servers", None), (self.enable_server, f"{self_prefix_}enable_server", None), (self.disable_server, f"{self_prefix_}disable_server", None), (self.search_servers, f"{self_prefix_}search_servers", None), (self.smart_configure, f"{self_prefix_}smart_configure", None), (self.analyze_servers, f"{self_prefix_}analyze_servers", None), (self.status, f"{self_prefix_}status", None), (self.check, f"{self_prefix_}check", None), (self.reload_config_tool, f"{self_prefix_}reload_config", None), (self.load_kit, f"{self_prefix_}load_kit", None), (self.unload_kit, f"{self_prefix_}unload_kit", None), (self.list_kits, f"{self_prefix_}list_kits", None), (self.kit_info, f"{self_prefix_}kit_info", None), ] def call_tool_wrapper(func): @wraps(func) async def wrapper(*args, **kwds): result = await func(*args, **kwds) if isinstance(result, MaggResponse): return result.as_json_text_content return result return wrapper for method, tool_name, options in tools: self.mcp.tool(name=tool_name, **(options or {}))(call_tool_wrapper(method))