`cosmotron_mcp.tools.io`¶

save_covariance ¶

save_covariance(cov_result: dict, output_path: str, bin_index: int | None = None, generated_by: str | None = None, agent_name: str | None = None) -> dict

Save a covariance result dict to JSON with provenance and auto-registration.

Accepts the direct output of compute_covariance. Numpy arrays (covariance, ells, sqrt_diag, cls_theory) are serialised to lists automatically. The caller's dict is never mutated.

Auto-registers in artefact_registry.json when output_path is inside a results/ subdirectory. Carries bin_index so the dashboard can pair this entry with its matching cls_json artefact for error-bar and theory-Cl overlay.

Parameters:

Name	Type	Description	Default
`cov_result`	`dict`	Output dict from `compute_covariance`. Must contain at minimum `covariance`, `ells`, `sqrt_diag`, and `cl_theory_mode`.	required
`output_path`	`str`	Destination JSON path.	required
`bin_index`	`int \| None`	Tomographic bin index — used to link with the cls_json artefact.	`None`
`generated_by`	`str \| None`	Infrastructure kwarg — omit in normal script usage.	`None`
`agent_name`	`str \| None`	Infrastructure kwarg — omit in normal script usage.	`None`

Returns:

Type	Description
`dict`	`dict` with keys: `path`, `n_bandpowers`, `cl_theory_mode`,
`dict`	`provenance`.

Source code in cosmotron_mcp/server.py

@mcp.tool()
@sync_budget_guard
def save_covariance(
    cov_result: dict,
    output_path: str,
    bin_index: int | None = None,
    generated_by: str | None = None,
    agent_name: str | None = None,
) -> dict:
    """Save a covariance result dict to JSON with provenance and auto-registration.

    Accepts the direct output of compute_covariance. Numpy arrays (covariance,
    ells, sqrt_diag, cls_theory) are serialised to lists automatically. The
    caller's dict is never mutated.

    Auto-registers in artefact_registry.json when output_path is inside a
    results/ subdirectory. Carries bin_index so the dashboard can pair this
    entry with its matching cls_json artefact for error-bar and theory-Cl overlay.

    Args:
        cov_result: Output dict from `compute_covariance`. Must contain at
            minimum ``covariance``, ``ells``, ``sqrt_diag``, and
            ``cl_theory_mode``.
        output_path: Destination JSON path.
        bin_index: Tomographic bin index — used to link with the cls_json
            artefact.
        generated_by: Infrastructure kwarg — omit in normal script usage.
        agent_name: Infrastructure kwarg — omit in normal script usage.

    Returns:
        ``dict`` with keys: ``path``, ``n_bandpowers``, ``cl_theory_mode``,
        ``provenance``.
    """
    return _save_covariance(
        cov_result,
        output_path,
        bin_index=bin_index,
        generated_by=generated_by,
        agent_name=agent_name,
    )

save_power_spectrum ¶

save_power_spectrum(cls_result: dict, output_path: str, generated_by: str | None = None, agent_name: str | None = None) -> dict

Save a power spectrum result dict to a JSON file.

Creates parent dirs, writes a _provenance block, and auto-registers the artefact in artefact_registry.json when output_path is inside a results/ subdirectory. The caller's dict is never mutated.

The optional generated_by and agent_name kwargs are auto-populated by the executor — omit them in all normal script usage.

Parameters:

Name	Type	Description	Default
`cls_result`	`dict`	Power-spectrum result dict to save.	required
`output_path`	`str`	Destination JSON path.	required
`generated_by`	`str \| None`	Infrastructure kwarg — omit in normal script usage.	`None`
`agent_name`	`str \| None`	Infrastructure kwarg — omit in normal script usage.	`None`

Returns:

Type	Description
`dict`	`dict` with the saved artefact's path and provenance.

Source code in cosmotron_mcp/server.py

