Api maestro

kcidev/api.py

@@ -46,6 +46,12 @@
     resolve_kcidb_config,
     submit_to_kcidb,
 )
+from kcidev.libs.maestro_common import (
+    maestro_get_node,
+    maestro_get_nodes,
+    send_checkout_full,
+    send_jobretry,
+)
 from kcidev.main import get_cli
 
 
@@ -88,6 +94,8 @@ def _as_library_error(action, func, *args, **kwargs):
         raise KciDevError(f"{action}: {exc.message}") from exc
     except click.Abort as exc:
         raise KciDevError(action) from exc
+    except SystemExit as exc:
+        raise KciDevError(action) from exc
 
 
 class KernelCIClient:
@@ -470,3 +478,85 @@ def get_tree_report(
             max_age_in_hours,
             min_age_in_hours,
         )
+
+    def _instance_setting(self, key):
+        return ((self.cfg or {}).get(self.instance) or {}).get(key)
+
+    def _require_maestro_setting(self, value, key):
+        if not value:
+            raise KciDevError(
+                f"No Maestro {key} configured; pass it explicitly or set it "
+                "in the instance config"
+            )
+        return value
+
+    def get_node(self, node_id, api_url=None):
+        """Fetch a single Maestro node by id."""
+        url = self._require_maestro_setting(
+            api_url or self._instance_setting("api"), "api URL"
+        )
+        return _as_library_error(
+            "Maestro node request failed", maestro_get_node, url, node_id
+        )
+
+    def get_nodes(self, limit=50, offset=0, filters=None, api_url=None):
+        """List Maestro nodes with 'field=value' filters and pagination."""
+        url = self._require_maestro_setting(
+            api_url or self._instance_setting("api"), "api URL"
+        )
+        return _as_library_error(
+            "Maestro nodes request failed",
+            maestro_get_nodes,
+            url,
+            limit,
+            offset,
+            filters or [],
+            True,
+        )
+
+    def retry_job(self, node_id, pipeline_url=None, token=None):
+        """Retry a failed or incomplete job by Maestro node id."""
+        url = self._require_maestro_setting(
+            pipeline_url or self._instance_setting("pipeline"), "pipeline URL"
+        )
+        token = self._require_maestro_setting(
+            token or self._instance_setting("token"), "token"
+        )
+        result = _as_library_error(
+            "Maestro job retry failed", send_jobretry, url, node_id, token
+        )
+        if result is None:
+            raise KciDevError(f"Maestro job retry failed for node {node_id}")
+        return result
+
+    def trigger_checkout(
+        self,
+        giturl,
+        branch,
+        commit,
+        job_filter,
+        platform_filter=None,
+        pipeline_url=None,
+        token=None,
+    ):
+        """Trigger a pipeline checkout of a tree/branch/commit with a job filter."""
+        url = self._require_maestro_setting(
+            pipeline_url or self._instance_setting("pipeline"), "pipeline URL"
+        )
+        token = self._require_maestro_setting(
+            token or self._instance_setting("token"), "token"
+        )
+        kwargs = {
+            "giturl": giturl,
+            "branch": branch,
+            "commit": commit,
+            "job_filter": job_filter,
+        }
+        if platform_filter:
+            kwargs["platform_filter"] = platform_filter
+        result = _as_library_error(
+            "Maestro checkout failed", send_checkout_full, url, token, **kwargs
+        )
+        if result is None:
+            raise KciDevError(f"Maestro checkout failed for {giturl} at {commit}")
+        return result

kcidev/libs/maestro_common.py

@@ -326,3 +326,76 @@ def maestro_watch_jobs(baseurl, token, treeid, job_filter, test):
         kci_msg_nonl(f"\rRunning job...")
         previous_nodes = nodes
         time.sleep(30)
