Index

The weave package provides functionality for creating MCP servers related to WEAVE-managed tools, like Maestro.

Subpackages

maestro: Handles MCP tooling for Maestro-specific logic. study_construction: Provides an ABC class for constructing studies that WEAVE-managed orchestration tools can interpret.

MaestroCommandExecutionServer

Bases: BaseMCPServer

MCP Server for Maestro command execution.

Attributes:

Name	Type	Description
command_executor		A MaestroCommandExecutor instance used to invoke Maestro CLI commands for running and managing workflows.

Methods:

Name	Description
register_jinja_study_tool	Register a tool backed by a single Jinja study template, or register multiple tools from a templates directory.
register_study_construction_tools	Abstract hook for subclasses to register their own study construction tools.
_register_path_tools	Register tools for working with filesystem paths.
_register_workflow_management_tools	Register tools for running, querying, canceling, and updating Maestro workflows.
_register_tools	Register all built-in and subclass-provided tools.

Source code in mada_tools/workflow/weave/maestro/server.py

class MaestroCommandExecutionServer(BaseMCPServer):
    """
    MCP Server for Maestro command execution.

    Attributes:
        command_executor: A `MaestroCommandExecutor` instance used to invoke
            Maestro CLI commands for running and managing workflows.

    Methods:
        register_jinja_study_tool: Register a tool backed by a single Jinja
            study template, or register multiple tools from a templates
            directory.
        register_study_construction_tools: Abstract hook for subclasses to
            register their own study construction tools.
        _register_path_tools: Register tools for working with filesystem paths.
        _register_workflow_management_tools: Register tools for running,
            querying, canceling, and updating Maestro workflows.
        _register_tools: Register all built-in and subclass-provided tools.
    """

    def __init__(self):
        """
        Constructor for `MaestroCommandExecutionServer`.
        """
        super().__init__("Maestro Command Execution Application", "A server for executing Maestro commands")

        # Command executor for running Maestro CLI commands
        self.command_executor = MaestroCommandExecutor()

    def _register_tools(self):
        """
        Register MCP tools for Maestro command execution operations.
        """

        @self.mcp.tool()
        async def run_workflow(
            workflow_yaml: str | Path,
            attempts: int = 1,
            rlimit: int = 1,
            throttle: int = 0,
            sleeptime: int = 60,
            output_path: str | Path = None,
            pgen: str | Path = None,
            pargs: List[str] | None = None,
            dry: bool = False,
            foreground: bool = False,
            hash_ws: bool = False,
            use_tmp: bool = False,
        ) -> str:
            """
            Given a workflow YAML file, run the workflow using Maestro.

            This tool will execute the `maestro run` command with the provided workflow YAML file in
            a subprocess. Behind the scenes of this command, Maestro will convert the workflow YAML into
            a DAG, expand parameters/variables throughout the spec file, and excecute the DAG by converting
            steps into shell scripts that get executed.

            Args:
                workflow_yaml (str | Path): Path to the Maestro workflow YAML file.
                attempts (int): Maximum number of submission attempts before a step is marked as failed.
                    Default 1.
                rlimit (int): Maximum number of restarts allowed when steps specify a restart command (0
                    denotes no limit). Default 1.
                throttle (int): Maximum number of inflight jobs allowed to execute simultaneously (0 denotes
                    not throttling). Default 0.
                sleeptime (int): Amount of time (in seconds) for the manager to wait between job status checks.
                    Default 60.
                output_path (str | Path): Output path to place study in (NOTE: overrides OUTPUT_PATH in the
                    specified specification). Default None.
                pgen (str | Path): Path to a Python file that defines a custom Maestro ParameterGenerator.
                    Passed through to `maestro run --pgen ...`. Default None.
                pargs (List[str] | None): Optional arguments to pass to the custom parameter generation
                    function. Passed through as one or more `maestro run --pargs ...` values. Requires `pgen`.
                    Each list entry should be of the form "PARAM_NAME:VALUE".
                dry (bool): Generate the directory structure and scripts for a study but do not launch it.
                    Default False.
                foreground (bool): Runs the backend conductor in the foreground instead of using nohup.
                    Default False.
                hash_ws (bool): Enable hashing of subdirectories in parameterized studies (NOTE: breaks commands
                    that use parameter labels to search directories). Default False.
                use_tmp (bool): make use of a temporary directory for dumping scripts and other Maestro-related
                    files. Default False.

            Returns:
                str: The output message from the command execution.

            Raises:
                ValueError: If argument validation fails.
                ToolExecutionError: If the underlying Maestro command fails.

            Example:

                ```python
                run_workflow("/path/to/workflow.yaml")
                run_workflow("/path/to/workflow.yaml", attempts=4)
                run_workflow("/path/to/workflow.yaml", throttle=10, dry=True)
                run_workflow("/path/to/workflow.yaml", pgen="/path/to/pgen.py")
                ```
            """
            return await self.run_tool(
                self.command_executor.run_workflow,
                workflow_yaml,
                attempts=attempts,
                rlimit=rlimit,
                throttle=throttle,
                sleeptime=sleeptime,
                output_path=output_path,
                pgen=pgen,
                pargs=pargs,
                dry=dry,
                foreground=foreground,
                hash_ws=hash_ws,
                use_tmp=use_tmp,
            )

        @self.mcp.tool()
        async def get_statuses(
            workflow_dirs: List[str | Path],
            layout: str = "flat",
            disable_theme: bool = False,
        ) -> str:
            """
            Get the statuses of currently running Maestro workflows.

            This tool will run a subprocess to execute the `maestro status` command.
            This command will take in a list of output directories where the output of
            currently running workflows are located and return their statuses.

            Args:
                workflow_dirs (List[str | Path]): A list of paths to Maestro workflow
                    output directories.
                layout (str): The layout of the status table. Options are "flat", "legacy",
                    and "narrow". Default "flat".
                disable_theme (bool): Turn off styling for the status layout.

            Returns:
                str: The output message from the command execution.

            Raises:
                ToolExecutionError: If the underlying Maestro command fails.

            Example:

                ```python
                get_statuses(["/path/to/workflow/dir", "/path/to/another_workflow/dir"])
                get_statuses(
                    ["/path/to/workflow/dir", "/path/to/another_workflow/dir"],
                    layout="narrow",
                    disable_theme=True
                )
                ```
            """
            return await self.run_tool(
                self.command_executor.get_statuses, workflow_dirs, layout=layout, disable_theme=disable_theme
            )

        @self.mcp.tool()
        async def cancel_workflows(workflow_dirs: List[str | Path]) -> str:
            """
            Cancel one or more running Maestro workflows.

            This tool will run a subprocess to execute the `maestro cancel` command.
            This command will take in a list of output directories where the output of
            currently running workflows are located, shut down any running jobs for these
            workflows, and stop any future jobs associated with these workflows from running.

            Args:
                workflow_dirs (List[str | Path]): A list of paths to Maestro workflow
                    output directories.

            Returns:
                str: The output message from the command execution.

            Raises:
                ToolExecutionError: If the underlying Maestro command fails.

            Example:

                ```python
                command_executor = MaestroCommandExecutor()
                workflows_to_cancel = ["/path/to/workflow/dir", "/path/to/another_workflow/dir"]
                success, msg = command_executor.cancel_workflows(workflows_to_cancel)
                if success:
                    print(f"Workflow cancelled successfully: {msg}")
                else:
                    print(f"Failed to cancel workflow: {msg}")
                ```
            """
            return await self.run_tool(self.command_executor.cancel_workflows, workflow_dirs)

        @self.mcp.tool()
        async def update_workflows(
            workflow_dirs: List[str | Path],
            rlimit: int = None,
            throttle: int = None,
            sleeptime: int = None,
        ) -> str:
            """
            Update the configs of running studies (throttle, rlimit, and/or sleep).

            This tool will run a subprocess to execute the `maestro update` command.
            This command will take in a list of output directories where the output of
            currently running workflows are located and rlimit, throttle, and/or sleeptime
            settings. It will then update the configurations of any workflow provided
            with the new settings.

            Args:
                workflow_dirs (List[str | Path]): A list of paths to Maestro workflow output directories.
                rlimit (int): Maximum number of restarts allowed when steps specify a restart command (0
                    denotes no limit). Default 1.
                throttle (int): Maximum number of inflight jobs allowed to execute simultaneously (0 denotes
                    not throttling). Default 0.
                sleeptime (int): Amount of time (in seconds) for the manager to wait between job status checks.
                    Default 60.

            Returns:
                str: The output message from the command execution.

            Raises:
                ToolExecutionError: If the underlying Maestro command fails.
            """
            return await self.run_tool(
                self.command_executor.update_workflows,
                workflow_dirs,
                rlimit=rlimit,
                throttle=throttle,
                sleeptime=sleeptime,
            )

