feat: results-inspector MCP server — mirrored core + lens layer

.claude/skills/al_inspect_results_mcp.md

@@ -0,0 +1 @@
+../../skills/al_inspect_results_mcp.md

.mcp.json

@@ -0,0 +1,8 @@
+{
+  "mcpServers": {
+    "pyauto-results-inspector": {
+      "command": "python",
+      "args": ["-m", "autoassistant.mcp"]
+    }
+  }
+}

autoassistant/audit_skill_apis.py

@@ -428,6 +428,9 @@ def select_files(root: Path, scope: str) -> list[Path]:
     scripts = [
         p for p in sorted((root / "scripts").rglob("*.py")) if p.name not in tooling
     ]
+    # The MCP server (autoassistant/mcp/) is the one part of autoassistant/ with
+    # real API usage rather than alias-pattern text, so it joins the scan.
+    scripts += sorted((root / "autoassistant" / "mcp").glob("*.py"))
     if scope == "skills":
         return skills
     if scope == "wiki":
@@ -450,6 +453,7 @@ def select_idiom_files(root: Path) -> list[Path]:
     tooling = {"audit_skill_apis.py", "refresh_api_docs.py", "test_api_gate.py"}
     md = sorted((root / "skills").glob("*.md")) + sorted((root / "wiki").rglob("*.md"))
     py = [p for p in sorted((root / "scripts").rglob("*.py")) if p.name not in tooling]
+    py += sorted((root / "autoassistant" / "mcp").glob("*.py"))
     return md + py
 
 

autoassistant/mcp/__init__.py

@@ -0,0 +1,10 @@
+"""
+The read-only results-inspector MCP server (lens edition).
+
+`tools.py` is mirrored verbatim from `autofit_assistant:autoassistant/mcp/tools.py`
+(keep in sync — graduation to an `autofit[mcp]` extra is the recorded de-dup
+path); `lens_tools.py` layers the PyAutoLens-specific image/FITS extraction on
+top; `server.py` registers both tool sets; `python -m autoassistant.mcp` runs
+the stdio server. Documentation and client configuration live in
+`skills/al_inspect_results_mcp.md`.
+"""

autoassistant/mcp/__main__.py

@@ -0,0 +1,3 @@
+from autoassistant.mcp.server import mcp
+
+mcp.run()

autoassistant/mcp/lens_tools.py

@@ -0,0 +1,77 @@
+"""
+Lens-specific results-inspector tools, layered on the mirrored core
+(`tools.py`).
+
+Everything here resolves string specs like "subplot_fit.data" against the
+`al.agg` enum groups and hands them to the existing aggregation machinery
+(`af.AggregateImages`, `af.AggregateFITS`) — argument parsing + one call,
+nothing more (`skills/al_inspect_results_mcp.md`, "glue, not code").
+"""
+
+import enum
+from pathlib import Path
+
+import autofit as af
+import autolens as al
+
+from autoassistant.mcp.tools import DirectoryAggregator, _stdout_to_stderr
+
+
+def _enum_groups() -> dict:
+    return {
+        name: obj
+        for name in dir(al.agg)
+        if not name.startswith("_")
+        and isinstance(obj := getattr(al.agg, name), type)
+        and issubclass(obj, enum.Enum)
+    }
+
+
+def list_image_names() -> dict:
+    return {
+        name: [member.name for member in group]
+        for name, group in sorted(_enum_groups().items())
+    }
+
+
+def _resolve(spec: str):
+    group_name, _, member_name = spec.partition(".")
+    groups = _enum_groups()
+    if group_name not in groups:
+        raise KeyError(
+            f"Unknown group {group_name!r} — one of {sorted(groups)}."
+        )
+    try:
+        return groups[group_name][member_name]
+    except KeyError:
+        raise KeyError(
+            f"Unknown name {member_name!r} in {group_name!r} — one of "
+            f"{[member.name for member in groups[group_name]]}."
+        ) from None
+
+
+def combine_images(directory: str, subplots: list, subplot_width: int = 0):
+    with _stdout_to_stderr():
+        aggregator = DirectoryAggregator.from_directory(directory)
+        agg_image = af.AggregateImages(aggregator=aggregator)
+        kwargs = dict(subplot_width=subplot_width) if subplot_width else {}
+        return agg_image.extract_image(
+            subplots=[_resolve(spec) for spec in subplots], **kwargs
+        )
+
+
+def extract_fits(
+    directory: str, hdus: list, destination_path: str, overwrite: bool = False
+) -> str:
+    destination = Path(destination_path).resolve()
+    if Path(directory).resolve() in destination.parents:
+        raise ValueError(
+            "destination_path must be outside the search-output directory — "
+            "nothing writes into output/."
+        )
+    with _stdout_to_stderr():
+        aggregator = DirectoryAggregator.from_directory(directory)
+        agg_fits = af.AggregateFITS(aggregator=aggregator)
+        hdu_list = agg_fits.extract_fits(hdus=[_resolve(spec) for spec in hdus])
+    hdu_list.writeto(destination, overwrite=overwrite)
+    return str(destination)

autoassistant/mcp/server.py

