list_executions
Retrieve and filter job execution records from Rundeck by project, status, time range, or job ID to monitor workflow activity and troubleshoot issues.
Instructions
List job executions with optional filtering.
Returns a list of executions matching the specified criteria. Filter by
project, job_id, status, or time range. Results are ordered by start time
(most recent first).
Args:
query: Query parameters for filtering executions
Returns:
List of Execution objects matching the query
Examples:
List recent executions in a project:
>>> result = list_executions(ExecutionQuery(project="myproject"))
List failed executions for a specific job:
>>> result = list_executions(ExecutionQuery(
... job_id="abc-123-def",
... status="failed"
... ))
List executions from the last hour:
>>> result = list_executions(ExecutionQuery(
... project="myproject",
... recent_filter="1h"
... ))
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| query | Yes |
Implementation Reference
- rundeck_mcp/tools/executions.py:14-58 (handler)The handler function that executes the list_executions tool, querying the Rundeck API for executions based on filters and parsing the response into Execution models.def list_executions(query: ExecutionQuery) -> ListResponseModel[Execution]: """List job executions with optional filtering. Returns a list of executions matching the specified criteria. Filter by project, job_id, status, or time range. Results are ordered by start time (most recent first). Args: query: Query parameters for filtering executions Returns: List of Execution objects matching the query Examples: List recent executions in a project: >>> result = list_executions(ExecutionQuery(project="myproject")) List failed executions for a specific job: >>> result = list_executions(ExecutionQuery( ... job_id="abc-123-def", ... status="failed" ... )) List executions from the last hour: >>> result = list_executions(ExecutionQuery( ... project="myproject", ... recent_filter="1h" ... )) """ client = get_client() params = query.to_params() # Use job-specific endpoint if job_id is provided if query.job_id: response = client.get(f"/job/{query.job_id}/executions", params=params) elif query.project: response = client.get(f"/project/{query.project}/executions", params=params) else: raise ValueError("Either project or job_id must be provided") # Handle response format (executions are nested in 'executions' key) executions_data = response.get("executions", []) if isinstance(response, dict) else response executions = [_parse_execution(exec_data) for exec_data in executions_data] return ListResponseModel[Execution](response=executions)
- Pydantic model defining the input schema for the list_executions tool, including all filter parameters and conversion to Rundeck API query params.class ExecutionQuery(BaseModel): """Query parameters for listing executions.""" model_config = ConfigDict(extra="forbid") project: str | None = Field( default=None, description="Filter by project name", ) job_id: str | None = Field( default=None, description="Filter by job ID (UUID)", ) status: ExecutionStatus | None = Field( default=None, description="Filter by execution status", ) user: str | None = Field( default=None, description="Filter by user who started the execution", ) recent_filter: str | None = Field( default=None, description="Filter by recent time period (e.g., '1h', '1d', '1w')", ) older_filter: str | None = Field( default=None, description="Filter for executions older than this period", ) begin: datetime | None = Field( default=None, description="Filter for executions started after this time", ) end: datetime | None = Field( default=None, description="Filter for executions started before this time", ) limit: int = Field( default=20, ge=1, le=MAX_RESULTS, description="Maximum number of results to return", ) offset: int = Field( default=0, ge=0, description="Offset for pagination", ) def to_params(self) -> dict[str, Any]: """Convert query to API parameters.""" params: dict[str, Any] = {"max": self.limit, "offset": self.offset} if self.status: params["statusFilter"] = self.status if self.user: params["userFilter"] = self.user if self.recent_filter: params["recentFilter"] = self.recent_filter if self.older_filter: params["olderFilter"] = self.older_filter if self.begin: params["begin"] = int(self.begin.timestamp() * 1000) if self.end: params["end"] = int(self.end.timestamp() * 1000) return params
- rundeck_mcp/server.py:108-109 (registration)Registers the list_executions tool (as part of read_tools) with the MCP server using appropriate safety annotations for read-only operations.for tool in read_tools: add_read_only_tool(mcp, tool)
- rundeck_mcp/tools/__init__.py:1-20 (registration)Exports and includes list_executions in the read_tools list used for MCP server registration.from .executions import ( get_execution, get_execution_output, list_executions, ) from .jobs import ( get_job, list_jobs, run_job, ) # Read-only tools (safe, non-destructive operations) read_tools = [ # Jobs list_jobs, get_job, # Executions list_executions, get_execution, get_execution_output,
- Pydantic model for the Execution objects returned by list_executions, with parsing, computed fields for duration and summary.class Execution(BaseModel): """A job execution instance. Represents a single run of a job, including its status, timing, and results. """ id: int = Field(description="The execution ID") href: str | None = Field(default=None, description="API URL for this execution") permalink: str | None = Field(default=None, description="Web UI URL for this execution") status: ExecutionStatus = Field(description="Execution status") project: str = Field(description="The project name") job: JobReference | None = Field(default=None, description="Reference to the job (None for adhoc executions)") user: str = Field(description="User who started this execution") date_started: datetime | None = Field( default=None, alias="date-started", description="When the execution started", ) date_ended: datetime | None = Field( default=None, alias="date-ended", description="When the execution ended (None if still running)", ) argstring: str | None = Field( default=None, description="The argument string used for this execution", ) description: str | None = Field(default=None, description="Execution description") successful_nodes: list[str] | None = Field( default=None, alias="successfulNodes", description="List of nodes that succeeded", ) failed_nodes: list[str] | None = Field( default=None, alias="failedNodes", description="List of nodes that failed", ) @field_validator("date_started", "date_ended", mode="before") @classmethod def parse_date_dict(cls, v: Any) -> datetime | None: """Parse date from Rundeck's dict format or ISO string.""" if v is None: return None if isinstance(v, datetime): return v if isinstance(v, dict): # Rundeck returns {"unixtime": 1234567890, "date": "..."} if "unixtime" in v: return datetime.fromtimestamp(v["unixtime"] / 1000) if "date" in v: return datetime.fromisoformat(v["date"].replace("Z", "+00:00")) if isinstance(v, str): return datetime.fromisoformat(v.replace("Z", "+00:00")) return None @computed_field @property def duration_seconds(self) -> float | None: """Calculate duration in seconds.""" if self.date_started and self.date_ended: return (self.date_ended - self.date_started).total_seconds() return None @computed_field @property def execution_summary(self) -> str: """Generate a human-readable summary of this execution.""" parts = [f"Execution #{self.id}: {self.status.upper()}"] if self.job: parts.append(f"Job: {self.job.name}") if self.job.group: parts[-1] = f"Job: {self.job.group}/{self.job.name}" if self.user: parts.append(f"User: {self.user}") if self.duration_seconds is not None: parts.append(f"Duration: {self.duration_seconds:.1f}s") if self.argstring: parts.append(f"Args: {self.argstring}") return " | ".join(parts) @computed_field @property def type(self) -> Literal["execution"]: return "execution"