__init__()

Constructor for MaestroCommandExecutionServer.

Source code in mada_tools/workflow/weave/maestro/server.py

def __init__(self):
    """
    Constructor for `MaestroCommandExecutionServer`.
    """
    super().__init__("Maestro Command Execution Application", "A server for executing Maestro commands")

    # Command executor for running Maestro CLI commands
    self.command_executor = MaestroCommandExecutor()

WEAVEStudyConstructionServer

Bases: BaseMCPServer, ABC

Abstract base class for constructing studies that WEAVE Orchestration tools can execute. WEAVE Orchestration tools include Maestro, Merlin, and StudyWeaver.

This class is not intended to be instantiated directly. Subclasses define project-specific tools for constructing studies from Jinja-templated workflow templates, while inheriting common utilities for path handling.

This server should be used alongside WEAVE-Orchestration-tool-specific command execution servers. For instance, the MaestroCommandExecutionServer.

Attributes:

Name	Type	Description
study_constructor		A WEAVEStudyConstructor instance used to build template contexts, render templates, and write WEAVE YAML files.

Methods:

Name	Description
register_jinja_study_tool	Register a tool backed by a single Jinja study template, or register multiple tools from a templates directory.
register_study_construction_tools	Abstract hook for subclasses to register their own study construction tools.
_register_path_tools	Register tools for working with filesystem paths.
_register_tools	Register all built-in and subclass-provided tools.

