"""MCP Server for running Gradle tasks."""
import os
from pathlib import Path
from fastmcp import Context, FastMCP
from pydantic import BaseModel
from gradle_mcp.dashboard import DaemonMonitor, DashboardServer
from gradle_mcp.gradle import GradleWrapper
# Re-export error models for external use
from gradle_mcp.gradle import CompilationError, FailedTask, ErrorInfo # noqa: F401
# MCP Server instructions for AI agents
MCP_INSTRUCTIONS = """
# *gradle-mcp*
This MCP server provides tools to interact with Gradle projects without using command-line operations.
Use these tools instead of running `./gradlew` or `gradle` commands directly in the terminal.
## Usage Guidelines
- use `list_projects` to explore available projects
- use `list_project_tasks` to see tasks before running them
- Use `run_task` for building, testing, assembling, etc.
- Always use `clean` tool for cleaning operations, never `run_task` with clean tasks
- Use qualified task names (e.g., ':app:build') when targeting specific subprojects
- Pass '--info' or '--debug' in args for verbose output when troubleshooting
## Available Tools
### Project & Task Management
- `list_projects`: List all Gradle projects in the workspace
- `list_project_tasks`: List available tasks for a project (with optional filtering by group)
- `run_task`: Execute one or more Gradle tasks
- `clean`: Clean build artifacts for a project
### Daemon Management
- `daemon_status`: Check the status of running Gradle daemons (memory usage, state, etc.)
- `stop_daemon`: Stop all Gradle daemons (useful for freeing memory or resolving daemon issues)
### Configuration
- `get_gradle_config`: View current Gradle configuration including JVM memory settings and environment
## Troubleshooting Memory Issues
If builds are slow, hanging, or running out of memory:
1. Use `get_gradle_config` to check if JVM args (Xmx, Xms) are properly configured
2. Use `daemon_status` to see running daemons and their memory consumption
3. Use `stop_daemon` to kill all daemons and free up memory
4. Check the web dashboard for real-time visibility into daemon activity
Common memory-related settings in gradle.properties:
- `org.gradle.jvmargs=-Xmx2048m -XX:+HeapDumpOnOutOfMemoryError`
- `org.gradle.daemon=true` (daemon reuses JVM, improving performance)
## Web Dashboard
The MCP server includes a web dashboard for monitoring Gradle activity:
- **Automatic startup**: The dashboard starts automatically when the MCP server runs
- **Access URL**: http://localhost:3333 (or next available port if 3333 is in use)
- **Features**:
- Real-time daemon status monitoring
- Active task tracking
- Build log viewing
- Memory usage visualization
Use the dashboard for continuous monitoring during long build sessions.
## Environment Variables
The following environment variables are supported:
- `GRADLE_PROJECT_ROOT`: Root directory of the Gradle project (default: current directory)
- `GRADLE_WRAPPER`: Custom path to Gradle wrapper script (auto-detected if not set)
- `GRADLE_OPTS`: JVM options for the Gradle client (passed to Gradle automatically)
- `JAVA_OPTS`: General Java options (may affect Gradle execution)
"""
# Initialize FastMCP server
mcp = FastMCP(name="gradle-mcp", instructions=MCP_INSTRUCTIONS)
# Global dashboard components
_dashboard_server: DashboardServer | None = None
_daemon_monitor: DaemonMonitor | None = None
def _get_daemon_monitor() -> DaemonMonitor:
"""Get or create the global DaemonMonitor instance.
Returns:
DaemonMonitor instance.
"""
global _daemon_monitor
if _daemon_monitor is None:
project_root = os.getenv("GRADLE_PROJECT_ROOT") or os.getcwd()
_daemon_monitor = DaemonMonitor(project_root)
return _daemon_monitor
def _start_dashboard() -> str:
"""Start the dashboard server if not already running.
Returns:
URL where the dashboard is running.
"""
global _dashboard_server
if _dashboard_server is None or not _dashboard_server.is_running:
daemon_monitor = _get_daemon_monitor()
_dashboard_server = DashboardServer(daemon_monitor)
url = _dashboard_server.start()
return url
return _dashboard_server.url or "Dashboard already running"
class TaskResult(BaseModel):
"""Result of running a Gradle task."""
success: bool
error: ErrorInfo | None = None
class DaemonStatusResult(BaseModel):
"""Result of daemon status query."""
running: bool
daemons: list[dict] # List of daemon info
error: str | None = None
class GradleConfigResult(BaseModel):
"""Result of Gradle configuration query."""
jvm_args: str | None
daemon_enabled: bool | None
parallel_enabled: bool | None
caching_enabled: bool | None
max_workers: int | None
distribution_url: str | None
gradle_version: str | None
class ProjectInfo(BaseModel):
"""Information about a Gradle project."""
name: str
path: str
description: str | None = None
class TaskInfo(BaseModel):
"""Information about a Gradle task."""
name: str
project: str
description: str | None = None
group: str | None = None
class TaskWithDescriptionInfo(BaseModel):
"""Task with its description."""
name: str
description: str
class GroupedTasksInfo(BaseModel):
"""Tasks grouped by their group name.
When include_descriptions is True, tasks contains TaskWithDescriptionInfo objects.
When include_descriptions is False, tasks contains task name strings.
"""
group: str
tasks: list[TaskWithDescriptionInfo] | list[str]
def _get_gradle_wrapper(ctx: Context | None = None) -> GradleWrapper:
"""Get a GradleWrapper instance, optionally using context to determine project root.
Respects the following environment variables:
- GRADLE_PROJECT_ROOT: Root directory of Gradle project (default: current directory)
- GRADLE_WRAPPER: Path to Gradle wrapper script (optional, auto-detected if not set)
- GRADLE_OPTS: JVM options for the Gradle client (logged for visibility)
- JAVA_OPTS: General Java options (logged for visibility)
Args:
ctx: MCP Context (optional).
Returns:
GradleWrapper instance.
Raises:
FileNotFoundError: If Gradle wrapper cannot be found.
"""
import logging
logger = logging.getLogger(__name__)
project_root = os.getenv("GRADLE_PROJECT_ROOT") or os.getcwd()
wrapper_path = os.getenv("GRADLE_WRAPPER")
# Log environment variables for debugging visibility
gradle_opts = os.getenv("GRADLE_OPTS")
java_opts = os.getenv("JAVA_OPTS")
if gradle_opts:
logger.debug(f"GRADLE_OPTS is set: {gradle_opts}")
if java_opts:
logger.debug(f"JAVA_OPTS is set: {java_opts}")
if wrapper_path:
logger.debug(f"GRADLE_WRAPPER is set: {wrapper_path}")
logger.debug(f"Using project root: {project_root}")
gradle = GradleWrapper(project_root)
# If a custom wrapper path is specified, override the auto-detected one
if wrapper_path:
wrapper_path_obj = Path(wrapper_path)
if not wrapper_path_obj.exists():
raise FileNotFoundError(
f"Gradle wrapper not found at specified path: {wrapper_path}. "
"Please verify GRADLE_WRAPPER environment variable."
)
gradle.wrapper_script = wrapper_path_obj
return gradle
@mcp.tool()
async def list_projects(ctx: Context) -> list[ProjectInfo]:
"""List all Gradle projects in the workspace.
Returns:
List of Gradle projects.
"""
try:
await ctx.info("Listing all Gradle projects")
gradle = _get_gradle_wrapper(ctx)
projects = gradle.list_projects()
await ctx.info(f"Found {len(projects)} projects: {', '.join(p.name for p in projects)}")
return [
ProjectInfo(
name=p.name,
path=p.path,
description=p.description,
)
for p in projects
]
except Exception as e:
raise ValueError(f"Failed to list projects: {str(e)}") from e
@mcp.tool()
async def list_project_tasks(
project: str | None = None,
include_descriptions: bool = False,
group: str | None = None,
ctx: Context | None = None,
) -> list[GroupedTasksInfo]:
"""List all tasks available in a Gradle project.
Returns a nested structure grouped by task group. This is more compact
than a flat list and avoids repeating project/group information for each task.
Args:
project: Project path (e.g., ':app' or 'lib:module').
Use None, empty string, or ':' for root project.
include_descriptions: If True, include task descriptions in the response.
If False, return only task names for a more compact response.
group: Optional group name to filter tasks (e.g., 'Build', 'Verification').
Case-insensitive. If not provided, all groups are returned.
Returns:
List of grouped tasks. Each group contains a group name and list of tasks.
If include_descriptions is True, tasks include name and description.
If include_descriptions is False, tasks are just name strings.
"""
try:
if ctx:
await ctx.info(f"Listing tasks for project: {project or 'root'}")
gradle = _get_gradle_wrapper(ctx)
# Normalize root project: None, empty, or ":" all mean root
project_arg = project if project and project != "" else ":"
grouped_tasks = gradle.list_tasks(project_arg, include_descriptions, group)
if ctx:
total_tasks = sum(len(g.tasks) for g in grouped_tasks)
group_info = f" (filtered by group '{group}')" if group else ""
await ctx.info(
f"Found {total_tasks} tasks in {len(grouped_tasks)} groups for project {project_arg}{group_info}"
)
# Convert to response model
result = []
for group in grouped_tasks:
if include_descriptions:
# Tasks are TaskWithDescription objects
tasks = [
TaskWithDescriptionInfo(name=t.name, description=t.description)
for t in group.tasks
]
else:
# Tasks are already strings
tasks = group.tasks
result.append(GroupedTasksInfo(group=group.group, tasks=tasks))
return result
except Exception as e:
raise ValueError(f"Failed to list tasks: {str(e)}") from e
@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=[],
),
)
@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=[],
),
)
@mcp.tool()
async def daemon_status(ctx: Context | None = None) -> DaemonStatusResult:
"""Get status of Gradle daemon(s).
Returns current daemon status including running daemons and their info.
Useful for monitoring daemon health and resource usage.
Returns:
DaemonStatusResult with running status, list of daemon info, and optional error.
"""
try:
if ctx:
await ctx.info("Getting Gradle daemon status")
gradle = _get_gradle_wrapper(ctx)
result = await gradle.daemon_status()
if ctx:
if result["success"]:
await ctx.info("Retrieved daemon status successfully")
else:
await ctx.error("Failed to get daemon status", extra={"error": result.get("error")})
# Parse daemon output to extract daemon info
daemons: list[dict] = []
running = False
output = result.get("output", "")
if result["success"] and output:
# Parse lines looking for daemon entries
# Format: " 12345 IDLE 8.5"
for line in output.split("\n"):
line = line.strip()
# Skip header and empty lines
if not line or line.startswith("PID") or line.startswith("-"):
continue
# Try to parse daemon status line
parts = line.split()
if len(parts) >= 2 and parts[0].isdigit():
daemon_info = {
"pid": parts[0],
"status": parts[1] if len(parts) > 1 else "UNKNOWN",
}
if len(parts) > 2:
daemon_info["info"] = " ".join(parts[2:])
daemons.append(daemon_info)
running = True
return DaemonStatusResult(
running=running,
daemons=daemons,
error=result.get("error"),
)
except Exception as e:
return DaemonStatusResult(
running=False,
daemons=[],
error=str(e),
)
@mcp.tool()
async def stop_daemon(ctx: Context | None = None) -> TaskResult:
"""Stop all Gradle daemons.
Stops all running Gradle daemons. Useful for freeing up memory
or when experiencing daemon-related issues.
Returns:
TaskResult with success status and error message if failed.
"""
try:
if ctx:
await ctx.info("Stopping Gradle daemons")
gradle = _get_gradle_wrapper(ctx)
result = await gradle.stop_daemon()
if ctx:
if result["success"]:
await ctx.info("Gradle daemons stopped successfully")
else:
await ctx.error("Failed to stop daemons", extra={"error": result.get("error")})
return TaskResult(
success=result["success"],
error=result.get("error"),
)
except Exception as e:
return TaskResult(
success=False,
error=str(e),
)
@mcp.tool()
async def get_gradle_config(ctx: Context | None = None) -> GradleConfigResult:
"""Get current Gradle configuration including memory settings.
Returns configuration from gradle.properties and gradle-wrapper.properties
including JVM args, daemon settings, and Gradle version info.
Useful for diagnosing memory issues or understanding project configuration.
Returns:
GradleConfigResult with JVM args, daemon/parallel/caching settings,
max workers, distribution URL, and Gradle version.
"""
try:
if ctx:
await ctx.info("Getting Gradle configuration")
gradle = _get_gradle_wrapper(ctx)
config = gradle.get_config()
if ctx:
await ctx.info("Retrieved Gradle configuration successfully")
return GradleConfigResult(
jvm_args=config["jvm_args"],
daemon_enabled=config["daemon_enabled"],
parallel_enabled=config["parallel_enabled"],
caching_enabled=config["caching_enabled"],
max_workers=config["max_workers"],
distribution_url=config["distribution_url"],
gradle_version=config["gradle_version"],
)
except Exception as e:
# Return a result with None values and could log the error
if ctx:
await ctx.error("Failed to get Gradle configuration", extra={"error": str(e)})
# Return empty config - all fields are optional (None)
return GradleConfigResult(
jvm_args=None,
daemon_enabled=None,
parallel_enabled=None,
caching_enabled=None,
max_workers=None,
distribution_url=None,
gradle_version=None,
)
def main() -> None:
"""Run the MCP server."""
# Start the dashboard in the background
try:
dashboard_url = _start_dashboard()
print(f"🐘 Gradle MCP Dashboard: {dashboard_url}")
except Exception as e:
print(f"⚠️ Failed to start dashboard: {e}")
# Run the MCP server
mcp.run()
if __name__ == "__main__":
main()