+
+
+def send_jobretry(baseurl, jobid, token):
+    url = baseurl + "api/jobretry"
+    headers = {
+        "Content-Type": "application/json; charset=utf-8",
+        "Authorization": f"{token}",
+    }
+    data = {"nodeid": jobid}
+    jdata = json.dumps(data)
+
+    logging.info(f"Sending job retry request for node: {jobid}")
+    logging.debug(f"Retry URL: {url}")
+    maestro_print_api_call(url, data)
+
+    try:
+        logging.debug("Sending POST request for job retry")
+        response = kcidev_session.post(url, headers=headers, data=jdata)
+        logging.debug(f"Response status: {response.status_code}")
+    except requests.exceptions.RequestException as e:
+        logging.error(f"Failed to send job retry request: {e}")
+        kci_err(f"API connection error: {e}")
+        return
+
+    if response.status_code != 200:
+        logging.error(f"Job retry failed with status {response.status_code}")
+        maestro_api_error(response)
+        return None
+
+    result = response.json()
+    logging.info(f"Job retry request successful: {result.get('message', 'No message')}")
+    return result
+
+
+def send_checkout_full(baseurl, token, **kwargs):
+    url = baseurl + "api/checkout"
+    headers = {
+        "Content-Type": "application/json; charset=utf-8",
+        "Authorization": f"{token}",
+    }
+    data = {
+        "url": kwargs["giturl"],
+        "branch": kwargs["branch"],
+        "commit": kwargs["commit"],
+        "jobfilter": kwargs["job_filter"],
+    }
+    if "platform_filter" in kwargs:
+        data["platformfilter"] = kwargs["platform_filter"]
+
+    logging.info(
+        f"Sending checkout request for {kwargs['giturl']} branch {kwargs['branch']} commit {kwargs['commit']}"
+    )
+    logging.debug(f"Checkout data: {json.dumps(data, indent=2)}")
+
+    jdata = json.dumps(data)
+    maestro_print_api_call(url, data)
+    try:
+        logging.debug(f"POST request to: {url}")
+        response = kcidev_session.post(url, headers=headers, data=jdata, timeout=30)
+        logging.debug(f"Checkout response status: {response.status_code}")
+    except requests.exceptions.RequestException as e:
+        logging.error(f"Checkout API request failed: {e}")
+        kci_err(f"API connection error: {e}")
+        return None
+
+    if response.status_code != 200:
+        logging.error(f"Checkout failed with status {response.status_code}")
+        maestro_api_error(response)
+        return None
+
+    result = response.json()
+    logging.info(f"Checkout successful - tree ID: {result.get('treeid', 'unknown')}")
+    return result

kcidev/mcp/__init__.py

@@ -3,6 +3,7 @@
 
 from mcp.server.fastmcp import FastMCP
 
+from kcidev.api import KernelCIClient
 from kcidev.libs.common import kcidev_version
 from kcidev.mcp import tools_dashboard, tools_maestro
 
@@ -20,8 +21,9 @@ def create_server(cfg=None, instance=None, host="127.0.0.1", port=8000):
     server = FastMCP("kci-dev", instructions=SERVER_INSTRUCTIONS, host=host, port=port)
     server._mcp_server.version = kcidev_version
     tools_dashboard.register_tools(server)
+    client = KernelCIClient(cfg=cfg, instance=instance)
     icfg = (cfg or {}).get(instance) or {}
     tools_maestro.register_tools(
-        server, icfg.get("api"), icfg.get("pipeline"), icfg.get("token")
+        server, client, icfg.get("api"), icfg.get("pipeline"), icfg.get("token")
     )
     return server

kcidev/mcp/tools_maestro.py

@@ -1,37 +1,10 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
 
-from kcidev.libs.maestro_common import maestro_get_node, maestro_get_nodes
-from kcidev.mcp.errors import ToolExecutionError, tool_errors
-from kcidev.subcommands.checkout import send_checkout_full
-from kcidev.subcommands.testretry import send_jobretry
+from kcidev.mcp.errors import tool_errors
 
 
-def _retry_job(pipeline_url, token, node_id):
-    result = send_jobretry(pipeline_url, node_id, token)
-    if result is None:
-        raise ToolExecutionError(f"Maestro job retry failed for node {node_id}")
-    return result
-
-
-def _trigger_checkout(
-    pipeline_url, token, giturl, branch, commit, job_filter, platform_filter
-):
-    kwargs = {
-        "giturl": giturl,
-        "branch": branch,
-        "commit": commit,
-        "job_filter": job_filter,
-    }
-    if platform_filter:
-        kwargs["platform_filter"] = platform_filter
-    result = send_checkout_full(pipeline_url, token, **kwargs)
-    if result is None:
-        raise ToolExecutionError(f"Maestro checkout failed for {giturl} at {commit}")
-    return result
-
-
-def register_tools(server, api_url, pipeline_url, token):
+def register_tools(server, client, api_url, pipeline_url, token):
     from mcp.types import ToolAnnotations
 
     read_only = ToolAnnotations(readOnlyHint=True)
@@ -49,7 +22,7 @@ def get_node(node_id: str):
             this to poll a job started with trigger_checkout or retry_job.
             Node ids are 24-character hex strings.
             """
-            return maestro_get_node(api_url, node_id)
+            return client.get_node(node_id)
 
         @server.tool(annotations=read_only)
         @tool_errors
@@ -67,7 +40,7 @@ def list_nodes(
             than paginating from the start. Use limit and offset to
             paginate within the window.
             """
