Merge pull request #20 from johndoe6345789/copilot/move-code-to-workflow-plugins

Move loader implementations into workflow plugins

backend/autometabuilder/__init__.py

@@ -5,7 +5,7 @@ import json
 import os
 from pathlib import Path
 
-from .loaders.metadata_loader import load_metadata
+from .utils import load_metadata
 
 
 def _load_messages_path(path: Path) -> dict:

backend/autometabuilder/app_runner.py

@@ -2,8 +2,7 @@
 import argparse
 import logging
 import os
-from .loaders import load_env
-from .loaders import load_metadata
+from dotenv import load_dotenv
 from .engine import load_workflow_config, build_workflow_context, build_workflow_engine
 
 TRACE_LEVEL = 5
@@ -75,9 +74,22 @@ def run_web_workflow(logger):
     engine.execute()
 
 
+def _load_metadata_for_workflow() -> dict:
+    """Load metadata.json for workflow config."""
+    import json
+    from pathlib import Path
+    
+    metadata_path = Path(__file__).resolve().parent / "metadata.json"
+    if not metadata_path.exists():
+        return {}
+    
+    with metadata_path.open("r", encoding="utf-8") as f:
+        return json.load(f)
+
+
 def run_app() -> None:
     """Run the AutoMetabuilder CLI."""
-    load_env()
+    load_dotenv()
     configure_logging()
     logger = logging.getLogger("autometabuilder")
 
@@ -98,7 +110,7 @@ def run_app() -> None:
     }
     workflow_context = build_workflow_context(context_parts)
     
-    metadata = load_metadata()
+    metadata = _load_metadata_for_workflow()
     workflow_config = load_workflow_config(metadata)
     
     logger.info("Starting workflow: %s", workflow_config.get("name", "Unnamed"))

backend/autometabuilder/data/__init__.py

@@ -138,5 +138,5 @@ def get_ui_messages(lang):
     return _run_plugin("web.get_ui_messages", {"lang": lang})
 
 
-# Metadata - still using loaders directly
-from autometabuilder.loaders.metadata_loader import load_metadata
+# Metadata - utility function
+from autometabuilder.utils import load_metadata

backend/autometabuilder/loaders/__init__.py

@@ -1,33 +0,0 @@
-"""
-Loaders module for AutoMetabuilder.
-
-This module contains various loader utilities:
-- callable_loader: Load callables by dotted path
-- env_loader: Load environment variables from .env
-- metadata_loader: Load metadata.json
-- plugin_loader: Load custom tools from plugins directory
-- prompt_loader: Load prompt configuration
-- tool_policy_loader: Load tool policies from JSON
-- tool_registry_loader: Load tool registry entries
-- tools_loader: Load tool specs from JSON
-"""
-
-from .callable_loader import load_callable
-from .env_loader import load_env
-from .metadata_loader import load_metadata
-from .plugin_loader import load_plugins
-from .prompt_loader import load_prompt_yaml
-from .tool_policy_loader import load_tool_policies
-from .tool_registry_loader import load_tool_registry
-from .tools_loader import load_tools
-
-__all__ = [
-    "load_callable",
-    "load_env",
-    "load_metadata",
-    "load_plugins",
-    "load_prompt_yaml",
-    "load_tool_policies",
-    "load_tool_registry",
-    "load_tools",
-]

backend/autometabuilder/loaders/callable_loader.py

@@ -1,9 +0,0 @@
-"""Load a callable by dotted path."""
-import importlib
-
-
-def load_callable(path: str):
-    """Import and return a callable."""
-    module_path, attr = path.rsplit(".", 1)
-    module = importlib.import_module(module_path)
-    return getattr(module, attr)

backend/autometabuilder/loaders/env_loader.py

@@ -1,7 +0,0 @@
-"""Load environment variables from .env."""
-from dotenv import load_dotenv
-
-
-def load_env() -> None:
-    """Load environment variables."""
-    load_dotenv()

backend/autometabuilder/loaders/metadata_loader.py

