`cosmotron_mcp.server`¶

bootstrap_session ¶

bootstrap_session(data_dir: str, user_task: str, force_new: bool = False, survey: str | None = None, resume: bool = False, fresh_confirmed: bool = False, session_dir: str | None = None) -> dict

Create the session directory and write session_context.json deterministically.

Scans data_dir for the catalogue (first-level .fits/.hdf5/.csv) and an optional DATA_DESCRIPTION.md, then resolves pipeline_config from the user task by pure-Python regex parsing (authority: user > scb > profile > data_description > default) and writes the validated session_context.json. No LLM chooses pipeline parameters.

Pass survey to match a declarative catalogue-type profile (by canonical name or alias, e.g. survey="glass"). A matched profile supplies default pipeline params (recorded with provenance source "profile") and column conventions later honoured by ingest_to_session. An unknown/omitted survey falls back to fully dynamic resolution + detection.

Full-sky data (declared in the task or DATA_DESCRIPTION.md) forces apodise_mask=False and apodisation_scale=None; the write step rejects any inconsistent combination.

Default = ALWAYS a new session. Every call auto-creates a fresh workspace/{date}{stem}/ (or workspace/{date}_02/ etc. if one already exists for today), regardless of any prior session for this data_dir. Re-using a session is the exception, not the default.

Reuse is explicit-only: pass session_dir=<path> (or embed a session_dir: <path> line in user_task — the tool parses it; the explicit param wins if both are given) naming an EXISTING session directory. That exact session is then resumed verbatim (its pinned pipeline_config is kept; plan.json is deleted so the orchestrator re-plans). If the named path has no session_context.json, the tool returns session_dir_not_found=True and creates NOTHING — it never silently creates a session at an unrecognised path or falls back to a fresh one.

force_new/resume/fresh_confirmed remain accepted for backward compatibility with older callers/tests, but only matter when an explicit session_dir (param or task line) is present. A PARTIAL session (an in-flight bootstrap of THIS task that stopped at an unresolved nside / probe / role-conflict clarification) is transparently reused on the clarification retry so one task never spawns several workspace folders — this anti-churn is unrelated to user-facing session reuse.

bootstrap_session is registry-agnostic. It never asks whether a catalogue matches a registered dataset; that decision is a data-ingest concern and lives in ingest_to_session, which opens its own file-backed registry-<basename> gate (options: candidate dataset ids + "register-new" + "ignore") when the description hash matches an existing dataset, and a registration-<basename> gate (options: "register" + "ignore") after standardising a previously unseen dataset. Both gates are closed by authorise_gate (permission: ask picker). No registry state is inherited across independent user tasks because none is stored at bootstrap time.

Spin / probe: spin is 0 (galaxy density) unless the task DECLARES otherwise — spin=2, probe=shear (or probe=galaxy_shear), or a ## probes: block (per-probe overrides of spin/lmax/lmin/galaxy_bias/mask_path/sigma_e; nside is global and may NOT appear there). Spin is NEVER inferred from prose like "cosmic shear".

Correlation selection: a ## correlations: section pins which pairs to compute — all (default), auto, adjacent: N, or pairs: [[0,1],[lens0,src1]]. Stored in pipeline_config.pairs and applied consistently to the spectra, covariance, and SACC steps (early subset — the covariance is only built for the selected pairs). If the task prose implies shear but nothing is declared, the tool returns probe_unresolved: True + probe_hint — ask the user for an explicit probe=/spin=. Note: the CURRENT_SESSION pointer IS written even in this case (so the clarification retry reuses the same partial directory instead of forking a new one), but get_active_session refuses to serve a session until bootstrap_complete is set — so no other caller can pick up this partial session in the meantime.

Returns a dict with: session_dir, catalogue_path, pipeline_config, pipeline_config_provenance, required_plots, nside_unresolved, probe_unresolved, catalogue_role_conflict, and (when an explicit session_dir names a path with no session_context.json) session_dir_not_found=True. When nside_unresolved is True the user task did not specify nside (no default) — ask the user, never guess.

Parameters:

Name	Type	Description	Default
`data_dir`	`str`	Directory holding the raw catalogue and any ancillary files (masks, n(z), `DATA_DESCRIPTION.md`).	required
`user_task`	`str`	The user's task text, verbatim — parsed deterministically for pipeline_config and the structured `## section:` blocks (see the "Writing TASK.md" docs page), including an optional `session_dir: <path>` line requesting reuse of that exact session.	required
`force_new`	`bool`	Legacy no-op on the default (new-session) path; only interacts with an explicit `session_dir` reuse request.	`False`
`survey`	`str \| None`	Catalogue-type profile name/alias to apply (e.g. `survey="glass"`).	`None`
`resume`	`bool`	Legacy alias for reusing the session named by `session_dir` (param or task line); ignored when no `session_dir` is given.	`False`
`fresh_confirmed`	`bool`	Legacy no-op; kept for backward compatibility.	`False`
`session_dir`	`str \| None`	Path to an EXISTING session directory to resume verbatim. This is the ONLY way a run reuses a prior session — omit it (the default) to always start fresh.	`None`

Returns:

Type	Description
`dict`	`dict` with: `session_dir`, `catalogue_path`,
`dict`	`pipeline_config`, `pipeline_config_provenance`,
`dict`	`required_plots`, `nside_unresolved`, `probe_unresolved`,
`dict`	`catalogue_role_conflict`, and (when an explicit `session_dir`
`dict`	names a path with no session_context.json) `session_dir_not_found`.

Source code in cosmotron_mcp/server.py