@@ -0,0 +1,180 @@
+"""
+The read-only results-inspector MCP stdio server (lens edition).
+
+The core tools mirror `autofit_assistant:autoassistant/mcp/server.py` verbatim
+(`tools.py`/`__main__.py` are byte-identical mirrors — `diff` them when
+syncing); the lens tools at the bottom are this repo's own layer. Run with
+`python -m autoassistant.mcp`; client configuration and the design rules live
+in `skills/al_inspect_results_mcp.md`.
+
+Every tool is read-only against `output/`: nothing here composes models, runs
+fits, or writes into a search-output directory.
+"""
+
+import io
+import logging
+import sys
+
+from mcp.server.fastmcp import FastMCP, Image
+
+from autoassistant.mcp import lens_tools, tools
+
+
+def _route_logging_to_stderr():
+    """
+    stdout carries the JSON-RPC channel, but autofit's logging config (loaded
+    on import) attaches stdout stream handlers — one stray log line corrupts
+    the protocol, so every stdout handler is rebound to stderr.
+    """
+    loggers = [logging.getLogger()] + [
+        logging.getLogger(name) for name in logging.root.manager.loggerDict
+    ]
+    for logger in loggers:
+        for handler in getattr(logger, "handlers", []):
+            if (
+                isinstance(handler, logging.StreamHandler)
+                and getattr(handler, "stream", None) is sys.stdout
+            ):
+                handler.setStream(sys.stderr)
+
+
+_route_logging_to_stderr()
+
+mcp = FastMCP("pyauto-results-inspector")
+
+
+def _png(image) -> Image:
+    buffer = io.BytesIO()
+    image.save(buffer, format="PNG")
+    return Image(data=buffer.getvalue(), format="png")
+
+
+@mcp.tool()
+def list_searches(
+    directory: str,
+    sort_by: str = "log_evidence",
+    limit: int = 20,
+    completed_only: bool = False,
+) -> list:
+    """
+    List every model-fit found under `directory` (searched recursively): one
+    row per fit with its name, unique tag, output directory, completion state,
+    log evidence, maximum log likelihood and free-parameter count.
+
+    Rows are sorted by `sort_by` (descending; fits without that value last) —
+    use "log_evidence" for nested samplers or "max_log_likelihood" generally —
+    and truncated to `limit` (pass 0 for all). The returned `directory` of a
+    row is what the other tools take as their `directory` argument.
+    """
+    return tools.list_searches(
+        directory, sort_by=sort_by, limit=limit, completed_only=completed_only
+    )
+
+
+@mcp.tool()
+def get_model(directory: str) -> dict:
+    """
+    The model that was fitted in one search-output directory: a human-readable
+    `info` block (component classes, priors) and the full model as a dict.
+    """
+    return tools.get_model(directory)
+
+
+@mcp.tool()
+def get_result_summary(directory: str) -> str:
+    """
+    The `model.results` text for one search-output directory: the fit's own
+    summary of the maximum-likelihood model and (when the search produces
+    them) parameter estimates with errors.
+    """
+    return tools.get_result_summary(directory)
+
+
+@mcp.tool()
+def get_samples_summary(directory: str) -> dict:
+    """
+    Posterior summary for one search-output directory: log evidence (None for
+    MLE/MCMC searches without one), maximum log likelihood, the model's
+    parameter paths, and the maximum-likelihood and median-PDF parameter
+    vectors (`median_pdf_parameters` is None for MLE searches, which have no
+    PDF).
+    """
+    return tools.get_samples_summary(directory)
+
+
+@mcp.tool()
+def get_search_info(directory: str) -> dict:
+    """
+    The non-linear search used in one search-output directory: name, unique
+    tag, completion state, and the search's serialized settings.
+    """
+    return tools.get_search_info(directory)
+
+
+@mcp.tool()
+def list_images(directory: str) -> list:
+    """
+    Names of the visualization images (`image/*.png`) available in one
+    search-output directory — pass a name (without `.png`) to `fetch_image`.
+    """
+    return tools.list_images(directory)
+
+
+@mcp.tool()
+def fetch_image(directory: str, name: str = "subplot_fit") -> Image:
+    """
+    One visualization image from a search-output directory (e.g.
+    "subplot_fit"), returned inline so it renders directly in chat. Use
+    `list_images` to see what is available.
+    """
+    return _png(tools.fetch_image(directory, name=name))
+
+
+# --- Lens layer (this repo's own tools) -------------------------------------
+
+
+@mcp.tool()
+def list_extractable_images() -> dict:
+    """
+    The lens image groups and names that `combine_lens_images` /
+    `extract_lens_fits` accept, e.g. group "subplot_fit" with names "data",
+    "model_data", "normalized_residual_map". Pass specs as "group.name".
+    """
+    return lens_tools.list_image_names()
+
+
+@mcp.tool()
+def combine_lens_images(
+    directory: str, subplots: list, subplot_width: int = 0
+) -> Image:
+    """
+    Extract named panels (specs like "subplot_fit.data",
+    "subplot_fit.model_data") from every fit found under `directory` and
+    combine them into one image, rendered inline in chat — one row per fit by
+    default (`subplot_width` overrides the layout). Use
+    `list_extractable_images` for the available specs.
+    """
+    return _png(
+        lens_tools.combine_images(
+            directory, subplots=subplots, subplot_width=subplot_width
+        )
+    )
+
+
+@mcp.tool()
+def extract_lens_fits(
+    directory: str, hdus: list, destination_path: str, overwrite: bool = False
+) -> str:
+    """
+    Extract named FITS HDUs (specs like "fits_fit.model_data",
+    "fits_tracer.convergence") from every fit found under `directory` into a
+    single .fits file written to `destination_path` (which must be outside the
+    search-output directory). Returns the written path.
+    """
+    return lens_tools.extract_fits(
+        directory, hdus=hdus, destination_path=destination_path, overwrite=overwrite
+    )
+
+
+if __name__ == "__main__":
+    mcp.run()