@@ -1,38 +0,0 @@
-"""Load metadata.json."""
-from __future__ import annotations
-
-import json
-from pathlib import Path
-from typing import Any
-
-METADATA_PATH = Path(__file__).resolve().parent / "metadata.json"
-INCLUDED_SECTIONS = {
-    "settings_descriptions_path": "settings_descriptions",
-    "suggestions_path": "suggestions",
-    "workflow_plugins_path": "workflow_plugins",
-}
-
-
-def _read_json(path: Path) -> dict[str, Any]:
-    if not path.exists():
-        return {}
-    with path.open("r", encoding="utf-8") as f:
-        return json.load(f)
-
-
-def load_metadata() -> dict[str, Any]:
-    """Load metadata.json with optional section includes."""
-    metadata = _read_json(METADATA_PATH)
-    base_dir = METADATA_PATH.parent
-    for path_key, dest_key in INCLUDED_SECTIONS.items():
-        include_path = metadata.get(path_key)
-        if include_path:
-            resolved_path = base_dir / include_path
-            if resolved_path.is_dir():
-                merged: dict[str, Any] = {}
-                for file_path in sorted(resolved_path.glob("*.json")):
-                    merged.update(_read_json(file_path))
-                metadata[dest_key] = merged
-            else:
-                metadata[dest_key] = _read_json(resolved_path)
-    return metadata

backend/autometabuilder/loaders/plugin_loader.py

@@ -1,28 +0,0 @@
-"""Load custom tools from the plugins directory."""
-import importlib
-import inspect
-import logging
-import os
-
-logger = logging.getLogger("autometabuilder")
-
-
-def load_plugins(tool_map: dict, tools: list) -> None:
-    """Load plugin tools and append metadata."""
-    plugins_dir = os.path.join(os.path.dirname(__file__), "plugins")
-    if not os.path.exists(plugins_dir):
-        return
-
-    for filename in os.listdir(plugins_dir):
-        if filename.endswith(".py") and filename != "__init__.py":
-            module_name = f".plugins.{filename[:-3]}"
-            try:
-                module = importlib.import_module(module_name, package="autometabuilder")
-                for name, obj in inspect.getmembers(module):
-                    if inspect.isfunction(obj) and hasattr(obj, "tool_metadata"):
-                        tool_metadata = getattr(obj, "tool_metadata")
-                        tool_map[name] = obj
-                        tools.append(tool_metadata)
-                        logger.info("Loaded plugin tool: %s", name)
-            except Exception as error:  # pylint: disable=broad-exception-caught
-                logger.error("Failed to load plugin %s: %s", filename, error)

backend/autometabuilder/loaders/prompt_loader.py

@@ -1,14 +0,0 @@
-"""Load prompt configuration."""
-import os
-import yaml
-
-DEFAULT_PROMPT_PATH = "prompt.yml"
-
-
-def load_prompt_yaml() -> dict:
-    """Load prompt YAML from disk."""
-    local_path = os.environ.get("PROMPT_PATH", DEFAULT_PROMPT_PATH)
-    if os.path.exists(local_path):
-        with open(local_path, "r", encoding="utf-8") as f:
-            return yaml.safe_load(f)
-    raise FileNotFoundError(f"Prompt file not found at {local_path}")

backend/autometabuilder/loaders/tool_policy_loader.py

@@ -1,16 +0,0 @@
-"""Load tool policies from JSON."""
-import json
-import os
-
-
-def load_tool_policies() -> dict:
-    """Load tool policies JSON."""
-    path = os.path.join(os.path.dirname(__file__), "tool_policies.json")
-    if not os.path.exists(path):
-        return {"modifying_tools": []}
-    try:
-        with open(path, "r", encoding="utf-8") as f:
-            data = json.load(f)
-    except json.JSONDecodeError:
-        return {"modifying_tools": []}
-    return data if isinstance(data, dict) else {"modifying_tools": []}

backend/autometabuilder/loaders/tool_registry_loader.py

@@ -1,16 +0,0 @@
-"""Load tool registry entries."""
-import json
-import os
-
-
-def load_tool_registry() -> list:
-    """Load tool registry entries."""
-    path = os.path.join(os.path.dirname(__file__), "tool_registry.json")
-    if not os.path.exists(path):
-        return []
-    try:
-        with open(path, "r", encoding="utf-8") as f:
-            data = json.load(f)
-    except json.JSONDecodeError:
-        return []
-    return data if isinstance(data, list) else []

backend/autometabuilder/loaders/tools_loader.py