@mcp.tool()
@sync_budget_guard
def save_power_spectrum(
    cls_result: dict,
    output_path: str,
    generated_by: str | None = None,
    agent_name: str | None = None,
) -> dict:
    """Save a power spectrum result dict to a JSON file.

    Creates parent dirs, writes a _provenance block, and auto-registers the
    artefact in artefact_registry.json when output_path is inside a results/
    subdirectory. The caller's dict is never mutated.

    The optional generated_by and agent_name kwargs are auto-populated by the
    executor — omit them in all normal script usage.

    Args:
        cls_result: Power-spectrum result dict to save.
        output_path: Destination JSON path.
        generated_by: Infrastructure kwarg — omit in normal script usage.
        agent_name: Infrastructure kwarg — omit in normal script usage.

    Returns:
        ``dict`` with the saved artefact's path and provenance.
    """
    return _save_power_spectrum(
        cls_result,
        output_path,
        generated_by=generated_by,
        agent_name=agent_name,
    )

save_healpix_map ¶

save_healpix_map(healpix_map: list, output_path: str, role: str = 'delta_map', nside: int | None = None, generated_by: str | None = None, agent_name: str | None = None, overwrite: bool = True) -> dict

Save a HEALPix map to a FITS file with provenance headers and auto-registration.

Creates the parent directory if needed, writes provenance FITS header keywords, and auto-registers the artefact in artefact_registry.json when output_path is inside a results/ subdirectory.

Parameters:

Name	Type	Description	Default
`healpix_map`	`list`	1-D HEALPix map as a list or np.ndarray.	required
`output_path`	`str`	Destination FITS file path.	required
`role`	`str`	One of `"delta_map"`, `"mask"`, `"systematic"`.	`'delta_map'`
`nside`	`int \| None`	HEALPix NSIDE. Inferred from map length if not supplied.	`None`
`generated_by`	`str \| None`	Infrastructure kwarg — omit in normal script usage.	`None`
`agent_name`	`str \| None`	Infrastructure kwarg — omit in normal script usage.	`None`
`overwrite`	`bool`	Overwrite an existing file.	`True`

Returns:

Type	Description
`dict`	`dict` with keys: `path`, `nside`, `role`, `n_pixels`,
`dict`	`provenance`.

Source code in cosmotron_mcp/server.py

@mcp.tool()
@sync_budget_guard
def save_healpix_map(
    healpix_map: list,
    output_path: str,
    role: str = "delta_map",
    nside: int | None = None,
    generated_by: str | None = None,
    agent_name: str | None = None,
    overwrite: bool = True,
) -> dict:
    """Save a HEALPix map to a FITS file with provenance headers and auto-registration.

    Creates the parent directory if needed, writes provenance FITS header
    keywords, and auto-registers the artefact in artefact_registry.json when
    output_path is inside a results/ subdirectory.

    Args:
        healpix_map: 1-D HEALPix map as a list or np.ndarray.
        output_path: Destination FITS file path.
        role: One of ``"delta_map"``, ``"mask"``, ``"systematic"``.
        nside: HEALPix NSIDE. Inferred from map length if not supplied.
        generated_by: Infrastructure kwarg — omit in normal script usage.
        agent_name: Infrastructure kwarg — omit in normal script usage.
        overwrite: Overwrite an existing file.

    Returns:
        ``dict`` with keys: ``path``, ``nside``, ``role``, ``n_pixels``,
        ``provenance``.
    """
    return _save_healpix_map(
        healpix_map,
        output_path,
        role=role,
        nside=nside,
        generated_by=generated_by,
        agent_name=agent_name,
        overwrite=overwrite,
    )

save_data_vector_sacc ¶

save_data_vector_sacc(output_path: str, tracers: list[dict], data_points: list[dict], covariance: list[list[float]] | None = None, overwrite: bool = True, session_dir: str | None = None) -> dict

Save a SACC data vector to a FITS file with Firecrown-compatible conventions.

PREFER assemble_sacc_from_session for standard clustering/shear sessions. Call this directly ONLY for CMB-lensing or shear×density cross-types with no composite tool.

