Gradle MCP Server

Instructions

Implementation Reference

• src/gradle_mcp/server.py:315-379 (handler)

  MCP tool handler for 'run_task': decorated with @mcp.tool(), normalizes input tasks and arguments, creates GradleWrapper instance, calls underlying run_task implementation, handles logging and returns structured TaskResult

  @mcp.tool()
  async def run_task(
      task: str | list[str],
      args: list[str] | None = None,
      ctx: Context | None = None,
  ) -> TaskResult:
      """Run one or more Gradle tasks.
  
      Args:
          task: Task(s) to run. Single task, space-separated tasks, or list of tasks.
                Examples: 'build', ':app:build :core:build', [':core:build', ':app:assemble'].
          args: Optional Gradle arguments (e.g., ['--info', '-x', 'test']).
  
      Returns:
          TaskResult with success status and error message if failed.
      """
      try:
          # Normalize to list - handle space-separated string or list
          if isinstance(task, str):
              tasks = task.split()
          else:
              tasks = task
          task_str = ", ".join(tasks)
  
          if ctx:
              await ctx.info(f"Running task(s): {task_str}" + (f" with args: {args}" if args else ""))
  
          gradle = _get_gradle_wrapper(ctx)
          daemon_monitor = _get_daemon_monitor()
  
          # run_task handles multiple tasks and progress reporting
          result = await gradle.run_task(tasks, args, ctx, daemon_monitor=daemon_monitor)
  
          # Log Gradle output
          if ctx:
              if result.get("stdout"):
                  await ctx.debug(f"Gradle stdout:\n{result['stdout']}")
              if result.get("stderr"):
                  await ctx.debug(f"Gradle stderr:\n{result['stderr']}")
  
              if result["success"]:
                  await ctx.info(f"Task(s) {task_str} completed successfully")
              else:
                  await ctx.error(f"Task(s) {task_str} failed", extra={"error": result.get("error")})
  
          return TaskResult(
              success=result["success"],
              error=result.get("error"),
          )
      except ValueError as e:
          # Task is a cleaning task
          raise ValueError(str(e)) from e
      except Exception as e:
          # Report error progress
          if ctx:
              await ctx.report_progress(progress=100, total=100)
          return TaskResult(
              success=False,
              error=ErrorInfo(
                  summary=str(e),
                  failed_tasks=[],
                  compilation_errors=[],
              ),
          )

