feat: add system backends, generic and flux

Signed-off-by: vsoch <vsoch@users.noreply.github.com>

Author: vsoch

README.md

@@ -262,7 +262,20 @@ mcpserver start --hub --hub-secret potato
 In another terminal, start a worker using the token that is generated. Add some functions for fun.
 
 ```bash
-mcpserver start --config examples/jobspec/mcpserver.yaml --join http://0.0.0.0:8000 --join-secret portato --port 7777
+mcpserver start --config examples/jobspec/mcpserver.yaml --join http://0.0.0.0:8000 --join-secret potato --port 7777
+```
+
+Note that you can also set the secret in the environemnt.
+
+```bash
+export MCPSERVER_JOIN_SECRET=potato
+mcpserver start --config examples/jobspec/mcpserver.yaml --join http://0.0.0.0:8000 --port 7777
+```
+
+Register the worker sytem type instead as flux:
+
+```bash
+mcpserver start --config examples/jobspec/mcpserver.yaml --join http://0.0.0.0:8000 --port 7777 --system-type flux
 ```
 
 Test doing queries for status:
@@ -293,10 +306,9 @@ Here are a few design choices (subject to change, of course). I am starting with
 
 ## TODO
 
-- [ ] join secret should be allowed from environment
-- [ ] debug why port not taking
-- [ ] design --system-name and --system
-- [ ] Need to handle worker disconnect and reconnect.
+- [ ] need to expose tools from system (child worker) instances
+- [ ] need to decide on dispatch strategy / algorithm
+- [ ] add in fluxion queue stats via RPC call to flux status
 
 ## License
 

examples/mcp-query.py

@@ -11,7 +11,6 @@
 
 console = Console()
 
-# --- Defaults ---
 DEFAULT_URL = "http://localhost:8000/mcp"
 DEFAULT_TOOL = "get_fleet_status"
 
@@ -76,7 +75,7 @@ async def query_mcp(url, tool_name):
                 except:
                     data = text_content
 
-            # --- PRETTY RENDERING ---
+            # MAKE IT PRETTY.
             if tool_name == "get_fleet_status" and isinstance(data, dict):
                 console.print("\n")
                 console.print(Panel(render_fleet_tree(data), border_style="cyan", expand=False))

mcpserver/cli/args.py

@@ -64,13 +64,22 @@ def populate_start_args(start):
         default=os.environ.get("MCP_HUB_SECRET"),
         help="Secret key required for workers to register. (Auto-generated if omitted)",
     )
+    start.add_argument(
+        "--system-type",
+        default="generic",
+        help="System type/template (e.g., 'generic', 'flux', 'kubernetes') or a full python module path.",
+    )
 
     # Worker Registration Group
     worker_group = start.add_argument_group("🐝 Worker Registration")
     worker_group.add_argument(
         "--join", help="URL of the MCP Hub to join (e.g., http://hub-host:8089)"
     )
-    worker_group.add_argument("--join-secret", help="The registration secret provided by the Hub.")
+    worker_group.add_argument(
+        "--join-secret",
+        help="The registration secret provided by the Hub.",
+        default=os.environ.get("MCPSERVER_JOIN_SECRET"),
+    )
     worker_group.add_argument(
         "--register-id",
         help="Unique ID for this worker. Defaults to the hostname.",

mcpserver/cli/manager.py

@@ -6,7 +6,7 @@
 manager.register()
 
 
-def get_manager(mcp, cfg):
+def get_manager(mcp, cfg, system_type=None):
     """
     Get the common tool manager and register tools.
     """
@@ -20,7 +20,13 @@ def get_manager(mcp, cfg):
         print(f"   ✅ Registered: {endpoint.name}")
 
     # Load into the manager (tools, resources, prompts)
-    for tool in manager.load_tools(mcp, cfg.include, cfg.exclude):
+    # We pass the system_name and system path here
+    for tool in manager.load_tools(
+        mcp,
+        cfg.include,
+        cfg.exclude,
+        system_type=system_type,
+    ):
         print(f"   ✅ Registered: {tool.name}")
 
     # Visual to show user we have ssl

mcpserver/cli/start.py

@@ -35,7 +35,7 @@ def main(args, extra, **kwargs):
 
     # Get the tool manager and register discovered tools
     mcp = init_mcp(cfg.exclude, cfg.include, args.mask_error_details)
-    get_manager(mcp, cfg)
+    get_manager(mcp, cfg, system_type=args.system_type)
 
     # Create ASGI app from MCP server
     mcp_app = mcp.http_app(path=cfg.server.path)

mcpserver/core/hub.py

@@ -76,14 +76,13 @@ async def get_one(wid, info):
             }
             try:
                 async with info["client"] as sess:
-                    # 1. Call the tool
                     mcp_result = await sess.call_tool("get_status", {})
 
-                    # 2. Extract the text content from the MCP wrapper
+                    # Extract the text content from the MCP wrapper
                     # FastMCP result.content is a list of content blocks
                     raw_text = mcp_result.content[0].text
 
-                    # 3. Parse the string back into a dictionary
+                    # Parse the string back into a dictionary
                     try:
                         # Handle potential single quotes from Python's str(dict)
                         status_data = json.loads(raw_text.replace("'", '"'))
@@ -150,7 +149,7 @@ def _create_proxy(self, worker_id: str, tool: Tool):
         # Generate a safe function name
         proxy_name = f"{utils.sanitize(worker_id)}_{utils.sanitize(tool.name)}"
 