Source code in mada_tools/workflow/weave/study_construction/server.py

class WEAVEStudyConstructionServer(BaseMCPServer, ABC):
    """
    Abstract base class for constructing studies that WEAVE Orchestration tools
    can execute. WEAVE Orchestration tools include Maestro, Merlin, and StudyWeaver.

    This class is not intended to be instantiated directly. Subclasses define
    project-specific tools for constructing studies from Jinja-templated
    workflow templates, while inheriting common utilities for path handling.

    This server should be used alongside WEAVE-Orchestration-tool-specific command
    execution servers. For instance, the `MaestroCommandExecutionServer`.

    Attributes:
        study_constructor: A `WEAVEStudyConstructor` instance used to build
            template contexts, render templates, and write WEAVE YAML files.

    Methods:
        register_jinja_study_tool: Register a tool backed by a single Jinja
            study template, or register multiple tools from a templates
            directory.
        register_study_construction_tools: Abstract hook for subclasses to
            register their own study construction tools.
        _register_path_tools: Register tools for working with filesystem paths.
        _register_tools: Register all built-in and subclass-provided tools.
    """

    def __init__(
        self,
        server_name: str,
        description: str,
        study_templates_dir: str | Path,
    ):
        """
        Constructor for `WEAVEStudyConstructionServer`.

        Args:
            server_name (str):
                The name of the MCP server
            description (str):
                Description of the server
            study_templates_dir (str | Path):
                The path to the directory where templated study YAML files live.
        """
        super().__init__(server_name, description)

        # Class for constructing workflows from templates
        self.study_constructor = WEAVEStudyConstructor(study_templates_dir)

    def register_jinja_study_tool(
        self,
        *,
        tool_name: str | None = None,
        template_name: str | Path | None = None,
        templates_dir: str | Path | None = None,
        preprocess: Callable[[Dict[str, Any]], Dict[str, Any]] | None = None,
    ) -> None:
        """Register a template-backed tool with a minimal call signature.

        The tool signature is intentionally small so we don't have to re-declare
        template keys/defaults in Python. Agents discover valid keys via the
        template `mcp_doc` section (plus the auto-generated key list).

        Args:
            tool_name: MCP tool name to register (required when `templates_dir` is not provided).
            template_name: Template filename or absolute path (required when `templates_dir` is not provided).
            templates_dir: If provided, register one tool per `*.yaml` file in this directory.
                Tool names are derived as `construct_{template_stem}_study` and preprocess hooks are
                resolved as `self._prep_{template_stem}` when present.
            preprocess: Optional hook to validate/normalize override values before rendering.

        Returns:
            None

        Raises:
            ValueError: If arguments are inconsistent (e.g., both `templates_dir` and `template_name` are provided).
            FileNotFoundError: If `templates_dir` does not exist.
        """

        if templates_dir is not None:
            if tool_name is not None or template_name is not None or preprocess is not None:
                raise ValueError(
                    "When `templates_dir` is provided, do not pass `tool_name`, `template_name`, or `preprocess`."
                )
            dir_path = templates_dir if isinstance(templates_dir, Path) else Path(str(templates_dir))
            if not dir_path.is_absolute():
                dir_path = self.study_constructor.study_templates_dir / dir_path
            if not dir_path.exists():
                raise FileNotFoundError(f"Templates directory '{dir_path}' does not exist.")

            for template_path in sorted(dir_path.glob("*.yaml")):
                stem = template_path.stem
                if stem.endswith("_study"):
                    stem = stem[: -len("_study")]
                derived_tool_name = f"construct_{stem}_study"
                derived_preprocess = getattr(self, f"_prep_{stem}", None)
                self.register_jinja_study_tool(
                    tool_name=derived_tool_name,
                    template_name=template_path,
                    preprocess=derived_preprocess,
                )
            return

        if tool_name is None or template_name is None:
            raise ValueError("Pass `tool_name` + `template_name`, or pass `templates_dir`.")

        async def _tool(
            overrides: Dict[str, Any] | None = None,
            output_dir: str | Path | None = None,
        ) -> str:
            """Render the registered Jinja study template and write a Maestro YAML."""
            context = await self.run_tool(
                self.study_constructor.build_context,
                template_name,
                overrides=overrides,
                preprocess=preprocess,
                background=False,
            )
            return await self.run_tool(
                self.study_constructor.write_yaml_tool,
                template_name,
                context,
                output_dir=output_dir,
                background=False,
            )

        _tool.__name__ = tool_name
        _tool.__doc__ = self.study_constructor.get_tool_doc_from_template(template_name)
        self.mcp.tool()(_tool)

    def _register_path_tools(self):
        """Register tools related to file system paths."""

        @self.mcp.tool()
        async def abspath(output_path: str, base_path: str = None) -> str:
            """Convert a user-provided path into an absolute path on the server.

            Args:
                output_path: Path to convert. Relative paths are resolved against the `base_path`.
                base_path: The base path that the `output_path` is relative to. Defaults to the
                    current working directory.

            Returns:
                The absolute path as a string.
            """
            return await self.run_tool(
                self.study_constructor.abspath,
                output_path,
                base_path=base_path,
                background=False,
            )

    @abstractmethod
    def register_study_construction_tools(self) -> None:
        """Hook for subclasses to register their own template-backed study tools.

        Each tool defined here should utilize the `self.register_jinja_study_tool()`.
        Preprocessing can be added to each tool by creating a separate method in your
        class and linking your tool to it.

        Example:

            Here's a full example that defines a study construction tool with preprocessing.
            In this example, we're targeting the creation of a study called `my_cool_study`
            that needs a path `py3_utils` in order to run. As part of the preprocessing, we
            coerce the path of `py3_utils` to ensure it's provided and exists.

            ```python
            from mada_tools.workflow.weave import WEAVEStudyConstructionServer

            class MyProjectStudyConstructionServer(WEAVEStudyConstructionServer):
                def _prep_my_cool_study(self, overrides: Dict[str, Any]) -> Dict[str, Any]:
                    # Preprocess this study by validating a path that's needed
                    raw_py3 = overrides.get("py3_utils")
                    if raw_py3 is None:
                        py3_utils_path = Path(__file__).parent / "py3utils"
                    else:
                        py3_utils_path = raw_py3 if isinstance(raw_py3, Path) else Path(str(raw_py3))
                    if not py3_utils_path.exists():
                        raise FileNotFoundError(f"The py3utils directory '{py3_utils_path}' does not exist.")
                    overrides["py3_utils"] = str(py3_utils_path)
                    return overrides

                def register_study_construction_tools(self):

                    # Register the tool that constructs your cool study
                    self.register_jinja_study_tool(
                        tool_name="construct_my_cool_study",
                        template_name="my_cool_study.yaml",
                        preprocess=self._prep_my_cool_study,
                    )
            ```
        """
        pass

    def _register_tools(self):
        """Register MCP tools for WEAVE study construction operations."""

        self._register_path_tools()
        self.register_study_construction_tools()

