✨ MCP Service Containerization Integration #1368

✨ MCP Service Containerization Integration #1368

Author: WMC001

backend/apps/remote_mcp_app.py

@@ -1,11 +1,11 @@
 import logging
 from typing import Optional
 
-from fastapi import APIRouter, Header, HTTPException
+from fastapi import APIRouter, Header, HTTPException, UploadFile, File, Form
 from fastapi.responses import JSONResponse
 from http import HTTPStatus
 
-from consts.const import NEXENT_MCP_DOCKER_IMAGE
+from consts.const import NEXENT_MCP_DOCKER_IMAGE, ENABLE_UPLOAD_IMAGE
 from consts.exceptions import MCPConnectionError, MCPNameIllegal, MCPContainerError
 from consts.model import MCPConfigRequest
 from services.remote_mcp_service import (
@@ -14,7 +14,9 @@
     get_remote_mcp_server_list,
     check_mcp_health_and_update_db,
     delete_mcp_by_container_id,
+    upload_and_start_mcp_image,
 )
+from database.remote_mcp_db import check_mcp_name_exists
 from services.tool_configuration_service import get_tool_from_remote_mcp_server
 from services.mcp_container_service import MCPContainerManager
 from utils.auth_utils import get_current_user_id
@@ -116,6 +118,7 @@ async def get_remote_proxies(
         return JSONResponse(
             status_code=HTTPStatus.OK,
             content={"remote_mcp_server_list": remote_mcp_server_list,
+                     "enable_upload_image": ENABLE_UPLOAD_IMAGE,
                      "status": "success"}
         )
     except Exception as e:
@@ -196,6 +199,11 @@ async def add_mcp_from_config(
                     errors.append(f"{service_name}: port is required")
                     continue
 
+                # Check if MCP service name already exists before starting container
+                if check_mcp_name_exists(mcp_name=service_name, tenant_id=tenant_id):
+                    errors.append(f"{service_name}: MCP name already exists")
+                    continue
+
                 # Build full command to run inside nexent/nexent-mcp image
                 full_command = [
                     "python",
@@ -224,22 +232,13 @@ async def add_mcp_from_config(
                 )
 
                 # Register to remote MCP server list
-                try:
-                    await add_remote_mcp_server_list(
-                        tenant_id=tenant_id,
-                        user_id=user_id,
-                        remote_mcp_server=container_info["mcp_url"],
-                        remote_mcp_server_name=service_name,
-                        container_id=container_info["container_id"],
-                    )
-                except MCPNameIllegal:
-                    # If name already exists, try to stop the container we just created
-                    try:
-                        await container_manager.stop_mcp_container(container_info["container_id"])
-                    except Exception:
-                        pass
-                    errors.append(f"{service_name}: MCP name already exists")
-                    continue
+                await add_remote_mcp_server_list(
+                    tenant_id=tenant_id,
+                    user_id=user_id,
+                    remote_mcp_server=container_info["mcp_url"],
+                    remote_mcp_server_name=service_name,
+                    container_id=container_info["container_id"],
+                )
 
                 results.append({
                     "service_name": service_name,
@@ -251,10 +250,12 @@ async def add_mcp_from_config(
                 })
 
             except MCPContainerError as e:
-                logger.error(f"Failed to start MCP container {service_name}: {e}")
+                logger.error(
+                    f"Failed to start MCP container {service_name}: {e}")
                 errors.append(f"{service_name}: {str(e)}")
             except Exception as e:
-                logger.error(f"Unexpected error adding MCP {service_name}: {e}")
+                logger.error(
+                    f"Unexpected error adding MCP {service_name}: {e}")
                 errors.append(f"{service_name}: {str(e)}")
 
         if errors and not results:
@@ -404,3 +405,62 @@ async def get_container_logs(
             status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
             detail=f"Failed to get container logs: {str(e)}"
         )
+
+
+# Conditionally add upload-image route based on ENABLE_UPLOAD_IMAGE setting
+if ENABLE_UPLOAD_IMAGE:
+    @router.post("/upload-image")
+    async def upload_mcp_image(
+        file: UploadFile = File(..., description="Docker image tar file"),
+        port: int = Form(..., ge=1, le=65535,
+                         description="Host port to expose the MCP server on (1-65535)"),
+        service_name: Optional[str] = Form(
+            None, description="Name for the MCP service (auto-generated if not provided)"),
+        env_vars: Optional[str] = Form(
+            None, description="Environment variables as JSON string"),
+        authorization: Optional[str] = Header(None)
+    ):
+        """
+        Upload Docker image tar file and start MCP container.
+
+        Container naming: {filename-without-extension}-{tenant-id[:8]}-{user-id[:8]}
+        """
+        try:
+            user_id, tenant_id = get_current_user_id(authorization)
+
+            # Read file content
+            content = await file.read()
+
+            # Call service layer to handle the business logic
+            result = await upload_and_start_mcp_image(
+                tenant_id=tenant_id,
+                user_id=user_id,
+                file_content=content,
+                filename=file.filename,
+                port=port,
+                service_name=service_name,
+                env_vars=env_vars,
+            )
+
+            return JSONResponse(status_code=HTTPStatus.OK, content=result)
+
+        except ValueError as e:
+            logger.error(f"Validation error: {e}")
+            raise HTTPException(
+                status_code=HTTPStatus.BAD_REQUEST, detail=str(e))
+        except MCPNameIllegal as e:
+            logger.error(f"MCP name conflict: {e}")
+            raise HTTPException(status_code=HTTPStatus.CONFLICT, detail=str(e))
+        except MCPContainerError as e:
+            logger.error(f"Container error: {e}")
+            raise HTTPException(
+                status_code=HTTPStatus.SERVICE_UNAVAILABLE, detail=str(e))
+        except Exception as e:
+            logger.error(f"Failed to upload and start MCP container: {e}")
+            raise HTTPException(
+                status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
+                detail=f"Failed to upload and start MCP container: {str(e)}"
+            )
+else:
+    logger.info(
+        "MCP image upload feature is disabled (ENABLE_UPLOAD_IMAGE=false)")

backend/consts/const.py

@@ -130,6 +130,7 @@ class VectorDatabaseType(str, Enum):
     "DISABLE_CELERY_FLOWER", "false").lower() == "true"
 DOCKER_ENVIRONMENT = os.getenv("DOCKER_ENVIRONMENT", "false").lower() == "true"
 NEXENT_MCP_DOCKER_IMAGE = os.getenv("NEXENT_MCP_DOCKER_IMAGE", "nexent/nexent-mcp:latest")
+ENABLE_UPLOAD_IMAGE = os.getenv("ENABLE_UPLOAD_IMAGE", "false").lower() == "true"
 
 
 # Celery Configuration

backend/consts/model.py

@@ -232,8 +232,10 @@ class GeneratePromptRequest(BaseModel):
     task_description: str
     agent_id: int
     model_id: int
-    tool_ids: Optional[List[int]] = None  # Optional: tool IDs from frontend (takes precedence over database query)
-    sub_agent_ids: Optional[List[int]] = None  # Optional: sub-agent IDs from frontend (takes precedence over database query)
+    tool_ids: Optional[List[int]] = Field(
+        None, description="Optional: tool IDs from frontend (takes precedence over database query)")
+    sub_agent_ids: Optional[List[int]] = Field(
+        None, description="Optional: sub-agent IDs from frontend (takes precedence over database query)")
 
 
 class GenerateTitleRequest(BaseModel):
@@ -399,19 +401,22 @@ def default(cls) -> "MemoryAgentShareMode":
 # ---------------------------------------------------------------------------
 class VoiceConnectivityRequest(BaseModel):
     """Request model for voice service connectivity check"""
-    model_type: str = Field(..., description="Type of model to check ('stt' or 'tts')")
+    model_type: str = Field(...,
+                            description="Type of model to check ('stt' or 'tts')")
 
 
 class VoiceConnectivityResponse(BaseModel):
     """Response model for voice service connectivity check"""
-    connected: bool = Field(..., description="Whether the service is connected")
+    connected: bool = Field(...,
+                            description="Whether the service is connected")
     model_type: str = Field(..., description="Type of model checked")
     message: str = Field(..., description="Status message")
 
 
 class TTSRequest(BaseModel):
     """Request model for TTS text-to-speech conversion"""
-    text: str = Field(..., min_length=1, description="Text to convert to speech")
+    text: str = Field(..., min_length=1,
+                      description="Text to convert to speech")
     stream: bool = Field(True, description="Whether to stream the audio")
 
 
@@ -435,7 +440,8 @@ class ToolValidateRequest(BaseModel):
 class MCPServerConfig(BaseModel):
     """Configuration for a single MCP server"""
     command: str = Field(..., description="Command to run (e.g., 'npx')")
-    args: List[str] = Field(default_factory=list, description="Command arguments")
+    args: List[str] = Field(default_factory=list,
+                            description="Command arguments")
     env: Optional[Dict[str, str]] = Field(
         None, description="Environment variables for the MCP server")
     port: Optional[int] = Field(
@@ -449,4 +455,4 @@ class MCPServerConfig(BaseModel):
 class MCPConfigRequest(BaseModel):
     """Request model for adding MCP servers from configuration"""
     mcpServers: Dict[str, MCPServerConfig] = Field(
-        ..., description="Dictionary of MCP server configurations")
+        ..., description="Dictionary of MCP server configurations")

backend/services/mcp_container_service.py

@@ -42,11 +42,44 @@ def __init__(self, docker_socket_path: Optional[str] = None):
             )
             # Create container client from config
             self.client = create_container_client_from_config(config)
-            logger.info("MCPContainerManager initialized using SDK container module")
+            logger.info(
+                "MCPContainerManager initialized using SDK container module")
         except ContainerError as e:
             logger.error(f"Failed to initialize container manager: {e}")
             raise MCPContainerError(f"Cannot connect to Docker: {e}")
 
+    async def load_image_from_tar_file(self, tar_file_path: str) -> str:
+        """
+        Load Docker image from tar file
+
+        Args:
+            tar_file_path: Path to the tar file containing the Docker image
+
+        Returns:
+            Image name/tag that was loaded
+
+        Raises:
+            MCPContainerError: If image loading fails
+        """
+        try:
+            # Load image from tar file
+            with open(tar_file_path, 'rb') as tar_file:
+                images = self.client.client.images.load(tar_file.read())
+
+            if not images:
+                raise MCPContainerError("No images found in tar file")
+
+            # Get the first loaded image
+            loaded_image = images[0]
+            image_name = loaded_image.tags[0] if loaded_image.tags else str(
+                loaded_image.id)
+
+        except Exception as e:
+            logger.error(f"Failed to load image from tar file: {e}")
+            raise MCPContainerError(f"Failed to load image from tar file: {e}")
+        logger.info(f"Successfully loaded image: {image_name}")
+        return image_name
+
     async def start_mcp_container(
         self,
         service_name: str,
@@ -73,8 +106,6 @@ async def start_mcp_container(
             MCPContainerError: If container startup fails
         """
         try:
-            if not full_command:
-                raise MCPContainerError("full_command is required to start MCP container")
             result = await self.client.start_container(
                 service_name=service_name,
                 tenant_id=tenant_id,
@@ -87,7 +118,8 @@ async def start_mcp_container(
             # Map SDK response to existing interface (mcp_url instead of service_url)
             return {
                 "container_id": result["container_id"],
-                "mcp_url": result["service_url"],  # Map service_url to mcp_url for compatibility
+                # Map service_url to mcp_url for compatibility
+                "mcp_url": result["service_url"],
                 "host_port": result["host_port"],
                 "status": result["status"],
                 "container_name": result.get("container_name"),
@@ -99,6 +131,54 @@ async def start_mcp_container(
             logger.error(f"MCP connection error: {e}")
             raise MCPConnectionError(f"MCP connection failed: {e}")
 
+    async def start_mcp_container_from_tar(
+        self,
+        tar_file_path: str,
+        service_name: str,
+        tenant_id: str,
+        user_id: str,
+        env_vars: Optional[Dict[str, str]] = None,
+        host_port: Optional[int] = None,
+        full_command: Optional[List[str]] = None,
+    ) -> Dict[str, str]:
+        """
+        Load image from tar file and start MCP container
+
+        Args:
+            tar_file_path: Path to the tar file containing the Docker image
+            service_name: Name of the MCP service
+            tenant_id: Tenant ID for isolation
+            user_id: User ID for isolation
+            env_vars: Optional environment variables
+            host_port: Optional host port to bind
+            full_command: Optional command to run in container
+
+        Returns:
+            Dictionary with container_id, mcp_url, host_port, and status
+
+        Raises:
+            MCPContainerError: If container startup fails
+        """
+        try:
+            # Load image from tar file
+            image_name = await self.load_image_from_tar_file(tar_file_path)
+
+            # Start container with the loaded image
+            return await self.start_mcp_container(
+                service_name=service_name,
+                tenant_id=tenant_id,
+                user_id=user_id,
+                env_vars=env_vars,
+                host_port=host_port,
+                image=image_name,
+                full_command=full_command,
+            )
+
+        except Exception as e:
+            logger.error(f"Failed to start MCP container from tar file: {e}")
+            raise MCPContainerError(
+                f"Failed to start container from tar file: {e}")
+
     async def stop_mcp_container(self, container_id: str) -> bool:
         """
         Stop and remove MCP container
@@ -117,7 +197,7 @@ async def stop_mcp_container(self, container_id: str) -> bool:
             stop_result = await self.client.stop_container(container_id)
             if not stop_result:
                 return False
-            
+
             # Then remove the container
             remove_result = await self.client.remove_container(container_id)
             return remove_result