pipeline_health

Instructions

Implementation Reference

• src/lutris_source_mcp/tools/diagnostics.py:14-53 (handler)

  The `pipeline_health` function is the tool handler. It checks Prowlarr reachability (list_indexers), qBittorrent reachability (app_version), and surfaces config/policy info. Uses the @mcp.tool decorator with description 'Verify Prowlarr + qBittorrent reachability and surface config. mutates: false'. The function instantiates Prowlarr and Qbittorrent adapters from config, catches their respective errors, and returns a dict with prowlarr, qbittorrent, and policy status.

  @mcp.tool(description="Verify Prowlarr + qBittorrent reachability and surface config. mutates: false")
  def pipeline_health() -> dict[str, Any]:
      cfg = _cfg.load()
      out: dict[str, Any] = {
          "config_path": str(_cfg.CONFIG_PATH),
          "config_present": _cfg.CONFIG_PATH.exists(),
      }
      try:
          pw = Prowlarr(cfg.prowlarr)
          idx = pw.list_indexers()
          out["prowlarr"] = {
              "ok": True,
              "base_url": cfg.prowlarr.base_url,
              "indexers": [
                  {"id": i.get("id"), "name": i.get("name"),
                   "priority": i.get("priority"), "enable": i.get("enable")}
                  for i in idx
              ],
          }
      except ProwlarrError as exc:
          out["prowlarr"] = {"ok": False, "error": str(exc)}
  
      try:
          qb = Qbittorrent(cfg.qbittorrent)
          out["qbittorrent"] = {
              "ok": True,
              "base_url": cfg.qbittorrent.base_url,
              "version": qb.app_version(),
              "download_dir": cfg.qbittorrent.download_dir,
          }
      except QbError as exc:
          out["qbittorrent"] = {"ok": False, "error": str(exc)}
  
      out["policy"] = {
          "blocklist": cfg.policy.blocklist,
          "max_size_gb": cfg.policy.max_size_gb,
          "freeleech_only": cfg.policy.freeleech_only,
          "min_seeders": cfg.policy.min_seeders,
      }
      return out

• src/lutris_source_mcp/tools/diagnostics.py:14-14 (registration)

  The `@mcp.tool()` decorator registers the `pipeline_health` function as an MCP tool on the 'mcp' server object.

  @mcp.tool(description="Verify Prowlarr + qBittorrent reachability and surface config. mutates: false")

• src/lutris_source_mcp/tools/diagnostics.py:15-53 (helper)

  The function uses `_cfg.load()` to read configuration from a TOML file, `Prowlarr` and `Qbittorrent` adapter classes to check service health, and `_cfg.CONFIG_PATH` for path info. These are supporting utilities imported from other modules.

  def pipeline_health() -> dict[str, Any]:
      cfg = _cfg.load()
      out: dict[str, Any] = {
          "config_path": str(_cfg.CONFIG_PATH),
          "config_present": _cfg.CONFIG_PATH.exists(),
      }
      try:
          pw = Prowlarr(cfg.prowlarr)
          idx = pw.list_indexers()
          out["prowlarr"] = {
              "ok": True,
              "base_url": cfg.prowlarr.base_url,
              "indexers": [
                  {"id": i.get("id"), "name": i.get("name"),
                   "priority": i.get("priority"), "enable": i.get("enable")}
                  for i in idx
              ],
          }
      except ProwlarrError as exc:
          out["prowlarr"] = {"ok": False, "error": str(exc)}
  
      try:
          qb = Qbittorrent(cfg.qbittorrent)
          out["qbittorrent"] = {
              "ok": True,
              "base_url": cfg.qbittorrent.base_url,
              "version": qb.app_version(),
              "download_dir": cfg.qbittorrent.download_dir,
          }
      except QbError as exc:
          out["qbittorrent"] = {"ok": False, "error": str(exc)}
  
      out["policy"] = {
          "blocklist": cfg.policy.blocklist,
          "max_size_gb": cfg.policy.max_size_gb,
          "freeleech_only": cfg.policy.freeleech_only,
          "min_seeders": cfg.policy.min_seeders,
      }
      return out