GATED when the output lands inside a session. This tool takes an output_path, not a session_dir, so it used to be the one route to a FITS data vector that bypassed BOTH the plan gate and the systematics gate that assemble_sacc_from_session enforces. Writing into workspace/<session>/ now requires session_dir and passes those gates; a path outside any session (the genuine standalone use) is unaffected.

Supports NZ (galaxy clustering: lens0/lens1, weak lensing: src0/src1), Map (CMB lensing: cmbk), and NuMap (frequency maps: B20_T/B20_P) tracers.

Tracer names MUST follow Firecrown patterns: lens0/src0/cmbk/etc. Data type strings MUST use Firecrown measurement strings (e.g. "galaxy_density_cl", "galaxy_shearDensity_cl_e"), NOT raw spin strings (cl_00, cl_0e). Tracer pair ordering must satisfy Firecrown enum ordering: CMB < Clusters < src (SHEAR_E) < lens (COUNTS).

Parameters:

Name	Type	Description	Default
`output_path`	`str`	Path for the output FITS file.	required
`tracers`	`list[dict]`	Each dict must have a `"type"` key (`"NZ"`, `"Map"`, or `"NuMap"`) plus type-specific keys: NZ tracer (galaxy clustering / weak lensing):: `{"type": "NZ", "name": "lens0", "quantity": "galaxy_density", "spin": 0, "z": [0.0, 0.1, ...], "nz": [0.0, 0.05, ...]} Optional: "sigma_g" (float, shear intrinsic dispersion for WL).` Map tracer (CMB lensing):: `{"type": "Map", "name": "cmbk", "quantity": "cmb_convergence", "spin": 0, "ell": [0, 1, 2, ...], "beam": [1.0, 0.99, ...]}` NuMap tracer (frequency map):: `{"type": "NuMap", "name": "B20_T", "quantity": "cmb_temperature", "spin": 0, "nu": [1.0, ...], "bandpass": [0.0, ...], "ell": [0, 1, ...], "beam": [1.0, ...]} Optional: "nu_unit" (default "GHz"), "map_unit" (default "uK_RJ").`	required
`data_points`	`list[dict]`	Each dict must contain: `data_type` (Firecrown string), `tracer1`, `tracer2`, `ells` (list[float]), `cls` (list[float]). Optional: `window_ells` (list), `window` (2D list, n_bandpower × n_ell).	required
`covariance`	`list[list[float]] \| None`	Full covariance matrix as a 2D list; omitted if `None`.	`None`
`overwrite`	`bool`	Overwrite an existing file.	`True`

Returns:

Type	Description
`dict`	`dict` with keys: `sacc_path`, `n_tracers`, `n_data_points`,
`dict`	`has_covariance`, `tracer_names`.

Source code in cosmotron_mcp/server.py