__init__(server_name, description, study_templates_dir)

Constructor for WEAVEStudyConstructionServer.

Parameters:

Name	Type	Description	Default
server_name	str	The name of the MCP server	required
description	str	Description of the server	required
study_templates_dir	str | Path	The path to the directory where templated study YAML files live.	required

Source code in mada_tools/workflow/weave/study_construction/server.py

def __init__(
    self,
    server_name: str,
    description: str,
    study_templates_dir: str | Path,
):
    """
    Constructor for `WEAVEStudyConstructionServer`.

    Args:
        server_name (str):
            The name of the MCP server
        description (str):
            Description of the server
        study_templates_dir (str | Path):
            The path to the directory where templated study YAML files live.
    """
    super().__init__(server_name, description)

    # Class for constructing workflows from templates
    self.study_constructor = WEAVEStudyConstructor(study_templates_dir)

register_jinja_study_tool(*, tool_name=None, template_name=None, templates_dir=None, preprocess=None)

Register a template-backed tool with a minimal call signature.

The tool signature is intentionally small so we don't have to re-declare template keys/defaults in Python. Agents discover valid keys via the template mcp_doc section (plus the auto-generated key list).

Parameters:

Name	Type	Description	Default
tool_name	str | None	MCP tool name to register (required when templates_dir is not provided).	None
template_name	str | Path | None	Template filename or absolute path (required when templates_dir is not provided).	None
templates_dir	str | Path | None	If provided, register one tool per *.yaml file in this directory. Tool names are derived as construct_{template_stem}_study and preprocess hooks are resolved as self._prep_{template_stem} when present.	None
preprocess	Callable[[Dict[str, Any]], Dict[str, Any]] | None	Optional hook to validate/normalize override values before rendering.	None

