Skip to main content
Glama

DARPEngine

by DARPAI
docker_service.py4.27 kB
import subprocess from pathlib import Path from .constants import BaseImage from .dockerfile_generators import DockerfileGenerator from .dockerfile_generators import get_dockerfile_generator from .errors import InvalidData from .errors import ProcessingError from .user_logger import UserLogger from registry.src.database import Server from registry.src.logger import logger from registry.src.settings import settings class DockerService: def __init__( self, docker_env: dict, repo_folder: str, server: Server, user_logger: UserLogger, ) -> None: self.docker_env = docker_env self.repo_folder = repo_folder self.server = server self.image_name = f"hipasus/{self.server.name}" self.container_name = f"darp_server_{server.id}" self.user_logger = user_logger async def clone_repo(self) -> None: await self._execute(["git", "clone", self.server.repo_url, self.repo_folder]) def dockerfile_exists(self) -> bool: path = Path(f"{self.repo_folder}/Dockerfile") return path.exists() async def generate_dockerfile(self) -> None: if not self.server.base_image: await self.user_logger.error(message="No base image provided.") raise InvalidData generator: DockerfileGenerator = await get_dockerfile_generator( repo_folder=self.repo_folder, base_image=BaseImage(self.server.base_image), build_instructions=self.server.build_instructions, user_logger=self.user_logger, ) try: logs = await generator.generate_dockerfile() await self.user_logger.info(message=f"Generated Dockerfile:\n{logs}") logger.info(logs) except ProcessingError as error: await self.user_logger.error( message=f"Error generating Dockerfile: {error.message or ''}" ) raise async def build_image(self) -> None: command = [ "docker", "build", "-t", self.image_name, "-f", f"{self.repo_folder}/Dockerfile", self.repo_folder, ] await self._execute(command) async def push_image(self) -> None: docker_push_command = ( f"docker login -u {settings.dockerhub_login} -p {settings.dockerhub_password} && docker image push {self.image_name}", ) await self.user_logger.info("Pushing image...") result = subprocess.run(docker_push_command, shell=True) if result.returncode != 0: await self.user_logger.error(message="Docker push failed") logger.error("docker push failed") raise ProcessingError async def run_container(self) -> None: subprocess.run( ["docker", "rm", self.container_name, "--force"], env=self.docker_env ) run_command = [ "docker", "run", "--network", settings.deployment_network, "-d", "--restart", "always", "--pull", "always", "--name", self.container_name, self.image_name, ] if self.server.command: run_command.extend(["sh", "-c", self.server.command]) await self._execute(command=run_command, env=self.docker_env) async def _execute(self, command: list[str], env: dict | None = None) -> None: command_str = " ".join(command) logger.info(f"{command_str=}") await self.user_logger.command_start(command=command_str) arguments: dict = dict( args=command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, ) if env: arguments["env"] = env result = subprocess.run(**arguments) output = result.stdout.decode() if result.stdout else "" # noqa if result.returncode != 0: await self.user_logger.command_result(output=output, success=False) logger.error(f"{' '.join(command[:2])} failed") raise ProcessingError await self.user_logger.command_result(output=output, success=True)

MCP directory API

We provide all the information about MCP servers via our MCP API.

curl -X GET 'https://glama.ai/api/mcp/v1/servers/DARPAI/darp_engine'

If you have feedback or need assistance with the MCP directory API, please join our Discord server