@mcp.tool()
@sync_budget_guard
def bootstrap_session(
    data_dir: str,
    user_task: str,
    force_new: bool = False,
    survey: str | None = None,
    resume: bool = False,
    fresh_confirmed: bool = False,
    session_dir: str | None = None,
) -> dict:
    """Create the session directory and write session_context.json deterministically.

    Scans data_dir for the catalogue (first-level .fits/.hdf5/.csv) and an
    optional DATA_DESCRIPTION.md, then resolves pipeline_config from the user
    task by pure-Python regex parsing (authority: user > scb > profile >
    data_description > default) and writes the validated session_context.json.
    No LLM chooses pipeline parameters.

    Pass ``survey`` to match a declarative catalogue-type profile (by canonical
    name or alias, e.g. ``survey="glass"``). A matched profile supplies default
    pipeline params (recorded with provenance source ``"profile"``) and column
    conventions later honoured by ingest_to_session. An unknown/omitted survey
    falls back to fully dynamic resolution + detection.

    Full-sky data (declared in the task or DATA_DESCRIPTION.md) forces
    apodise_mask=False and apodisation_scale=None; the write step rejects any
    inconsistent combination.

    **Default = ALWAYS a new session.** Every call auto-creates a fresh
    workspace/{date}_{stem}/ (or workspace/{date}_{stem}_02/ etc. if one
    already exists for today), regardless of any prior session for this
    data_dir. Re-using a session is the exception, not the default.

    **Reuse is explicit-only**: pass ``session_dir=<path>`` (or embed a
    ``session_dir: <path>`` line in ``user_task`` — the tool parses it; the
    explicit param wins if both are given) naming an EXISTING session
    directory. That exact session is then resumed verbatim (its pinned
    pipeline_config is kept; ``plan.json`` is deleted so the orchestrator
    re-plans). If the named path has no ``session_context.json``, the tool
    returns ``session_dir_not_found=True`` and creates NOTHING — it never
    silently creates a session at an unrecognised path or falls back to a
    fresh one.

    ``force_new``/``resume``/``fresh_confirmed`` remain accepted for backward
    compatibility with older callers/tests, but only matter when an explicit
    ``session_dir`` (param or task line) is present. A PARTIAL session (an
    in-flight bootstrap of THIS task that stopped at an unresolved
    nside / probe / role-conflict clarification) is transparently reused on
    the clarification retry so one task never spawns several workspace
    folders — this anti-churn is unrelated to user-facing session reuse.

    **bootstrap_session is registry-agnostic.** It never asks whether a
    catalogue matches a registered dataset; that decision is a data-ingest
    concern and lives in ``ingest_to_session``, which opens its own
    file-backed ``registry-<basename>`` gate (options: candidate dataset ids
    + ``"register-new"`` + ``"ignore"``) when the description hash matches an
    existing dataset, and a ``registration-<basename>`` gate (options:
    ``"register"`` + ``"ignore"``) after standardising a previously unseen
    dataset. Both gates are closed by ``authorise_gate`` (``permission: ask``
    picker). No registry state is inherited across independent user tasks
    because none is stored at bootstrap time.

    Spin / probe: spin is 0 (galaxy density) unless the task DECLARES otherwise —
    ``spin=2``, ``probe=shear`` (or ``probe=galaxy_shear``), or a ``## probes:``
    block (per-probe overrides of spin/lmax/lmin/galaxy_bias/mask_path/sigma_e;
    ``nside`` is global and may NOT appear there). Spin is NEVER inferred from prose
    like "cosmic shear".

    Correlation selection: a ``## correlations:`` section pins which pairs to
    compute — ``all`` (default), ``auto``, ``adjacent: N``, or
    ``pairs: [[0,1],[lens0,src1]]``. Stored in ``pipeline_config.pairs`` and applied
    consistently to the spectra, covariance, and SACC steps (early subset — the
    covariance is only built for the selected pairs). If the task prose implies shear but nothing is declared,
    the tool returns ``probe_unresolved: True`` + ``probe_hint`` — ask the user
    for an explicit ``probe=``/``spin=``. Note: the CURRENT_SESSION pointer IS
    written even in this case (so the clarification retry reuses the same
    partial directory instead of forking a new one), but ``get_active_session``
    refuses to serve a session until ``bootstrap_complete`` is set — so no
    other caller can pick up this partial session in the meantime.

    Returns a dict with: session_dir, catalogue_path, pipeline_config,
    pipeline_config_provenance, required_plots, nside_unresolved,
    probe_unresolved, catalogue_role_conflict, and (when an explicit
    session_dir names a path with no session_context.json)
    session_dir_not_found=True. When nside_unresolved is True the user task
    did not specify nside (no default) — ask the user, never guess.

    Args:
        data_dir: Directory holding the raw catalogue and any ancillary
            files (masks, n(z), `DATA_DESCRIPTION.md`).
        user_task: The user's task text, verbatim — parsed deterministically
            for pipeline_config and the structured `## section:` blocks (see
            the "Writing TASK.md" docs page), including an optional
            `session_dir: <path>` line requesting reuse of that exact session.
        force_new: Legacy no-op on the default (new-session) path; only
            interacts with an explicit `session_dir` reuse request.
        survey: Catalogue-type profile name/alias to apply (e.g.
            ``survey="glass"``).
        resume: Legacy alias for reusing the session named by `session_dir`
            (param or task line); ignored when no `session_dir` is given.
        fresh_confirmed: Legacy no-op; kept for backward compatibility.
        session_dir: Path to an EXISTING session directory to resume
            verbatim. This is the ONLY way a run reuses a prior session —
            omit it (the default) to always start fresh.

    Returns:
        ``dict`` with: ``session_dir``, ``catalogue_path``,
        ``pipeline_config``, ``pipeline_config_provenance``,
        ``required_plots``, ``nside_unresolved``, ``probe_unresolved``,
        ``catalogue_role_conflict``, and (when an explicit `session_dir`
        names a path with no session_context.json) ``session_dir_not_found``.
    """
    # ── Explicit session reuse (the ONLY sanctioned path) ───────────────────
    # Default = always a new session. Reuse happens ONLY when the caller (or
    # the task text) names an exact session_dir to resume. force_new/resume/
    # fresh_confirmed are legacy params kept for callers/tests but no longer
    # gate a prior *finalised* session for this data_dir — that gate is gone.
    requested_dir = session_dir or _explicit_session_dir_from_task(user_task)
    if requested_dir is None and resume:
        # `resume=True` with no session_dir means "the session I am already in".
        # It used to work only by accident, via the partial-session anti-churn
        # path — which now (correctly) refuses to re-enter a session that has
        # already ingested, so an explicit resume has to resolve the active
        # session itself rather than silently forking a new one.
        _active = _active_session_for(data_dir)
        if _active is not None:
            requested_dir = _active["session_dir"]
    if requested_dir:
        existing = _load_session_by_dir(requested_dir)
        if existing is None:
            return {
                "needs_input": (
                    f"session_dir '{requested_dir}' does not exist — omit "
                    "session_dir to start a new session, or point it at an "
                    "existing workspace/... session directory"
                ),
                "session_dir_not_found": True,
                "gate_id": "session-dir-not-found",
            }
        pc = existing.get("pipeline_config", {})
        # Reset plan.json on resume so the orchestrator is forced through @planner again.
        # A leftover approved plan from a prior run would let the orchestrator skip the
        # planning gate entirely and dispatch stale steps. write_plan() will overwrite this.
        (Path(existing["session_dir"]) / "plan.json").unlink(missing_ok=True)
        _write_current_session(existing["session_dir"])
        return {
            "session_dir": existing["session_dir"],
            "catalogue_path": existing.get("catalogue_path", ""),
            "pipeline_config": pc,
            "pipeline_config_provenance": existing.get("pipeline_config_provenance", {}),
            "required_plots": existing.get("required_plots", []),
            "nside_unresolved": pc.get("nside") is None,
            "resumed": True,
        }

    # ── Default path: always a new session ──────────────────────────────────
    # A partial session — a prior bootstrap attempt on this data_dir that
    # stopped at an unresolved nside / probe / role-conflict clarification —
    # is transparently reused on the clarification retry so the corrected
    # call lands in the same directory rather than spawning a fresh
    # workspace/{date}_{stem}_NN folder. That is the ONLY implicit reuse
    # bootstrap performs; the registry (reuse + registration) is handled
    # by ingest_to_session and its own file-backed gates, not here.
    existing = _active_session_for(data_dir)
    existing_partial = (
        existing is not None
        and not bool(existing.get("bootstrap_complete"))
        and not _session_has_work(existing.get("session_dir"))
    )
    reuse_dir: str | None = existing["session_dir"] if existing_partial else None

    ctx = _build_session_context(data_dir, session_dir=reuse_dir)
    # Track the session as active immediately — even if bootstrap stops at an
    # unresolved gate below (nside/probe/role-conflict). This is what lets the
    # retry find and REUSE this partial session (via _active_session_for)
    # instead of creating a fresh workspace folder each time. Finalisation is
    # marked separately via bootstrap_complete on the success path.
    _write_current_session(ctx["session_dir"])
    # Location-awareness (Phase R4): when the data lives on a remote site, pin
    # the session to it (remote_state.json) so every data-bound tool submits
    # there. Mixed sites across the enumerated inputs → stop and ask (never
    # guess which cluster the run belongs to).
    from cosmotron_mcp.remote.locations import parse_ref as _parse_ref
    _sites = {
        _parse_ref(f).site for f in ctx.get("all_data_files", [])
        if isinstance(f, str)
    } - {None}
    if len(_sites) > 1:
        return {
            "needs_input": (
                "the data inputs reference more than one execution site "
                f"({sorted(_sites)}); a session must live on a single site. "
                "Stage all inputs on one site, or start separate sessions."
            ),
            "session_dir": ctx["session_dir"],
            "gate_id": "remote-site-conflict",
        }
    prof = get_profile(survey)
    profile_defaults = prof.defaults.model_dump(exclude_none=True) if prof else None
    pc, prov = resolve_pipeline_config(
        user_task, None, ctx.get("data_description", ""), profile_defaults,
        profile_probe=(prof.probe if prof else None),
    )
    ctx["pipeline_config"] = pc
    ctx["pipeline_config_provenance"] = prov
    ctx["survey_profile"] = prof.name if prof else None

    # ── Reconcile remote-execution intent (task-triggerable remote) ───────────
    # Merge the task's `## remote:`/`## site:` spec (pc["remote"], site may be
    # None) with a `site:path` data_dir's execution_site into a single pinned
    # {site, steps}. Remote data forces `ingest` onto the cluster (a catalogue
    # there can't be read locally). No site + no remote data → runs fully local.
    _remote_gate = _reconcile_remote(ctx, pc)
    if _remote_gate is not None:
        return _remote_gate
    # Persist the full task text so later tools can read task-embedded config
    # (e.g. a '## firecrown config:' YAML block for inference). The SessionContext
    # schema already declares `task`; it just was never written.
    if user_task:
        ctx["task"] = user_task

    required_plots = extract_required_plots(user_task)
    if required_plots:
        ctx["required_plots"] = required_plots
    raw_plot_reqs = extract_plot_requirements_section(user_task)
    if raw_plot_reqs:
        ctx["plot_user_requirements"] = raw_plot_reqs
    ctx["skip_gates"] = extract_skip_gates(user_task)

    if pc.get("nside") is None:
        # Mirror the orchestrator: write the partial context so it persists,
        # and flag nside as unresolved for the caller to resolve with the user.
        ctx_path = Path(ctx["session_dir"]) / "session_context.json"
        ctx_path.write_text(json.dumps(ctx, indent=2), encoding="utf-8")
        return {**ctx, "nside_unresolved": True, "gate_id": "nside"}

    # Probe/spin consistency gate: the task prose implies shear/weak lensing but no
    # spin-2 probe was DECLARED (probe=/spin=/## probes:). Never guess spin — ask.
    probe_unresolved = _probe_unresolved(user_task, pc)
    if probe_unresolved:
        ctx_path = Path(ctx["session_dir"]) / "session_context.json"
        ctx_path.write_text(json.dumps(ctx, indent=2), encoding="utf-8")
        return {
            **ctx,
            "nside_unresolved": False,
            "probe_unresolved": True,
            "probe_hint": (
                "task mentions shear / weak lensing but no spin-2 probe is declared — "
                "add `probe=shear` (or `spin=2`, or a `## probes:` block) to the task, "
                "or drop the shear wording if this is a clustering analysis"
            ),
            "gate_id": "probe",
        }

    # Catalogue-role conflict gate. The task and DATA_DESCRIPTION.md each assign
    # a probe to a named catalogue file; when they DISAGREE about the same file,
    # one of them has the lens and source samples swapped. Picking either mapping
    # silently analyses the shear catalogue as clustering (and vice versa) — a
    # wrong-science outcome that no downstream check would catch (ses_05bf).
    from cosmotron_mcp.pipeline_resolve import extract_catalogue_roles
    _task_roles = extract_catalogue_roles(user_task or "")
    _desc_roles = extract_catalogue_roles(ctx.get("data_description", ""))
    _conflicts = {
        f: (_task_roles[f], _desc_roles[f])
        for f in set(_task_roles) & set(_desc_roles)
        if _task_roles[f] != _desc_roles[f]
    }
    if _conflicts:
        _write_session_context(ctx["session_dir"], ctx)
        _lines = "; ".join(
            f"{f}: task says {t}, DATA_DESCRIPTION.md says {d}"
            for f, (t, d) in sorted(_conflicts.items())
        )
        return {
            **ctx,
            "nside_unresolved": False,
            "probe_unresolved": False,
            "catalogue_role_conflict": True,
            "catalogue_role_conflict_hint": (
                f"the task and DATA_DESCRIPTION.md disagree about which probe "
                f"these catalogues hold — {_lines}. One of them has the lens "
                "(clustering) and source (shear) samples swapped, and analysing "
                "them the wrong way round is silent and unrecoverable. Ask the "
                "user which mapping is correct; do not pick one."
            ),
            "gate_id": "role-conflict",
        }

    # Mark the session finalised — every clarification gate passed.
    # bootstrap_session is registry-agnostic: any registered dataset match
    # is handled by ingest_to_session's own registry-<basename> gate, so
    # bootstrap completes cleanly whether or not the data has a match.
    ctx["bootstrap_complete"] = True
    _write_session_context(ctx["session_dir"], ctx)
    _write_current_session(ctx["session_dir"])
    return {
        "session_dir": ctx["session_dir"],
        "catalogue_path": ctx["catalogue_path"],
        "pipeline_config": pc,
        "pipeline_config_provenance": prov,
        "survey_profile": ctx.get("survey_profile"),
        "required_plots": ctx.get("required_plots", []),
        "nside_unresolved": False,
        "probe_unresolved": False,
    }