@@ -1,18 +0,0 @@
-"""Load tool specs from JSON."""
-import json
-import os
-
-
-def load_tools(metadata: dict) -> list:
-    """Load tool specs from metadata reference."""
-    tools_path = os.path.join(os.path.dirname(__file__), metadata.get("tools_path", "tools.json"))
-    if os.path.isdir(tools_path):
-        tools = []
-        for filename in sorted(os.listdir(tools_path)):
-            if not filename.endswith(".json"):
-                continue
-            with open(os.path.join(tools_path, filename), "r", encoding="utf-8") as f:
-                tools.extend(json.load(f))
-        return tools
-    with open(tools_path, "r", encoding="utf-8") as f:
-        return json.load(f)

backend/autometabuilder/utils.py

@@ -0,0 +1,101 @@
+"""Utility functions for AutoMetabuilder.
+
+This module provides helper functions that are used across the codebase.
+These are pure utility functions that don't contain business logic.
+"""
+import importlib
+import json
+import os
+import yaml
+from pathlib import Path
+from typing import Any
+
+
+def get_package_root() -> Path:
+    """Get the AutoMetabuilder package root directory.
+    
+    Returns the absolute path to the autometabuilder package root.
+    """
+    return Path(__file__).resolve().parent
+
+
+def read_json(path: Path) -> dict[str, Any]:
+    """Read JSON file."""
+    if not path.exists():
+        return {}
+    with path.open("r", encoding="utf-8") as f:
+        return json.load(f)
+
+
+def load_callable(path: str):
+    """Import and return a callable by dotted path.
+    
+    Args:
+        path: Dotted path to callable (e.g., 'module.submodule.function')
+        
+    Returns:
+        The callable object
+    """
+    module_path, attr = path.rsplit(".", 1)
+    module = importlib.import_module(module_path)
+    return getattr(module, attr)
+
+
+def load_metadata() -> dict[str, Any]:
+    """Load metadata.json with optional section includes.
+    
+    This is a utility function for loading metadata configuration.
+    """
+    included_sections = {
+        "settings_descriptions_path": "settings_descriptions",
+        "suggestions_path": "suggestions",
+        "workflow_plugins_path": "workflow_plugins",
+    }
+    
+    # Locate metadata.json in package root
+    metadata_path = get_package_root() / "metadata.json"
+    metadata = read_json(metadata_path)
+    base_dir = metadata_path.parent
+    
+    for path_key, dest_key in included_sections.items():
+        include_path = metadata.get(path_key)
+        if include_path:
+            resolved_path = base_dir / include_path
+            if resolved_path.is_dir():
+                merged: dict[str, Any] = {}
+                for file_path in sorted(resolved_path.glob("*.json")):
+                    merged.update(read_json(file_path))
+                metadata[dest_key] = merged
+            else:
+                metadata[dest_key] = read_json(resolved_path)
+    
+    return metadata
+
+
+def load_prompt_yaml() -> dict:
+    """Load prompt YAML from disk.
+    
+    This is a utility function for loading prompt configuration.
+    """
+    default_prompt_path = "prompt.yml"
+    local_path = os.environ.get("PROMPT_PATH", default_prompt_path)
+    if os.path.exists(local_path):
+        with open(local_path, "r", encoding="utf-8") as f:
+            return yaml.safe_load(f)
+    raise FileNotFoundError(f"Prompt file not found at {local_path}")
+
+
+def load_tool_registry() -> list:
+    """Load tool registry entries from tool_registry.json.
+    
+    This is a utility function for loading tool registry configuration.
+    """
+    path = get_package_root() / "tool_registry.json"
+    if not os.path.exists(path):
+        return []
+    try:
+        with open(path, "r", encoding="utf-8") as f:
+            data = json.load(f)
+    except json.JSONDecodeError:
+        return []
+    return data if isinstance(data, list) else []

backend/autometabuilder/workflow/plugin_loader.py

@@ -1,5 +1,5 @@
 """Load workflow plugins by dotted path."""
-from ..loaders.callable_loader import load_callable
+from ..utils import load_callable
 
 
 def load_plugin_callable(path: str):

backend/autometabuilder/workflow/plugins/backend/backend_build_tool_map/backend_build_tool_map.py

@@ -1,6 +1,5 @@
 """Workflow plugin: build tool map."""
