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