• src/gradle_mcp/gradle.py:886-1068 (handler)

  Core implementation of run_task in GradleWrapper: executes gradlew subprocess asynchronously, validates security/safety, streams output for progress/dashboard, parses structured errors (failed tasks, compilation errors), handles timeout

  async def run_task(
      self,
      tasks: str | list[str],
      args: list[str] | None = None,
      ctx: "Context | None" = None,
      timeout: int | None = None,
      daemon_monitor: "DaemonMonitor | None" = None,
  ) -> dict:
      """Run one or more Gradle tasks with real-time progress reporting.
  
      Args:
          tasks: Task(s) to run. Single string or list of task names.
          args: Additional Gradle arguments. Daemon is enabled by default.
          ctx: Optional FastMCP Context for progress reporting.
          timeout: Optional timeout in seconds. If specified, the task will be
                   terminated if it exceeds this duration.
  
      Returns:
          Dictionary with 'success' (bool) and 'error' (str or None).
  
      Raises:
          ValueError: If any task is a cleaning task.
      """
      # Normalize to list
      task_list = tasks if isinstance(tasks, list) else [tasks]
  
      # Check all tasks for cleaning patterns
      for task in task_list:
          if self._is_cleaning_task(task):
              raise ValueError(
                  f"Task '{task}' is a cleaning task and cannot be run via run_task. "
                  "Please use the clean tool instead."
              )
  
      # Validate arguments to prevent command injection
      if args:
          self._validate_gradle_args(args, task_list)
  
      # Build command with all tasks - daemon is enabled by default in Gradle
      # Use --console=plain for clean, parseable output without ANSI codes
      cmd = [str(self.wrapper_script), "--console=plain"] + task_list
  
      if args:
          cmd.extend(args)
  
      logger.info(f"Executing: {' '.join(cmd)}")
  
      # Track this build in the dashboard
      build_id = None
      try:
          build_id = _log_store.start_build(" ".join(task_list))
      except Exception:
          pass
  
      process: asyncio.subprocess.Process | None = None
      try:
          process = await asyncio.create_subprocess_exec(
              *cmd,
              cwd=str(self.project_root),
              stdout=asyncio.subprocess.PIPE,
              stderr=asyncio.subprocess.PIPE,
              env=self._build_execution_environment(),
          )
  
          # Notify dashboard that a task started (may have spawned a new daemon)
          if _emit_daemons_changed:
              try:
                  _emit_daemons_changed()
              except Exception:
                  pass
  
          # Stream output in real-time while collecting it
          stdout_lines: list[str] = []
          stderr_lines: list[str] = []
  
          async def read_stream(
              stream: asyncio.StreamReader, lines: list[str], is_stderr: bool = False
          ) -> None:
              while True:
                  line = await stream.readline()
                  if not line:
                      break
                  decoded = line.decode().rstrip()
                  lines.append(decoded)
                  # Log to shared log_store for dashboard
                  if _log_store and decoded.strip():
                      level = _classify_gradle_log_level(decoded, is_stderr)
                      _log_store.add_log(decoded, level)
  
          # Handle timeout with streaming
          try:
              if timeout:
                  await asyncio.wait_for(
                      asyncio.gather(
                          read_stream(process.stdout, stdout_lines, False),
                          read_stream(process.stderr, stderr_lines, True),
                      ),
                      timeout=timeout,
                  )
              else:
                  await asyncio.gather(
                      read_stream(process.stdout, stdout_lines, False),
                      read_stream(process.stderr, stderr_lines, True),
                  )
              await process.wait()
          except asyncio.TimeoutError:
              # Graceful termination first
              logger.warning(f"Task timed out after {timeout} seconds, terminating process...")
              process.terminate()
              try:
                  await asyncio.wait_for(process.wait(), timeout=5)
              except asyncio.TimeoutError:
                  # Force kill if graceful termination fails
                  logger.warning("Process did not terminate gracefully, killing...")
                  process.kill()
                  await process.wait()
              if daemon_monitor:
                  daemon_monitor.add_log("gradle", f"Task timed out after {timeout} seconds", "ERROR")
              return {"success": False, "error": f"Task timed out after {timeout} seconds"}
  
          stdout = "\n".join(stdout_lines)
          stderr = "\n".join(stderr_lines)
  
          # Log output for debugging
          for line in stdout_lines:
              if line.strip():
                  logger.debug(line)
          for line in stderr_lines:
              if line.strip():
                  logger.debug(line)
  
          # Report progress if context is available (final progress)
          if ctx:
              await ctx.report_progress(progress=100, total=100)
  
          if process.returncode == 0:
              logger.info(f"Task {task_list} completed successfully")
              # End build tracking
              if build_id and _log_store:
                  try:
                      _log_store.end_build(build_id)
                  except Exception:
                      pass
              return {"success": True, "error": None}
          else:
              # Extract structured error information
              error_info = self._extract_structured_error(stdout, stderr, "Task failed")
              logger.error(f"Task {task_list} failed: {error_info.summary}")
              # End build tracking
              if build_id and _log_store:
                  try:
                      _log_store.end_build(build_id)
                  except Exception:
                      pass
              return {"success": False, "error": error_info}
  
      except Exception as e:
          # Ensure process cleanup on any exception
          if process and process.returncode is None:
              logger.warning(f"Exception occurred, cleaning up process: {e}")
              process.terminate()
              try:
                  await asyncio.wait_for(process.wait(), timeout=5)
              except asyncio.TimeoutError:
                  process.kill()
                  await process.wait()
          logger.error(f"Task {task_list} failed with exception: {e}")
          # End build tracking
          if build_id and _log_store:
              try:
                  _log_store.end_build(build_id)
              except Exception:
                  pass
          # Return structured error for exceptions
          return {
              "success": False,
              "error": ErrorInfo(
                  summary=str(e),
                  failed_tasks=[],
                  compilation_errors=[],
              ),
          }

• src/gradle_mcp/server.py:117-122 (schema)

  Pydantic output schema for the run_task tool: indicates success and optional structured error details

  class TaskResult(BaseModel):
      """Result of running a Gradle task."""
  
      success: bool
      error: ErrorInfo | None = None

• src/gradle_mcp/gradle.py:95-101 (schema)

  Core error schema used in TaskResult: includes build failure summary, list of failed tasks, and parsed compilation errors

  class ErrorInfo(BaseModel):
      """Structured error information."""
  
      summary: str  # e.g., "Build failed: 2 tasks failed with 12 compilation errors"
      failed_tasks: list[FailedTask]  # List of failed tasks
      compilation_errors: list[CompilationError]  # Deduplicated, first occurrence only

• src/gradle_mcp/gradle.py:79-86 (schema)

  Schema for individual compilation errors parsed from Gradle output

  class CompilationError(BaseModel):
      """A single compilation error."""
  
      file: str  # Full path without file:// prefix
      line: int
      column: int | None
      message: str

• src/gradle_mcp/gradle.py:88-93 (schema)

  Schema for failed Gradle tasks parsed from output

  class FailedTask(BaseModel):
      """Information about a failed task."""
  
      name: str  # e.g., ":composeApp:compileTestKotlinIosArm64"
      reason: str  # e.g., "Compilation finished with errors"