get_active_session ¶

get_active_session() -> dict

Return the currently active session directory and its pipeline_config.

Reads workspace/CURRENT_SESSION (written by bootstrap_session) and returns the authoritative session_dir + pipeline_config from session_context.json.

RECOVERY TOOL ONLY — use it mid-run when you already know a session was bootstrapped THIS run and you merely lost track of its path (never guess or pick the most recent workspace/ subdirectory by hand). On a brand-new task do NOT call this before @session_bootstrapper/bootstrap_session has actually run and returned a session_dir: CURRENT_SESSION is a cross-conversation file left over from whatever ran last (including an unrelated or cancelled prior run) — calling this first silently hands you a stale session instead of the fresh one the task asked for.

bootstrap_session writes the CURRENT_SESSION pointer as soon as it picks a directory — including when it then stops at an unresolved clarification (nside/probe/role-conflict) — so that a clarification retry lands back in the same partial session (anti-churn). This tool refuses to serve such a partial session (bootstrap_complete not yet True in session_context.json): resolving the clarification is @session_bootstrapper's job, not a recovery caller's.

Raises:

Type	Description
`RuntimeError`	If no session has been bootstrapped yet, or the pointed-at session is still partial (bootstrap did not finish).

Returns:

Type	Description
`dict`	`dict` with `session_dir`, `pipeline_config`,
`dict`	`catalogue_path`, `required_plots`.