Returns:

Type	Description
None	None

Raises:

Type	Description
ValueError	If arguments are inconsistent (e.g., both templates_dir and template_name are provided).
FileNotFoundError	If templates_dir does not exist.

Source code in mada_tools/workflow/weave/study_construction/server.py

def register_jinja_study_tool(
    self,
    *,
    tool_name: str | None = None,
    template_name: str | Path | None = None,
    templates_dir: str | Path | None = None,
    preprocess: Callable[[Dict[str, Any]], Dict[str, Any]] | None = None,
) -> None:
    """Register a template-backed tool with a minimal call signature.

    The tool signature is intentionally small so we don't have to re-declare
    template keys/defaults in Python. Agents discover valid keys via the
    template `mcp_doc` section (plus the auto-generated key list).

    Args:
        tool_name: MCP tool name to register (required when `templates_dir` is not provided).
        template_name: Template filename or absolute path (required when `templates_dir` is not provided).
        templates_dir: If provided, register one tool per `*.yaml` file in this directory.
            Tool names are derived as `construct_{template_stem}_study` and preprocess hooks are
            resolved as `self._prep_{template_stem}` when present.
        preprocess: Optional hook to validate/normalize override values before rendering.

    Returns:
        None

    Raises:
        ValueError: If arguments are inconsistent (e.g., both `templates_dir` and `template_name` are provided).
        FileNotFoundError: If `templates_dir` does not exist.
    """

    if templates_dir is not None:
        if tool_name is not None or template_name is not None or preprocess is not None:
            raise ValueError(
                "When `templates_dir` is provided, do not pass `tool_name`, `template_name`, or `preprocess`."
            )
        dir_path = templates_dir if isinstance(templates_dir, Path) else Path(str(templates_dir))
        if not dir_path.is_absolute():
            dir_path = self.study_constructor.study_templates_dir / dir_path
        if not dir_path.exists():
            raise FileNotFoundError(f"Templates directory '{dir_path}' does not exist.")

        for template_path in sorted(dir_path.glob("*.yaml")):
            stem = template_path.stem
            if stem.endswith("_study"):
                stem = stem[: -len("_study")]
            derived_tool_name = f"construct_{stem}_study"
            derived_preprocess = getattr(self, f"_prep_{stem}", None)
            self.register_jinja_study_tool(
                tool_name=derived_tool_name,
                template_name=template_path,
                preprocess=derived_preprocess,
            )
        return

    if tool_name is None or template_name is None:
        raise ValueError("Pass `tool_name` + `template_name`, or pass `templates_dir`.")

    async def _tool(
        overrides: Dict[str, Any] | None = None,
        output_dir: str | Path | None = None,
    ) -> str:
        """Render the registered Jinja study template and write a Maestro YAML."""
        context = await self.run_tool(
            self.study_constructor.build_context,
            template_name,
            overrides=overrides,
            preprocess=preprocess,
            background=False,
        )
        return await self.run_tool(
            self.study_constructor.write_yaml_tool,
            template_name,
            context,
            output_dir=output_dir,
            background=False,
        )

    _tool.__name__ = tool_name
    _tool.__doc__ = self.study_constructor.get_tool_doc_from_template(template_name)
    self.mcp.tool()(_tool)

