feat: read-only results-inspector MCP server (core aggregator tools)

.claude/skills/af_inspect_results_mcp.md

@@ -0,0 +1 @@
+../../skills/af_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

@@ -416,6 +416,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":
@@ -438,6 +441,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,8 @@
+"""
+The read-only results-inspector MCP server.
+
+`tools.py` holds the plain tool functions (no MCP dependency — the test suite
+exercises them directly); `server.py` registers them with an MCP stdio server;
+`python -m autoassistant.mcp` runs it. Documentation, client configuration and
+the design rules live in `skills/af_inspect_results_mcp.md`.
+"""

autoassistant/mcp/__main__.py

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

autoassistant/mcp/server.py

@@ -0,0 +1,133 @@
+"""
+The read-only results-inspector MCP stdio server.
+
+Registers the `tools` functions with an MCP server so chat harnesses without
+code execution (Claude Desktop, Claude Code) can inspect PyAutoFit output
+directories. Run with `python -m autoassistant.mcp`; client configuration and
+the design rules live in `skills/af_inspect_results_mcp.md`.
+
+Every tool is read-only: nothing here composes models, runs fits, or writes
+into `output/`.
+"""
+
+import io
+import logging
+import sys
+
+from mcp.server.fastmcp import FastMCP, Image
+
+from autoassistant.mcp import 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))
+
+
+if __name__ == "__main__":
+    mcp.run()

autoassistant/mcp/tools.py

@@ -0,0 +1,148 @@
+"""
+Read-only results-inspector tools over PyAutoFit output directories.
+
+Each function is a thin wrapper over an existing public PyAutoFit aggregator
+API (`autofit.aggregator.Aggregator`, `af.SearchOutput`): argument parsing, one call, and
+JSON-friendly serialization — nothing more. Any behaviour beyond that belongs
+in PyAutoFit itself, not here (`skills/af_inspect_results_mcp.md`,
+"glue, not code").
+
+This module deliberately has no MCP dependency: `server.py` registers these
+functions as MCP tools, and the test suite exercises them without the `mcp`
+package installed.
+"""
+
+import contextlib
+import json
+import sys
+from pathlib import Path
+
+from autoconf.dictable import to_dict
+
+import autofit as af
+
+# The directory-backed aggregator — af.Aggregator is the database-backed one,
+# and the alias also keeps the API audit from resolving it there.
+from autofit.aggregator import Aggregator as DirectoryAggregator
+
+
+@contextlib.contextmanager
+def _stdout_to_stderr():
+    """
+    An MCP stdio server must keep stdout clean — it carries the JSON-RPC
+    channel — but the directory aggregator prints progress to stdout,
+    so every autofit call runs with stdout redirected to stderr.
+    """
+    with contextlib.redirect_stdout(sys.stderr):
+        yield
+
+
+def _float_or_none(value):
+    try:
+        return None if value is None else float(value)
+    except (TypeError, ValueError):
+        return None
+
+
+def _search_row(search) -> dict:
+    summary = search.samples_summary
+    max_lh_sample = getattr(summary, "max_log_likelihood_sample", None)
+    return dict(
+        name=search.name,
+        unique_tag=search.unique_tag,
+        directory=str(search.directory),
+        is_complete=search.is_complete,
+        log_evidence=_float_or_none(getattr(summary, "log_evidence", None)),
+        max_log_likelihood=_float_or_none(
+            getattr(max_lh_sample, "log_likelihood", None)
+        ),
+        model_free_parameters=getattr(search.model, "prior_count", None),
+    )
+
+
+def list_searches(
+    directory: str,
+    sort_by: str = "log_evidence",
+    limit: int = 20,
+    completed_only: bool = False,
+) -> list:
+    with _stdout_to_stderr():
+        aggregator = DirectoryAggregator.from_directory(
+            directory, completed_only=completed_only
+        )
+        rows = [_search_row(search) for search in aggregator]
+    rows.sort(
+        key=lambda row: (row.get(sort_by) is not None, row.get(sort_by)),
+        reverse=True,
+    )
+    return rows[:limit] if limit else rows
+
+
+def get_model(directory: str) -> dict:
+    with _stdout_to_stderr():
+        model = af.SearchOutput(Path(directory)).model
+    return dict(info=model.info, model=to_dict(model))
+
+
+def get_result_summary(directory: str) -> str:
+    with _stdout_to_stderr():
+        return af.SearchOutput(Path(directory)).model_results
+
+
+def get_samples_summary(directory: str) -> dict:
+    with _stdout_to_stderr():
+        search_output = af.SearchOutput(Path(directory))
+        summary = search_output.samples_summary
+        if summary is None:
+            raise FileNotFoundError(
+                f"No samples summary found under {directory}/files/ — "
+                "is this a completed search output directory?"
+            )
+        model = search_output.model
+        return dict(
+            log_evidence=_float_or_none(summary.log_evidence),
+            max_log_likelihood=_float_or_none(
+                summary.max_log_likelihood_sample.log_likelihood
+            ),
+            parameter_paths=[".".join(path) for path in model.paths],
+            max_log_likelihood_parameters=[
+                float(value)
+                for value in summary.max_log_likelihood_sample.parameter_lists_for_model(
+                    model
+                )
+            ],
+            # MLE searches (e.g. LBFGS) have no PDF, so no median-PDF sample.
+            median_pdf_parameters=None
+            if summary.median_pdf_sample is None
+            else [
+                float(value)
+                for value in summary.median_pdf_sample.parameter_lists_for_model(
+                    model
+                )
+            ],
+        )
+
+
+def get_search_info(directory: str) -> dict:
+    search_json = Path(directory) / "files" / "search.json"
+    with _stdout_to_stderr():
+        search_output = af.SearchOutput(Path(directory))
+        return dict(
+            name=search_output.name,
+            unique_tag=search_output.unique_tag,
+            is_complete=search_output.is_complete,
+            search=json.loads(search_json.read_text())
+            if search_json.exists()
+            else None,
+        )
+
+
+def list_images(directory: str) -> list:
+    # Visualization outputs live under <directory>/image/ (SearchOutput.image
+    # reads there, despite its docstring saying `files/`).
+    return sorted(path.name for path in (Path(directory) / "image").glob("*.png"))
+
+
+def fetch_image(directory: str, name: str = "subplot_fit"):
+    with _stdout_to_stderr():
+        return af.SearchOutput(Path(directory)).image(name)