@mcp.tool()
@sync_budget_guard
def save_data_vector_sacc(
    output_path: str,
    tracers: list[dict],
    data_points: list[dict],
    covariance: list[list[float]] | None = None,
    overwrite: bool = True,
    session_dir: str | None = None,
) -> dict:
    """Save a SACC data vector to a FITS file with Firecrown-compatible conventions.

    PREFER `assemble_sacc_from_session` for standard clustering/shear
    sessions. Call this directly ONLY for CMB-lensing or shear×density
    cross-types with no composite tool.

    GATED when the output lands inside a session. This tool takes an
    `output_path`, not a `session_dir`, so it used to be the one route to a FITS
    data vector that bypassed BOTH the plan gate and the systematics gate that
    `assemble_sacc_from_session` enforces. Writing into `workspace/<session>/`
    now requires `session_dir` and passes those gates; a path outside any
    session (the genuine standalone use) is unaffected.

    Supports NZ (galaxy clustering: lens0/lens1, weak lensing: src0/src1),
    Map (CMB lensing: cmbk), and NuMap (frequency maps: B20_T/B20_P) tracers.

    Tracer names MUST follow Firecrown patterns: lens0/src0/cmbk/etc.
    Data type strings MUST use Firecrown measurement strings (e.g.
    "galaxy_density_cl", "galaxy_shearDensity_cl_e"), NOT raw spin strings
    (cl_00, cl_0e). Tracer pair ordering must satisfy Firecrown enum ordering:
    CMB < Clusters < src (SHEAR_E) < lens (COUNTS).

    Args:
        output_path: Path for the output FITS file.
        tracers: Each dict must have a ``"type"`` key (``"NZ"``, ``"Map"``,
            or ``"NuMap"``) plus type-specific keys:

            NZ tracer (galaxy clustering / weak lensing)::

                {"type": "NZ", "name": "lens0", "quantity": "galaxy_density",
                 "spin": 0, "z": [0.0, 0.1, ...], "nz": [0.0, 0.05, ...]}
                Optional: "sigma_g" (float, shear intrinsic dispersion for WL).

            Map tracer (CMB lensing)::

                {"type": "Map", "name": "cmbk", "quantity": "cmb_convergence",
                 "spin": 0, "ell": [0, 1, 2, ...], "beam": [1.0, 0.99, ...]}

            NuMap tracer (frequency map)::

                {"type": "NuMap", "name": "B20_T", "quantity": "cmb_temperature",
                 "spin": 0, "nu": [1.0, ...], "bandpass": [0.0, ...],
                 "ell": [0, 1, ...], "beam": [1.0, ...]}
                Optional: "nu_unit" (default "GHz"), "map_unit" (default "uK_RJ").
        data_points: Each dict must contain: ``data_type`` (Firecrown
            string), ``tracer1``, ``tracer2``, ``ells`` (list[float]),
            ``cls`` (list[float]). Optional: ``window_ells`` (list),
            ``window`` (2D list, n_bandpower × n_ell).
        covariance: Full covariance matrix as a 2D list; omitted if `None`.
        overwrite: Overwrite an existing file.

    Returns:
        ``dict`` with keys: ``sacc_path``, ``n_tracers``, ``n_data_points``,
        ``has_covariance``, ``tracer_names``.
    """
    _sd = session_dir or _session_dir_for_output(output_path)
    if _sd is not None:
        if session_dir is None:
            return {"needs_input": (
                f"output_path {output_path!r} writes into session {_sd!r}, so this "
                "call must be gated like assemble_sacc_from_session. Pass "
                "session_dir=<that session> (the plan and systematics gates are "
                "then enforced), or prefer assemble_sacc_from_session, which "
                "builds the data vector from the session's own artefacts. Write "
                "outside any workspace session only for a genuinely standalone "
                "SACC file."
            )}
        _require_approved_plan(_sd)
        _require_resolved_dataset_decision(_sd)
        from cosmotron_mcp.systematics_tools import enforce_systematics_gate
        _gate = enforce_systematics_gate(_sd)
        if _gate is not None:
            return _gate
    return _save_data_vector_sacc(
        output_path,
        tracers,
        data_points,
        covariance=covariance,
        overwrite=overwrite,
    )

find_file ¶

find_file(filename: str, search_dir: str = 'workspace') -> dict

Search recursively for a file by name under search_dir.

Useful when the exact path is uncertain (e.g. after DataIngestor writes to a session directory whose name was not communicated to downstream agents). Returns the path of the lexicographically first match found.

Parameters:

Name	Type	Description	Default
`filename`	`str`	Base filename to search for, e.g. `"standardised_catalogue.fits"`.	required
`search_dir`	`str`	Root directory to search under.	`'workspace'`

Returns:

Type	Description
`dict`	`dict` with keys: `found` (str — path of the first match),
`dict`	`all_matches` (list[str] — all matching paths, sorted),
`dict`	`n_matches` (int).

Source code in cosmotron_mcp/server.py