-            return maestro_get_nodes(api_url, limit, offset, filters or [], True)
+            return client.get_nodes(limit=limit, offset=offset, filters=filters or [])
 
     if pipeline_url and token:
 
@@ -79,7 +52,7 @@ def retry_job(node_id: str):
             Creates a new job for the given Maestro node id. Use list_nodes
             or the dashboard tools to find the node id of the failed job.
             """
-            return _retry_job(pipeline_url, token, node_id)
+            return client.retry_job(node_id)
 
         @server.tool(annotations=action)
         @tool_errors
@@ -98,6 +71,6 @@ def trigger_checkout(
             platform_filter. Returns a tree id; poll progress with list_nodes
             using 'treeid=<tree id>'.
             """
-            return _trigger_checkout(
-                pipeline_url, token, giturl, branch, commit, job_filter, platform_filter
+            return client.trigger_checkout(
+                giturl, branch, commit, job_filter, platform_filter
             )

kcidev/subcommands/checkout.py

@@ -2,62 +2,19 @@
 # -*- coding: utf-8 -*-
 
 import datetime
-import json
 import logging
 import re
 import subprocess
 import sys
 import time
 
 import click
-import requests
 from git import Repo
 
 from kcidev.libs.common import *
 from kcidev.libs.maestro_common import *
 
 
-def send_checkout_full(baseurl, token, **kwargs):
-    url = baseurl + "api/checkout"
-    headers = {
-        "Content-Type": "application/json; charset=utf-8",
-        "Authorization": f"{token}",
-    }
-    data = {
-        "url": kwargs["giturl"],
-        "branch": kwargs["branch"],
-        "commit": kwargs["commit"],
-        "jobfilter": kwargs["job_filter"],
-    }
-    if "platform_filter" in kwargs:
-        data["platformfilter"] = kwargs["platform_filter"]
-
-    logging.info(
-        f"Sending checkout request for {kwargs['giturl']} branch {kwargs['branch']} commit {kwargs['commit']}"
-    )
-    logging.debug(f"Checkout data: {json.dumps(data, indent=2)}")
-
-    jdata = json.dumps(data)
-    maestro_print_api_call(url, data)
-    try:
-        logging.debug(f"POST request to: {url}")
-        response = kcidev_session.post(url, headers=headers, data=jdata, timeout=30)
-        logging.debug(f"Checkout response status: {response.status_code}")
-    except requests.exceptions.RequestException as e:
-        logging.error(f"Checkout API request failed: {e}")
-        kci_err(f"API connection error: {e}")
-        return None
-
-    if response.status_code != 200:
-        logging.error(f"Checkout failed with status {response.status_code}")
-        maestro_api_error(response)
-        return None
-
-    result = response.json()
-    logging.info(f"Checkout successful - tree ID: {result.get('treeid', 'unknown')}")
-    return result
-
-
 def retrieve_tot_commit(repourl, branch):
     """
     Retrieve the latest commit on a branch

kcidev/subcommands/testretry.py

@@ -1,49 +1,15 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
 
-import json
 import logging
 
 import click
-import requests
 from git import Repo
 
 from kcidev.libs.common import *
 from kcidev.libs.maestro_common import *
 
 
-def send_jobretry(baseurl, jobid, token):
-    url = baseurl + "api/jobretry"
-    headers = {
-        "Content-Type": "application/json; charset=utf-8",
-        "Authorization": f"{token}",
-    }
-    data = {"nodeid": jobid}
-    jdata = json.dumps(data)
-
-    logging.info(f"Sending job retry request for node: {jobid}")
-    logging.debug(f"Retry URL: {url}")
-    maestro_print_api_call(url, data)
-
-    try:
-        logging.debug("Sending POST request for job retry")
-        response = kcidev_session.post(url, headers=headers, data=jdata)
-        logging.debug(f"Response status: {response.status_code}")
-    except requests.exceptions.RequestException as e:
-        logging.error(f"Failed to send job retry request: {e}")
-        kci_err(f"API connection error: {e}")
-        return
-
-    if response.status_code != 200:
-        logging.error(f"Job retry failed with status {response.status_code}")
-        maestro_api_error(response)
-        return None
-
-    result = response.json()
-    logging.info(f"Job retry request successful: {result.get('message', 'No message')}")
-    return result
-
-
 @click.command(
     help="""Retry a failed or incomplete test job on KernelCI.