litmus-mcp-server

from config import logger
from starlette.requests import Request
from mcp.types import TextContent
from mcp.shared.exceptions import McpError
from mcp.types import ErrorData, INVALID_PARAMS
from litmussdk.marketplace import list_all_containers, run_container

from utils.auth import get_litmus_connection
from utils.formatting import format_success_response, format_error_response


async def get_all_containers_on_litmusedge(request: Request) -> list[TextContent]:
    """Lists all Docker containers running in the Litmus Edge marketplace."""
    try:
        connection = get_litmus_connection(request)
        container_list = list_all_containers(le_connection=connection)

        logger.info(f"Retrieved {len(container_list)} containers")

        result = {
            "count": len(container_list),
            "containers": container_list,
        }
        return format_success_response(result)

    except McpError:
        raise
    except Exception as e:
        logger.error(f"Error retrieving containers: {e}", exc_info=True)
        return format_error_response("retrieval_failed", str(e), count=0, containers=[])


async def run_docker_container_on_litmusedge(
    request: Request, arguments: dict
) -> list[TextContent]:
    """
    Deploys and runs a new Docker container on the Litmus Edge marketplace.

    SECURITY NOTE: Ensure the container image is trusted and command is validated.
    """
    try:
        docker_run_command = arguments.get("docker_run_command")

        if not docker_run_command:
            raise McpError(
                ErrorData(
                    code=INVALID_PARAMS,
                    message="'docker_run_command' parameter is required",
                )
            )

        connection = get_litmus_connection(request)
        result = run_container(docker_run_command, le_connection=connection)

        container_id = result.get("id", "Unknown container ID")

        logger.info(f"Deployed container: {container_id}")

        response = {
            "container_id": container_id,
            "command": docker_run_command,
            "result": result,
        }
        return format_success_response(response)

    except McpError:
        raise
    except Exception as e:
        logger.error(f"Error running container: {e}", exc_info=True)
        return format_error_response("deployment_failed", str(e))