register_study_construction_tools() abstractmethod

Hook for subclasses to register their own template-backed study tools.

Each tool defined here should utilize the self.register_jinja_study_tool(). Preprocessing can be added to each tool by creating a separate method in your class and linking your tool to it.

Example:

Here's a full example that defines a study construction tool with preprocessing.
In this example, we're targeting the creation of a study called `my_cool_study`
that needs a path `py3_utils` in order to run. As part of the preprocessing, we
coerce the path of `py3_utils` to ensure it's provided and exists.

```python
from mada_tools.workflow.weave import WEAVEStudyConstructionServer

class MyProjectStudyConstructionServer(WEAVEStudyConstructionServer):
    def _prep_my_cool_study(self, overrides: Dict[str, Any]) -> Dict[str, Any]:
        # Preprocess this study by validating a path that's needed
        raw_py3 = overrides.get("py3_utils")
        if raw_py3 is None:
            py3_utils_path = Path(__file__).parent / "py3utils"
        else:
            py3_utils_path = raw_py3 if isinstance(raw_py3, Path) else Path(str(raw_py3))
        if not py3_utils_path.exists():
            raise FileNotFoundError(f"The py3utils directory '{py3_utils_path}' does not exist.")
        overrides["py3_utils"] = str(py3_utils_path)
        return overrides

    def register_study_construction_tools(self):

        # Register the tool that constructs your cool study
        self.register_jinja_study_tool(
            tool_name="construct_my_cool_study",
            template_name="my_cool_study.yaml",
            preprocess=self._prep_my_cool_study,
        )
```

Source code in mada_tools/workflow/weave/study_construction/server.py

@abstractmethod
def register_study_construction_tools(self) -> None:
    """Hook for subclasses to register their own template-backed study tools.

    Each tool defined here should utilize the `self.register_jinja_study_tool()`.
    Preprocessing can be added to each tool by creating a separate method in your
    class and linking your tool to it.

    Example:

        Here's a full example that defines a study construction tool with preprocessing.
        In this example, we're targeting the creation of a study called `my_cool_study`
        that needs a path `py3_utils` in order to run. As part of the preprocessing, we
        coerce the path of `py3_utils` to ensure it's provided and exists.

        ```python
        from mada_tools.workflow.weave import WEAVEStudyConstructionServer

        class MyProjectStudyConstructionServer(WEAVEStudyConstructionServer):
            def _prep_my_cool_study(self, overrides: Dict[str, Any]) -> Dict[str, Any]:
                # Preprocess this study by validating a path that's needed
                raw_py3 = overrides.get("py3_utils")
                if raw_py3 is None:
                    py3_utils_path = Path(__file__).parent / "py3utils"
                else:
                    py3_utils_path = raw_py3 if isinstance(raw_py3, Path) else Path(str(raw_py3))
                if not py3_utils_path.exists():
                    raise FileNotFoundError(f"The py3utils directory '{py3_utils_path}' does not exist.")
                overrides["py3_utils"] = str(py3_utils_path)
                return overrides

            def register_study_construction_tools(self):

                # Register the tool that constructs your cool study
                self.register_jinja_study_tool(
                    tool_name="construct_my_cool_study",
                    template_name="my_cool_study.yaml",
                    preprocess=self._prep_my_cool_study,
                )
        ```
    """
    pass