magg_search_servers
Search for MCP servers dynamically to extend LLM capabilities by querying and managing server configurations with defined result limits.
Instructions
Search for MCP servers online.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| limit | No | Maximum number of results to return per search source | |
| query | Yes | Search query for MCP servers |
Implementation Reference
- magg/server/server.py:636-668 (handler)The core handler function for the 'magg_search_servers' tool. It takes a query and optional limit, uses CatalogManager to search for MCP servers, formats the results, and returns them via MaggResponse.async def search_servers( self, query: Annotated[str, Field(description="Search query for MCP servers")], limit: Annotated[int, Field(description="Maximum number of results to return per search source")] = 5, ) -> MaggResponse: """Search for MCP servers online.""" try: catalog = CatalogManager() results = await catalog.search_only(query, limit) search_results = [] for source, items in results.items(): for item in items: result_data = { "source": source, "name": item.name, "description": item.description } if item.url: result_data["url"] = item.url if hasattr(item, 'install_command') and item.install_command: result_data["install_command"] = item.install_command search_results.append(result_data) return MaggResponse.success({ "query": query, "results": search_results, "total": len(search_results) }) except Exception as e: return MaggResponse.error(f"Failed to search servers: {str(e)}")
- magg/server/server.py:43-82 (registration)The _register_tools method that registers the 'magg_search_servers' tool by including (self.search_servers, f'{self_prefix_}search_servers', None) in the tools list and calling self.mcp.tool() with the handler wrapped.def _register_tools(self): """Register all Magg management tools programmatically. """ self_prefix_ = self.self_prefix_ 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)) self._register_resources() self._register_prompts()