@mcp.tool()
@sync_budget_guard
def find_file(filename: str, search_dir: str = "workspace") -> dict:
    """Search recursively for a file by name under search_dir.

    Useful when the exact path is uncertain (e.g. after DataIngestor writes
    to a session directory whose name was not communicated to downstream agents).
    Returns the path of the lexicographically first match found.

    Args:
        filename: Base filename to search for, e.g.
            ``"standardised_catalogue.fits"``.
        search_dir: Root directory to search under.

    Returns:
        ``dict`` with keys: ``found`` (str — path of the first match),
        ``all_matches`` (list[str] — all matching paths, sorted),
        ``n_matches`` (int).
    """
    return _find_file(filename, search_dir=search_dir)

read_session_context_tool ¶

read_session_context_tool(session_dir: str) -> dict

Read session_context.json from the given session directory.

Returns the full session_context dict including pipeline_config. Use this at the start of every analysis task to obtain the authoritative nside, lmax, lmin, n_bandpowers, bandwidth, apodise_mask, and full_sky.

Raises:

Type	Description
`FileNotFoundError`	If session_context.json is missing (i.e. DataIngestor has not yet completed).

Parameters:

Name	Type	Description	Default
`session_dir`	`str`	Session workspace directory.	required

Returns:

Type	Description
`dict`	The full session_context dict, including `pipeline_config`.

Source code in cosmotron_mcp/server.py

@mcp.tool()
@sync_budget_guard
def read_session_context_tool(session_dir: str) -> dict:
    """Read session_context.json from the given session directory.

    Returns the full session_context dict including pipeline_config.
    Use this at the start of every analysis task to obtain the authoritative
    nside, lmax, lmin, n_bandpowers, bandwidth, apodise_mask, and full_sky.

    Raises:
        FileNotFoundError: If session_context.json is missing (i.e.
            DataIngestor has not yet completed).

    Args:
        session_dir: Session workspace directory.

    Returns:
        The full session_context dict, including ``pipeline_config``.
    """
    return _read_session_context_tool(session_dir)

register_artefact ¶

register_artefact(session_dir: str, artefact_type: str, artefact_path: str, generated_by: str | None = None, agent_name: str | None = None, bin_index: int | None = None, metadata: dict | None = None) -> dict

Register a result artefact in the session artefact registry.

Only needed for artefacts NOT produced by save_power_spectrum, run_null_test, or compute_covariance — those three auto-register when output_path is inside results/. Use register_artefact for: HEALPix maps, SACC files, custom plots, or any output file not written by a pipeline tool.

Parameters:

Name	Type	Description	Default
`session_dir`	`str`	Session workspace directory.	required
`artefact_type`	`str`	One of: `cls_json`, `null_test`, `covariance`, `healpix_map`, `sacc_file`, `script`, `plot`, `posterior`.	required
`artefact_path`	`str`	Path to the artefact, relative to `session_dir`.	required
`generated_by`	`str \| None`	Script path (optional — auto-populated from the executor env if absent).	`None`
`agent_name`	`str \| None`	Agent name (optional — defaults to `"AnalysisCoder"` if absent).	`None`
`bin_index`	`int \| None`	Tomographic bin index, or `None` for non-bin artefacts.	`None`
`metadata`	`dict \| None`	Type-specific extra fields dict (e.g. `{"nside": 64, "role": "mask"}`).	`None`

Returns:

Type	Description
`dict`	`{"registered": True, "registry_path": ..., "entry": ...}`.

Source code in cosmotron_mcp/server.py

