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"