Source code in cosmotron_mcp/server.py

@mcp.tool()
@sync_budget_guard
def get_active_session() -> dict:
    """Return the currently active session directory and its pipeline_config.

    Reads workspace/CURRENT_SESSION (written by bootstrap_session) and returns
    the authoritative session_dir + pipeline_config from session_context.json.

    RECOVERY TOOL ONLY — use it mid-run when you already know a session was
    bootstrapped THIS run and you merely lost track of its path (never guess
    or pick the most recent workspace/ subdirectory by hand). On a brand-new
    task do NOT call this before `@session_bootstrapper`/`bootstrap_session`
    has actually run and returned a session_dir: CURRENT_SESSION is a
    cross-conversation file left over from whatever ran last (including an
    unrelated or cancelled prior run) — calling this first silently hands you
    a stale session instead of the fresh one the task asked for.

    ``bootstrap_session`` writes the CURRENT_SESSION pointer as soon as it
    picks a directory — including when it then stops at an unresolved
    clarification (nside/probe/role-conflict) — so that a clarification
    retry lands back in the same partial session (anti-churn). This tool
    refuses to serve such a partial session (``bootstrap_complete`` not
    yet ``True`` in ``session_context.json``): resolving the clarification
    is `@session_bootstrapper`'s job, not a recovery caller's.

    Raises:
        RuntimeError: If no session has been bootstrapped yet, or the
            pointed-at session is still partial (bootstrap did not finish).

    Returns:
        ``dict`` with ``session_dir``, ``pipeline_config``,
        ``catalogue_path``, ``required_plots``.
    """
    if not _CURRENT_SESSION_FILE.exists():
        raise RuntimeError(
            "No active session. workspace/CURRENT_SESSION does not exist — "
            "run bootstrap_session first."
        )
    session_dir = _CURRENT_SESSION_FILE.read_text(encoding="utf-8").strip()
    ctx = _read_session_context_tool(session_dir)
    if not ctx.get("bootstrap_complete"):
        raise RuntimeError(
            f"bootstrap incomplete for '{session_dir}' — an unresolved "
            "clarification (nside/probe/role-conflict) stopped "
            "bootstrap_session before it finished. Re-dispatch "
            "@session_bootstrapper (with the user's answer to the "
            "outstanding NEEDS_INPUT) instead of calling get_active_session."
        )
    return {
        "session_dir": session_dir,
        "pipeline_config": ctx.get("pipeline_config", {}),
        "catalogue_path": ctx.get("catalogue_path", ""),
        "required_plots": ctx.get("required_plots", []),
    }