@mcp.tool()
@sync_budget_guard
def register_artefact(
    session_dir: str,
    artefact_type: str,
    artefact_path: str,
    generated_by: str | None = None,
    agent_name: str | None = None,
    bin_index: int | None = None,
    metadata: dict | None = None,
) -> dict:
    """Register a result artefact in the session artefact registry.

    Only needed for artefacts NOT produced by save_power_spectrum,
    run_null_test, or compute_covariance — those three auto-register when
    output_path is inside results/. Use register_artefact for: HEALPix maps,
    SACC files, custom plots, or any output file not written by a pipeline
    tool.

    Args:
        session_dir: Session workspace directory.
        artefact_type: One of: ``cls_json``, ``null_test``, ``covariance``,
            ``healpix_map``, ``sacc_file``, ``script``, ``plot``,
            ``posterior``.
        artefact_path: Path to the artefact, relative to `session_dir`.
        generated_by: Script path (optional — auto-populated from the
            executor env if absent).
        agent_name: Agent name (optional — defaults to ``"AnalysisCoder"``
            if absent).
        bin_index: Tomographic bin index, or `None` for non-bin artefacts.
        metadata: Type-specific extra fields dict (e.g.
            ``{"nside": 64, "role": "mask"}``).

    Returns:
        ``{"registered": True, "registry_path": ..., "entry": ...}``.
    """
    _require_human_gates(session_dir)
    return _register_artefact(
        session_dir,
        artefact_type,
        artefact_path,
        generated_by=generated_by,
        agent_name=agent_name,
        bin_index=bin_index,
        metadata=metadata,
    )

get_provenance ¶

get_provenance(session_dir: str) -> dict

Return the pipeline_config_provenance dict from session_context.json. Returns an empty dict if provenance was not recorded (legacy sessions). Use this to check whether a parameter was user-specified before modifying it in a script.

Example

prov = get_provenance(session_dir)
if prov.get("lmax", {}).get("source") == "user":
    lmax = cfg["lmax"]   # honour user value exactly

Parameters:

Name	Type	Description	Default
`session_dir`	`str`	Session workspace directory.	required

Returns:

Type	Description
`dict`	The `pipeline_config_provenance` dict, or `{}` for legacy
`dict`	sessions that predate provenance recording.

Source code in cosmotron_mcp/server.py

@mcp.tool()
@sync_budget_guard
def get_provenance(session_dir: str) -> dict:
    """Return the pipeline_config_provenance dict from session_context.json.
    Returns an empty dict if provenance was not recorded (legacy sessions).
    Use this to check whether a parameter was user-specified before
    modifying it in a script.

    Example:
        ```python
        prov = get_provenance(session_dir)
        if prov.get("lmax", {}).get("source") == "user":
            lmax = cfg["lmax"]   # honour user value exactly
        ```

    Args:
        session_dir: Session workspace directory.

    Returns:
        The ``pipeline_config_provenance`` dict, or ``{}`` for legacy
        sessions that predate provenance recording.
    """
    return _get_provenance(session_dir)

load_text_columns ¶

load_text_columns(path: str, max_rows: int | None = None) -> dict

Read a plain-text columnar file (.dat/.txt/.csv) as JSON columns.

The canonical way to read n(z) / theory Cℓ files — the opencode read tool rejects .dat as binary. Never use read on data files.

Parameters:

Name	Type	Description	Default
`path`	`str`	Path to the text file.	required
`max_rows`	`int \| None`	Optional cap on the number of rows read.	`None`

Returns:

Type	Description
`dict`	`{path, ncols, nrows, columns}` where `columns` is column-major
`dict`	(one list per column).

Source code in cosmotron_mcp/server.py

@mcp.tool()
@sync_budget_guard
def load_text_columns(path: str, max_rows: int | None = None) -> dict:
    """Read a plain-text columnar file (.dat/.txt/.csv) as JSON columns.

    The canonical way to read n(z) / theory Cℓ files — the opencode `read` tool
    rejects `.dat` as binary. Never use `read` on data files.

    Args:
        path: Path to the text file.
        max_rows: Optional cap on the number of rows read.

    Returns:
        ``{path, ncols, nrows, columns}`` where ``columns`` is column-major
        (one list per column).
    """
    return _load_text_columns(path, max_rows=max_rows)

cosmotron_mcp.tools.io¶

save_covariance ¶

save_power_spectrum ¶

save_healpix_map ¶

save_data_vector_sacc ¶

find_file ¶

read_session_context_tool ¶

register_artefact ¶

get_provenance ¶

load_text_columns ¶

`cosmotron_mcp.tools.io`¶