-from ....loaders.callable_loader import load_callable
-from ....loaders.tool_registry_loader import load_tool_registry
+from .....utils import load_callable, load_tool_registry
 
 
 def _build_tool_map(gh, registry_entries: list) -> dict:

backend/autometabuilder/workflow/plugins/backend/backend_load_env/backend_load_env.py

@@ -1,8 +1,8 @@
 """Workflow plugin: load environment variables."""
-from ....loaders.env_loader import load_env
+from dotenv import load_dotenv
 
 
 def run(_runtime, _inputs):
     """Load environment variables from .env file."""
-    load_env()
+    load_dotenv()
     return {"result": "Environment loaded"}

backend/autometabuilder/workflow/plugins/backend/backend_load_metadata/backend_load_metadata.py

@@ -1,10 +1,10 @@
 """Workflow plugin: load metadata."""
-from ....loaders.metadata_loader import load_metadata
+from .....utils import load_metadata as util_load_metadata
 
 
 def run(runtime, _inputs):
     """Load metadata.json."""
-    metadata = load_metadata()
+    metadata = util_load_metadata()
     # Store in both store (for workflow) and context (for other plugins)
     runtime.context["metadata"] = metadata
     return {"result": metadata}

backend/autometabuilder/workflow/plugins/backend/backend_load_plugins/backend_load_plugins.py

@@ -1,10 +1,38 @@
 """Workflow plugin: load and register plugins."""
-from ....loaders.plugin_loader import load_plugins
+import importlib
+import inspect
+import logging
+import os
+from .....utils import get_package_root
+
+logger = logging.getLogger("autometabuilder")
+
+
+def _load_plugins(tool_map: dict, tools: list) -> None:
+    """Load plugin tools and append metadata."""
+    # Locate plugins directory in package root
+    plugins_dir = get_package_root() / "plugins"
+    if not os.path.exists(plugins_dir):
+        return
+
+    for filename in os.listdir(plugins_dir):
+        if filename.endswith(".py") and filename != "__init__.py":
+            module_name = f".plugins.{filename[:-3]}"
+            try:
+                module = importlib.import_module(module_name, package="autometabuilder")
+                for name, obj in inspect.getmembers(module):
+                    if inspect.isfunction(obj) and hasattr(obj, "tool_metadata"):
+                        tool_metadata = getattr(obj, "tool_metadata")
+                        tool_map[name] = obj
+                        tools.append(tool_metadata)
+                        logger.info("Loaded plugin tool: %s", name)
+            except Exception as error:  # pylint: disable=broad-exception-caught
+                logger.error("Failed to load plugin %s: %s", filename, error)
 
 
 def run(runtime, _inputs):
     """Load and register plugins."""
     tool_map = runtime.context.get("tool_map", {})
     tools = runtime.context.get("tools", [])
-    load_plugins(tool_map, tools)
+    _load_plugins(tool_map, tools)
     return {"result": True}

backend/autometabuilder/workflow/plugins/backend/backend_load_prompt/backend_load_prompt.py

@@ -1,10 +1,20 @@
 """Workflow plugin: load prompt configuration."""
 import os
-from ....loaders.prompt_loader import load_prompt_yaml
+import yaml
 
+DEFAULT_PROMPT_PATH = "prompt.yml"
 DEFAULT_MODEL = "openai/gpt-4o"
 
 
+def _load_prompt_yaml() -> dict:
+    """Load prompt YAML from disk."""
+    local_path = os.environ.get("PROMPT_PATH", DEFAULT_PROMPT_PATH)
+    if os.path.exists(local_path):
+        with open(local_path, "r", encoding="utf-8") as f:
+            return yaml.safe_load(f)
+    raise FileNotFoundError(f"Prompt file not found at {local_path}")
+
+
 def _resolve_model_name(prompt: dict) -> str:
     """Resolve model name from env or prompt."""
     return os.environ.get("LLM_MODEL", prompt.get("model", DEFAULT_MODEL))
@@ -12,7 +22,7 @@ def _resolve_model_name(prompt: dict) -> str:
 
 def run(runtime, _inputs):
     """Load prompt.yml."""
-    prompt = load_prompt_yaml()
+    prompt = _load_prompt_yaml()
     # Store in both store (for workflow) and context (for other plugins)
     runtime.context["prompt"] = prompt
     # Update model_name based on loaded prompt