-        # FIX: Check if this tool is already registered to avoid ValueError on re-registration
+        # Check if this tool is already registered to avoid ValueError on re-registration
         if proxy_name in self._registered_proxies:
             print(f"🛰️  Re-discovered worker tool: [blue]{proxy_name}[/blue]")
             return
@@ -164,8 +163,8 @@ def _create_proxy(self, worker_id: str, tool: Tool):
         # Create the signature string: arg_1=None, arg_2=None
         arg_string = ", ".join([f"{safe_name}=None" for safe_name in arg_mapping.keys()])
 
-        # 3. Build the dynamic function
-        # FIX: We pass 'self' (the HubManager instance) as 'hub' to the exec scope.
+        # Build the dynamic function
+        # We pass 'self' (the HubManager instance) as 'hub' to the exec scope.
         # This allows the proxy function to look up the LATEST url for the worker
         # every time it is called, instead of using a hardcoded stale URL.
         exec_globals = {

mcpserver/core/worker.py

@@ -42,13 +42,16 @@ def from_args(cls, mcp, args, cfg) -> Optional["WorkerManager"]:
         public_url = (
             args.public_url or f"http://{cfg.server.host}:{cfg.server.port}{cfg.server.path}"
         )
+
+        sys_type = getattr(args, "system_type", "generic")
+        worker_type = sys_type.split(".")[-1] if "." in sys_type else sys_type
         return cls(
             mcp,
             hub_url=args.join,
             secret=args.join_secret,
             worker_id=args.register_id,
             public_url=public_url,
-            worker_type=args.worker_type,
+            worker_type=worker_type,
             labels=args.labels,
         )
 

mcpserver/tools/manager.py

@@ -11,6 +11,9 @@
 # These are the function types we want to discover
 from fastmcp.tools import Tool
 
+import mcpserver.defaults as defaults
+import mcpserver.tools.system as systems
+
 from .base import BaseTool
 
 
@@ -104,13 +107,20 @@ def discover_tools(self, root_path: str, module_path: str) -> Dict[str, Path]:
             discovered[tool_id] = {"path": file_path, "module": import_path, "root": root_path}
         return discovered
 
-    def load_tools(self, mcp, include=None, exclude=None, status_tool: str = None):
+    def load_tools(self, mcp, include=None, exclude=None, system_type="generic"):
         """
         Load a set of named tools, or default to all those discovered.
         """
-        # Start with the system status tool
-        status_tool = status_tool or "mcpserver.tools.system.tool"
-        self.tools["system"] = {"module": status_tool}
+        if system_type in systems.system_tools:
+            # Map short name to internal library path
+            sys_module = f"mcpserver.tools.system.{system_type}"
+        else:
+            # Assume it is a custom external module path provided by the user
+            # e.g., 'my_custom_package.system_logic'
+            sys_module = system_type
+
+        # Seed the system tool into discovery
+        self.tools["system"] = {"module": sys_module}
 
         # If no tools are selected... select all tools discovered
         names = self.tools
@@ -199,13 +209,13 @@ def get_available_prompts(self):
         """
         prompts = set()
 
-        # 2. Load them (to execute decorators)
+        # Load them (to execute decorators)
         for tool_id, path in self.tools.items():
             mod = self.load_tool_module(tool_id, path)
             if not mod:
                 continue
 
-            # 3. Inspect the classes/functions in the module
+            # Inspect the classes/functions in the module
             for name, obj in inspect.getmembers(mod):
                 # We usually look for classes inheriting from BaseTool
                 # But we can also just scan the class attributes

mcpserver/tools/system/__init__.py

@@ -0,0 +1,9 @@
+from .flux import SystemTool as FluxSystemTool
+from .generic import SystemTool as GenericSystemTool
+from .kubernetes import SystemTool as KubernetesSystemTool
+
+system_tools = {
+    "flux": FluxSystemTool,
+    "generic": GenericSystemTool,
+    "kubernetes": KubernetesSystemTool,
+}

mcpserver/tools/system/flux.py

@@ -0,0 +1,42 @@
+import time
+from typing import Any, Dict
+
+from mcpserver.tools.base import BaseTool
+from mcpserver.tools.decorator import mcp
+
+
+class SystemTool(BaseTool):
+    """
+    System tool specialized for Flux Framework.
+    """
+
+    def setup(self, manager=None):
+        self.manager = manager
+
+    @mcp.tool(name="get_status")
+    def get_status(self) -> Dict[str, Any]:
+        """
+        Get status of the flux cluster.
+        """
+        import flux
+        import flux.resource
+
+        flux_meta = {"status": "error", "message": "Flux handle failed"}
+        try:
+            h = flux.Flux()
+            listing = flux.resource.list.resource_list(h).get()
+            flux_meta = {
+                "status": "online",
+                "free_cores": listing.free.ncores,
+                "up_nodes": listing.up.nnodes,
+            }
+        except Exception as e:
+            flux_meta["error"] = str(e)
+
+        res = {"timestamp": time.time(), "system_type": "flux", "flux": flux_meta, "tools": {}}
+        if self.manager:
+            for tid, inst in self.manager.instances.items():
+                if inst == self:
+                    continue
+                res["tools"][tid] = {"class": inst.__class__.__name__}
+        return res