MAGG

Instructions

Implementation Reference

• magg/server/server.py:669-723 (handler)

  The main handler function that executes the tool logic: analyzes configured servers, computes statistics, and optionally uses an LLM (via provided Context) to generate insights.

      async def analyze_servers(
          self,
          ctx: Context | None = None,
      ):
          """Analyze configured servers and provide insights using LLM."""
          try:
              config = self.config
  
              if not config.servers:
                  return MaggResponse.success({
                      "analysis": f"No servers configured yet. Use {self.self_prefix_}add_server to add servers."
                  })
  
              analysis_data = {
                  "total_servers": len(config.servers),
                  "enabled_servers": len(config.get_enabled_servers()),
                  "mounted_servers": len(self.server_manager.mounted_servers),
                  "servers": {}
              }
  
              for name, server in config.servers.items():
                  server_info = {
                      "source": server.source,
                      "enabled": server.enabled,
                      "mounted": name in self.server_manager.mounted_servers,
                      "command": server.command,
                      "uri": server.uri,
                      "prefix": server.prefix,
                      "notes": server.notes
                  }
                  analysis_data["servers"][name] = server_info
  
              if ctx:
                  prompt = f"""Analyze this Magg server configuration and provide insights:
  
  {json.dumps(analysis_data, indent=2)}
  
  Please provide:
  1. Overview of the current setup
  2. Any potential issues or conflicts
  3. Suggestions for optimization
  4. Missing capabilities that could be added"""
  
                  result = await ctx.sample(
                      messages=prompt,
                      max_tokens=1000
                  )
  
                  if result and result.text:
                      analysis_data["insights"] = result.text
  
              return MaggResponse.success(analysis_data)
  
          except Exception as e:
              return MaggResponse.error(f"Failed to analyze servers: {str(e)}")

• magg/server/server.py:43-80 (registration)

  Registers the analyze_servers method as the tool named f'{self.self_prefix_}analyze_servers' (likely 'magg_analyze_servers') within the list of Magg management tools, using FastMCP's tool decorator.

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