mirror of
https://github.com/johndoe6345789/metabuilder.git
synced 2026-04-25 06:14:59 +00:00
johndoe6345789
5ac579a2ed
Restructure workflow/ for multi-language plugin support:
- Rename src/ to core/ (engine code: DAG executor, registry, types)
- Create executor/{cpp,python,ts}/ for language-specific runtimes
- Consolidate plugins to plugins/{ts,python}/ by language then category
Add 80+ Python plugins from AutoMetabuilder in 14 categories:
- control: bot control, switch logic, state management
- convert: type conversions (json, boolean, dict, list, number, string)
- core: AI requests, context management, tool calls
- dict: dictionary operations (get, set, keys, values, merge)
- list: list operations (concat, find, sort, slice, filter)
- logic: boolean logic (and, or, xor, equals, comparisons)
- math: arithmetic operations (add, subtract, multiply, power, etc.)
- string: string manipulation (concat, split, replace, format)
- notifications: Slack, Discord integrations
- test: assertion helpers and test suite runner
- tools: file operations, git, docker, testing utilities
- utils: filtering, mapping, reducing, condition branching
- var: variable store operations (get, set, delete, exists)
- web: Flask server, environment variables, JSON handling
Add language executor runtimes:
- TypeScript: direct import execution (default, fast startup)
- Python: child process with JSON stdin/stdout communication
- C++: placeholder for native FFI bindings (Phase 3)
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
96 lines
2.7 KiB
Python
96 lines
2.7 KiB
Python
"""Python Plugin Executor
|
|
|
|
Runtime for Python workflow plugins.
|
|
Communicates with TypeScript via JSON over stdin/stdout.
|
|
"""
|
|
|
|
import json
|
|
import sys
|
|
import importlib
|
|
import importlib.util
|
|
from pathlib import Path
|
|
from typing import Any, Dict, Optional
|
|
|
|
|
|
class PythonRuntime:
|
|
"""Runtime environment for Python plugin execution."""
|
|
|
|
def __init__(self):
|
|
self.store: Dict[str, Any] = {}
|
|
self.context: Dict[str, Any] = {}
|
|
self.logger = self._create_logger()
|
|
|
|
def _create_logger(self):
|
|
"""Create a simple logger that writes to stderr."""
|
|
import logging
|
|
logger = logging.getLogger("workflow.python")
|
|
handler = logging.StreamHandler(sys.stderr)
|
|
handler.setFormatter(logging.Formatter("%(levelname)s: %(message)s"))
|
|
logger.addHandler(handler)
|
|
logger.setLevel(logging.INFO)
|
|
return logger
|
|
|
|
|
|
def load_plugin(plugin_path: str) -> Any:
|
|
"""Dynamically load a Python plugin module."""
|
|
path = Path(plugin_path)
|
|
if not path.exists():
|
|
raise FileNotFoundError(f"Plugin not found: {plugin_path}")
|
|
|
|
spec = importlib.util.spec_from_file_location(path.stem, path)
|
|
if spec is None or spec.loader is None:
|
|
raise ImportError(f"Cannot load plugin: {plugin_path}")
|
|
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def execute_plugin(plugin_path: str, inputs: Dict[str, Any], context: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""Execute a Python plugin and return the result."""
|
|
module = load_plugin(plugin_path)
|
|
|
|
if not hasattr(module, "run"):
|
|
raise AttributeError(f"Plugin {plugin_path} has no 'run' function")
|
|
|
|
runtime = PythonRuntime()
|
|
runtime.context = context
|
|
runtime.store = context.get("store", {})
|
|
|
|
result = module.run(runtime, inputs)
|
|
|
|
# Ensure result is a dict
|
|
if not isinstance(result, dict):
|
|
result = {"result": result}
|
|
|
|
return result
|
|
|
|
|
|
def main():
|
|
"""Main entry point for JSON-based communication."""
|
|
# Read input from stdin
|
|
try:
|
|
input_data = json.loads(sys.stdin.read())
|
|
except json.JSONDecodeError as e:
|
|
print(json.dumps({"error": f"Invalid JSON input: {e}"}))
|
|
sys.exit(1)
|
|
|
|
plugin_path = input_data.get("plugin_path")
|
|
inputs = input_data.get("inputs", {})
|
|
context = input_data.get("context", {})
|
|
|
|
if not plugin_path:
|
|
print(json.dumps({"error": "plugin_path is required"}))
|
|
sys.exit(1)
|
|
|
|
try:
|
|
result = execute_plugin(plugin_path, inputs, context)
|
|
print(json.dumps(result))
|
|
except Exception as e:
|
|
print(json.dumps({"error": str(e)}))
|
|
sys.exit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|