find_dataset ¶

find_dataset(data_dir: str | None = None, name: str | None = None) -> dict

Find registered datasets matching a data directory's DATA_DESCRIPTION.md or a name.

Candidates for the user to confirm before reuse.

Parameters:

Name	Type	Description	Default
`data_dir`	`str \| None`	Match by this directory's `DATA_DESCRIPTION.md` content.	`None`
`name`	`str \| None`	Match by dataset name.	`None`

Returns:

Type	Description
`dict`	`{"datasets": [{dataset_id, name, version, survey, created_at}, ...]}`.

Source code in cosmotron_mcp/server.py

@mcp.tool()
@sync_budget_guard
def find_dataset(data_dir: str | None = None, name: str | None = None) -> dict:
    """Find registered datasets matching a data directory's DATA_DESCRIPTION.md or a name.

    Candidates for the user to confirm before reuse.

    Args:
        data_dir: Match by this directory's `DATA_DESCRIPTION.md` content.
        name: Match by dataset name.

    Returns:
        ``{"datasets": [{dataset_id, name, version, survey, created_at}, ...]}``.
    """
    from cosmotron_mcp import registry
    description_hash = None
    if data_dir:
        desc = Path(data_dir) / "DATA_DESCRIPTION.md"
        if desc.exists():
            description_hash = registry.hash_description(desc.read_text(encoding="utf-8"))
    cands = registry.find_candidates(description_hash=description_hash, name=name)
    return {"datasets": [
        {"dataset_id": c.dataset_id, "name": c.name, "version": c.version,
         "survey": c.survey_profile, "created_at": c.created_at}
        for c in cands
    ]}

