MAGG

Instructions

Implementation Reference

• magg/server/manager.py:508-535 (handler)

  The primary handler function for the 'magg_load_kit' MCP tool. Accepts a kit name, loads the kit configuration using KitManager, saves the config, and mounts any new enabled servers.

  async def load_kit(
          self,
          name: Annotated[str, Field(description="Kit name to load (filename without .json)")],
  ) -> MaggResponse:
      """Load a kit and its servers into the configuration."""
      try:
          config = self.config
          success, message = self.kit_manager.load_kit_to_config(name, config)
  
          if success:
              if not self.save_config(config):
                  return MaggResponse.error("Failed to save configuration")
  
              for server_name, server in config.servers.items():
                  if server.enabled and server_name not in self.server_manager.mounted_servers:
                      await self.server_manager.mount_server(server)
  
              return MaggResponse.success({
                  "action": "kit_loaded",
                  "kit": name,
                  "message": message
              })
          else:
              return MaggResponse.error(message)
  
      except Exception as e:
          return MaggResponse.error(f"Failed to load kit: {str(e)}")

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

  Registers the 'magg_load_kit' tool (and other kit tools) by including self.load_kit in the tools list and applying the MCP tool decorator via self.mcp.tool.

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

• magg/kit.py:176-222 (helper)

  KitManager helper method called by the tool handler to perform the actual kit loading into the config, adding/updating servers, and tracking kits.

  def load_kit_to_config(self, kit_name: str, config: 'MaggConfig') -> tuple[bool, str]:
      """Load a kit and its servers into the configuration.
  
      Returns:
          Tuple of (success, message)
      """
      if kit_name in config.kits:
          return False, f"Kit '{kit_name}' is already loaded"
  
      available_kits = self.discover_kits()
      if kit_name not in available_kits:
          return False, f"Kit '{kit_name}' not found in any kit.d directory"
  
      kit_path = available_kits[kit_name]
      kit_config = self.load_kit(kit_path)
      if not kit_config:
          return False, f"Failed to load kit '{kit_name}' from {kit_path}"
  
      self.add_kit(kit_name, kit_config)
  
      servers_added = []
      servers_updated = []
  
      for server_name, server_config in kit_config.servers.items():
          if server_name in config.servers:
              if kit_name not in config.servers[server_name].kits:
                  config.servers[server_name].kits.append(kit_name)
                  servers_updated.append(server_name)
          else:
              server_config.kits = [kit_name]
              config.servers[server_name] = server_config
              servers_added.append(server_name)
  
      config.kits[kit_name] = KitInfo(
          name=kit_name,
          description=kit_config.description,
          path=str(kit_path),
          source="file"
      )
  
      msg_parts = [f"Kit '{kit_name}' loaded successfully"]
      if servers_added:
          msg_parts.append(f"Added servers: {', '.join(servers_added)}")
      if servers_updated:
          msg_parts.append(f"Updated servers: {', '.join(servers_updated)}")
      return True, ". ".join(msg_parts)

• magg/server/manager.py:510-511 (schema)

  Input schema defined via Pydantic Annotated Field in the handler function signature.

          name: Annotated[str, Field(description="Kit name to load (filename without .json)")],
  ) -> MaggResponse: