Gradle MCP Server

Instructions

Implementation Reference

• src/gradle_mcp/server.py:381-432 (handler)

  MCP tool handler for the 'clean' tool. Delegates execution to GradleWrapper.clean while handling MCP context, progress, and logging.

  @mcp.tool()
  async def clean(
      project: str | None = None,
      ctx: Context | None = None,
  ) -> TaskResult:
      """Clean build artifacts for a Gradle project.
  
      Args:
          project: Project path (e.g., ':app').
                  Use None, empty string, or ':' for root project.
  
      Returns:
          TaskResult with success status and error message if failed.
      """
      try:
          if ctx:
              await ctx.info(f"Cleaning project: {project or 'root'}")
  
          gradle = _get_gradle_wrapper(ctx)
          daemon_monitor = _get_daemon_monitor()
  
          # clean now handles progress reporting internally by parsing Gradle output
          result = await gradle.clean(project, 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"Clean completed successfully for project {project or 'root'}")
              else:
                  await ctx.error(
                      f"Clean failed for project {project or 'root'}",
                      extra={"error": result.get("error")},
                  )
  
          return TaskResult(
              success=result["success"],
              error=result.get("error"),
          )
      except Exception as e:
          return TaskResult(
              success=False,
              error=ErrorInfo(
                  summary=str(e),
                  failed_tasks=[],
                  compilation_errors=[],
              ),
          )

• src/gradle_mcp/server.py:381-381 (registration)

  FastMCP decorator that registers the 'clean' function as an MCP tool.

  @mcp.tool()

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

  Pydantic model defining the output schema for the 'clean' tool (and other task tools). Contains success flag and optional structured error info.

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

• src/gradle_mcp/gradle.py:1069-1243 (helper)

  Core implementation in GradleWrapper that executes the actual Gradle clean task via subprocess, handles streaming output, progress, timeouts, error parsing, and dashboard logging.

  async def clean(
      self,
      project: str | None = None,
      ctx: "Context | None" = None,
      timeout: int | None = None,
      daemon_monitor: "DaemonMonitor | None" = None,
  ) -> dict:
      """Run the clean task for a project.
  
      Args:
          project: Project path (e.g., ':app'). Use ':' or empty string or None for root project.
          ctx: FastMCP context for progress reporting and logging.
          timeout: Optional timeout in seconds. If specified, the task will be
                   terminated if it exceeds this duration.
          daemon_monitor: Optional DaemonMonitor for logging to the dashboard.
  
      Returns:
          Dictionary with 'success', 'error' keys.
          - success (bool): True if clean completed successfully
          - error (str or None): Error message if clean failed, None otherwise
  
      Raises:
          subprocess.CalledProcessError: If Gradle command fails.
      """
      # Root project if project is None, empty, or ":"
      is_root = project is None or project == "" or project == ":"
      project_arg = "" if is_root else f"{project}:"
      # Build command - daemon is enabled by default
      # Use --console=plain for clean, parseable output without ANSI codes
      cmd = [str(self.wrapper_script), "--console=plain", f"{project_arg}clean"]
  
      logger.info(f"Executing: {' '.join(cmd)}")
  
      # Track this build in the dashboard
      build_id = None
      try:
          build_id = _log_store.start_build(f"{project_arg}clean")
      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"Clean 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"Clean timed out after {timeout} seconds", "ERROR")
              return {"success": False, "error": f"Clean 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():
                  # Use appropriate log level based on content, not just stderr
                  level = _classify_gradle_log_level(line, is_stderr=True)
                  if level == "ERROR":
                      logger.error(line)
                  elif level == "WARN":
                      logger.warning(line)
                  else:
                      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"Clean completed successfully for project {project or 'root'}")
              # 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, "Clean failed")
              logger.error(f"Clean 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"Clean 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/gradle.py:95-101 (schema)

  Pydantic model used in TaskResult.error for structured error reporting from Gradle clean (and other tasks). Includes failed tasks and 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