list_datasets ¶

list_datasets() -> dict

List all registered datasets in the persistent L1 registry.

Returns:

Type	Description
`dict`	``{"datasets": [{dataset_id, name, version, probe, survey, n_bins,
`dict`	n_objects, locations, created_at}, ...]}`.`locations`` names the
`dict`	machines that hold this dataset's products (`"local"` = the canonical
`dict`	store; a site name = a cluster copy from
`dict`	`cosmotron-data-registry push`).

Source code in cosmotron_mcp/server.py

@mcp.tool()
@sync_budget_guard
def list_datasets() -> dict:
    """List all registered datasets in the persistent L1 registry.

    Returns:
        ``{"datasets": [{dataset_id, name, version, probe, survey, n_bins,
        n_objects, locations, created_at}, ...]}``. ``locations`` names the
        machines that hold this dataset's products (``"local"`` = the canonical
        store; a site name = a cluster copy from
        ``cosmotron-data-registry push``).
    """
    from cosmotron_mcp import registry
    return {"datasets": [
        {"dataset_id": c.dataset_id, "name": c.name, "version": c.version,
         "probe": c.probe, "survey": c.survey_profile,
         "n_bins": c.stats.get("n_bins"), "n_objects": c.stats.get("n_objects"),
         "locations": sorted((c.locations or {}).keys()),
         "created_at": c.created_at}
        for c in registry.list_datasets()
    ]}

get_dataset ¶

get_dataset(dataset_id: str) -> dict

Full record for one registered dataset (the agent-facing equivalent of cosmotron-data-registry show).

Parameters:

Name	Type	Description	Default
`dataset_id`	`str`	The dataset id (e.g. `"glass__v1"`, from `list_datasets`).	required

Returns:

Type	Description
`dict`	``{dataset_id, name, version, probe, survey, coordinate_convention,
`dict`	source_data_dir, description_hash, z_edges, column_map, stats,
`dict`	products, locations, created_at}`— or``` if the
`dict`	id isn't registered. `locations` shows where each copy of the
`dict`	products physically lives (`"local"` + any site names); a dataset
`dict`	that lives only on a cluster must be pulled
`dict`	(`cosmotron-data-registry pull`) or used remotely.

Source code in cosmotron_mcp/server.py

@mcp.tool()
@sync_budget_guard
def get_dataset(dataset_id: str) -> dict:
    """Full record for one registered dataset (the agent-facing equivalent of
    ``cosmotron-data-registry show``).

    Args:
        dataset_id: The dataset id (e.g. ``"glass__v1"``, from `list_datasets`).

    Returns:
        ``{dataset_id, name, version, probe, survey, coordinate_convention,
        source_data_dir, description_hash, z_edges, column_map, stats,
        products, locations, created_at}`` — or ``{"needs_input": ...}`` if the
        id isn't registered. ``locations`` shows where each copy of the
        products physically lives (``"local"`` + any site names); a dataset
        that lives only on a cluster must be pulled
        (``cosmotron-data-registry pull``) or used remotely.
    """
    from cosmotron_mcp import registry
    r = registry.get(dataset_id)
    if r is None:
        return {"needs_input": (
            f"dataset_id '{dataset_id}' is not registered — use list_datasets "
            "to see available datasets."
        )}
    return {
        "dataset_id": r.dataset_id, "name": r.name, "version": r.version,
        "probe": r.probe, "survey": r.survey_profile,
        "coordinate_convention": r.coordinate_convention,
        "source_data_dir": r.source_data_dir,
        "description_hash": r.description_hash,
        "z_edges": r.z_edges, "column_map": r.column_map,
        "stats": r.stats, "products": r.products,
        "locations": r.locations, "created_at": r.created_at,
    }

cosmotron_mcp.server¶

bootstrap_session ¶

get_active_session ¶

find_dataset ¶

list_datasets ¶

get_dataset ¶

`cosmotron_mcp.server`¶