Pipelines/docker_pipeline.py

import os
import requests
from typing import Literal, List, Optional


from blueprints.function_calling_blueprint import Pipeline as FunctionCallingBlueprint


class Pipeline(FunctionCallingBlueprint):
    class Tools:
        def __init__(self, pipeline) -> None:
            self.pipeline = pipeline

        def call_docker(
            self,image: str,command: Optional[str] = None):
            """
            Run a Docker container tool with the specified image and an optional command.
            :param image: The Docker image name to run
            :param command: Optional command to run in the container (default: None)
            :return: The output from running the Docker container
            """
            docker_cmd = f"docker run {image}"
            if command:
                docker_cmd += f" {command}"
            return os.system(docker_cmd)

    def __init__(self):
        self.valves = self.Valves(
            **{
                "pipelines": ["pipeline-testmodel"],  # Connect to test pipeline
            }
        )
        self.tools = self.Tools(self)
        # Optionally, you can set the id and name of the pipeline.
        # Best practice is to not specify the id so that it can be automatically inferred from the filename, so that users can install multiple versions of the same pipeline.
        # The identifier must be unique across all pipelines.
        # The identifier must be an alphanumeric string that can include underscores or hyphens. It cannot contain spaces, special characters, slashes, or backslashes.
        # self.id = "my_tools_pipeline"