MAGG

Instructions

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()