backend/autometabuilder/workflow/plugins/backend/backend_load_tool_policies/backend_load_tool_policies.py

@@ -1,10 +1,26 @@
 """Workflow plugin: load tool policies."""
-from ....loaders.tool_policy_loader import load_tool_policies
+import json
+import os
+from .....utils import get_package_root
+
+
+def _load_tool_policies() -> dict:
+    """Load tool policies JSON."""
+    # Locate tool_policies.json in package root
+    path = get_package_root() / "tool_policies.json"
+    if not os.path.exists(path):
+        return {"modifying_tools": []}
+    try:
+        with open(path, "r", encoding="utf-8") as f:
+            data = json.load(f)
+    except json.JSONDecodeError:
+        return {"modifying_tools": []}
+    return data if isinstance(data, dict) else {"modifying_tools": []}
 
 
 def run(runtime, _inputs):
     """Load tool_policies.json."""
-    tool_policies = load_tool_policies()
+    tool_policies = _load_tool_policies()
     # Store in both store (for workflow) and context (for other plugins)
     runtime.context["tool_policies"] = tool_policies
     return {"result": tool_policies}

backend/autometabuilder/workflow/plugins/backend/backend_load_tool_registry/backend_load_tool_registry.py

@@ -1,10 +1,10 @@
 """Workflow plugin: load tool registry."""
-from ....loaders.tool_registry_loader import load_tool_registry
+from .....utils import load_tool_registry as util_load_tool_registry
 
 
 def run(runtime, _inputs):
     """Load tool registry entries."""
-    tool_registry = load_tool_registry()
+    tool_registry = util_load_tool_registry()
     # Store in context for other plugins
     runtime.context["tool_registry"] = tool_registry
     return {"result": tool_registry}

backend/autometabuilder/workflow/plugins/backend/backend_load_tools/backend_load_tools.py

@@ -1,11 +1,32 @@
 """Workflow plugin: load tools."""
-from ....loaders.tools_loader import load_tools
+import json
+import os
+from .....utils import get_package_root
+
+
+def _load_tools(metadata: dict) -> list:
+    """Load tool specs from metadata reference."""
+    # Locate tools relative to autometabuilder package root
+    base_dir = get_package_root()
+    tools_path = base_dir / metadata.get("tools_path", "tools.json")
+    
+    if tools_path.is_dir():
+        tools = []
+        for filename in sorted(os.listdir(tools_path)):
+            if not filename.endswith(".json"):
+                continue
+            with open(tools_path / filename, "r", encoding="utf-8") as f:
+                tools.extend(json.load(f))
+        return tools
+    
+    with open(tools_path, "r", encoding="utf-8") as f:
+        return json.load(f)
 
 
 def run(runtime, _inputs):
     """Load tool definitions."""
     metadata = runtime.context.get("metadata", {})
-    tools = load_tools(metadata)
+    tools = _load_tools(metadata)
     # Store in both store (for workflow) and context (for other plugins)
     runtime.context["tools"] = tools
     return {"result": tools}

backend/autometabuilder/workflow/plugins/web/web_create_translation/web_create_translation.py

@@ -2,7 +2,7 @@
 import json
 import shutil
 from pathlib import Path
-from autometabuilder.loaders.metadata_loader import load_metadata
+from autometabuilder.utils import load_metadata
 
 
 def run(_runtime, inputs):

backend/autometabuilder/workflow/plugins/web/web_delete_translation/web_delete_translation.py

@@ -2,7 +2,7 @@
 import json
 import shutil
 from pathlib import Path
-from autometabuilder.loaders.metadata_loader import load_metadata
+from autometabuilder.utils import load_metadata
 
 
 def run(_runtime, inputs):

backend/autometabuilder/workflow/plugins/web/web_get_ui_messages/web_get_ui_messages.py

@@ -1,7 +1,7 @@
 """Workflow plugin: get UI messages."""
 import json
 from pathlib import Path
-from autometabuilder.loaders.metadata_loader import load_metadata
+from autometabuilder.utils import load_metadata
 
 
 def run(_runtime, inputs):

backend/autometabuilder/workflow/plugins/web/web_get_workflow_content/web_get_workflow_content.py

@@ -1,6 +1,6 @@
 """Workflow plugin: get workflow content."""
 from pathlib import Path
-from autometabuilder.loaders.metadata_loader import load_metadata
+from autometabuilder.utils import load_metadata
 
 
 def run(_runtime, _inputs):

backend/autometabuilder/workflow/plugins/web/web_list_translations/web_list_translations.py

@@ -1,7 +1,7 @@
 """Workflow plugin: list translations."""
 import json
 from pathlib import Path
-from autometabuilder.loaders.metadata_loader import load_metadata
+from autometabuilder.utils import load_metadata
 
 
 def run(_runtime, _inputs):

backend/autometabuilder/workflow/plugins/web/web_load_translation/web_load_translation.py

@@ -1,7 +1,7 @@
 """Workflow plugin: load translation."""
 import json
 from pathlib import Path
-from autometabuilder.loaders.metadata_loader import load_metadata
+from autometabuilder.utils import load_metadata
 
 
 def run(_runtime, inputs):

backend/autometabuilder/workflow/plugins/web/web_load_workflow_packages/web_load_workflow_packages.py

@@ -2,7 +2,7 @@
 import json
 import logging
 from pathlib import Path
-from autometabuilder.loaders.metadata_loader import load_metadata
+from autometabuilder.utils import load_metadata
 
 logger = logging.getLogger(__name__)
 

backend/autometabuilder/workflow/plugins/web/web_register_blueprint/web_register_blueprint.py

@@ -1,4 +1,5 @@
 """Workflow plugin: register Flask blueprint."""
+from .....utils import load_callable
 
 
 def run(runtime, inputs):
@@ -25,7 +26,6 @@ def run(runtime, inputs):
         if not blueprint_path:
             return {"error": "blueprint or blueprint_path is required"}
         
-        from ....loaders.callable_loader import load_callable
         try:
             blueprint = load_callable(blueprint_path)
         except Exception as e:

backend/autometabuilder/workflow/plugins/web/web_route_context/web_route_context.py

@@ -1,7 +1,7 @@
 """Workflow plugin: context API routes blueprint."""
 import os
 from flask import Blueprint, jsonify
-from autometabuilder.loaders.metadata_loader import load_metadata
+from autometabuilder.utils import load_metadata
 from autometabuilder.workflow.plugin_loader import load_plugin_callable
 from autometabuilder.roadmap_utils import is_mvp_reached
 

backend/autometabuilder/workflow/plugins/web/web_route_navigation/web_route_navigation.py

@@ -1,6 +1,6 @@
 """Workflow plugin: navigation API routes blueprint."""
 from flask import Blueprint, jsonify
-from autometabuilder.loaders.metadata_loader import load_metadata
+from autometabuilder.utils import load_metadata
 from autometabuilder.workflow.workflow_graph import build_workflow_graph
 
 

backend/autometabuilder/workflow/plugins/web/web_route_translations/web_route_translations.py

@@ -1,6 +1,6 @@
 """Workflow plugin: translation API routes blueprint."""
 from flask import Blueprint, jsonify, request
-from autometabuilder.loaders.metadata_loader import load_metadata
+from autometabuilder.utils import load_metadata
 
 
 def run(runtime, _inputs):

backend/autometabuilder/workflow/plugins/web/web_update_translation/web_update_translation.py

@@ -1,7 +1,7 @@
 """Workflow plugin: update translation."""
 import json
 from pathlib import Path
-from autometabuilder.loaders.metadata_loader import load_metadata
+from autometabuilder.utils import load_metadata
 
 
 def run(_runtime, inputs):

backend/autometabuilder/workflow/plugins/web/web_write_workflow/web_write_workflow.py

@@ -1,6 +1,6 @@
 """Workflow plugin: write workflow."""
 from pathlib import Path
-from autometabuilder.loaders.metadata_loader import load_metadata
+from autometabuilder.utils import load_metadata
 
 
 def run(_runtime, inputs):

backend/tests/test_main.py

@@ -1,7 +1,7 @@
 import os
 import unittest
 
-from autometabuilder.loaders.prompt_loader import load_prompt_yaml
+from autometabuilder.utils import load_prompt_yaml
 
 class TestMain(unittest.TestCase):
     def test_load_prompt_yaml(self):