Tools

The functions behind the agent's tool calls. See Agent tools for what each one is for.

Results

helioai.tools.results

What a tool call returns, typed once at the registry boundary.

Every tool hands back a dict (or, for the remote MCP proxies and the internal tools, a string), and the registry used to serialise it to JSON on the spot. Downstream, five readers — the history writer, the artifact extractor, the figure review, the MCP server's isError, the event display — each parsed that string back and sniffed its shape. The ToolResult carries the parsed payload alongside the exact text the model receives, so the shape is read once and the text is produced once.

for_llm() is the contract that lets this module exist without changing agent behaviour: it returns, byte for byte, the string registry.call_tool used to return — a string result untouched, a dict as json.dumps(..., ensure_ascii=False, default=str), a failure as {"error": ...}. tests/test_tool_results.py holds it to captured results.

ToolResult dataclass

The outcome of one tool call: the parsed payload and the text the model sees.

Attributes:

Name	Type	Description
tool	str	Name of the tool that produced it.
payload	Any	What the tool returned — a dict for every shipped tool, a list or a scalar in principle, a string for a remote MCP proxy whose text is not JSON. A string that is JSON (the internal tools, the task result) is parsed here so readers see the object, while for_llm() still returns the original text.
raw	str | None	The exact string a tool returned, when it returned one. for_llm() hands it back untouched: re-serialising a parsed copy could change the escaping of a non-ASCII character, and the model must see what it saw.

Source code in helioai/tools/results.py

@dataclass
class ToolResult:
    """The outcome of one tool call: the parsed payload and the text the model sees.

    Attributes:
        tool: Name of the tool that produced it.
        payload: What the tool returned — a dict for every shipped tool, a list or a
            scalar in principle, a string for a remote MCP proxy whose text is not
            JSON. A string that *is* JSON (the internal tools, the `task` result) is
            parsed here so readers see the object, while `for_llm()` still returns
            the original text.
        raw: The exact string a tool returned, when it returned one. `for_llm()`
            hands it back untouched: re-serialising a parsed copy could change the
            escaping of a non-ASCII character, and the model must see what it saw.
    """

    tool: str
    payload: Any
    raw: str | None = field(default=None, repr=False)

    @classmethod
    def from_raw(cls, tool: str, obj: Any) -> ToolResult:
        """Wrap whatever a tool returned.

        Args:
            tool: The tool's name.
            obj: Its return value — dict, list, string, anything JSON can carry.

        Returns:
            A result whose `payload` is the object (a JSON string is parsed) and
            whose `for_llm()` is what the registry used to return for that value.
        """
        if isinstance(obj, str):
            try:
                parsed = json.loads(obj)
            except ValueError:
                parsed = obj
            return cls(tool, parsed, obj)
        return cls(tool, obj)

    @classmethod
    def failure(cls, tool: str, message: str) -> ToolResult:
        """A result for a tool that never ran, or raised.

        Args:
            tool: The tool that was asked for.
            message: Why it failed. An empty message is replaced by the word `error`
                rather than kept: `str(TimeoutError())` is `""`, every reader tests the
                error field for truth, and a tool that never ran was once displayed as ok.

        Returns:
            A result whose `for_llm()` is the `{"error": ...}` line the registry has
            always returned for these cases, ASCII-escaped as before.
        """
        payload = {"error": message or "error"}
        return cls(tool, payload, json.dumps(payload))

    @property
    def ok(self) -> bool:
        """False when the payload carries a truthy `error`, which is how every shipped
        tool — and the registry itself — reports a failure."""
        return not (isinstance(self.payload, dict) and self.payload.get("error"))

    @property
    def error(self) -> str | None:
        """The error message, or None for a result that is ok."""
        if self.ok:
            return None
        return str(self.payload["error"])

    def with_payload(self, payload: Any) -> ToolResult:
        """The same call with an amended payload — what the figure review returns.

        The original text is dropped on purpose: `for_llm()` must describe the payload
        the model is about to read, amendments included.
        """
        return ToolResult(self.tool, payload)

    def for_llm(self) -> str:
        """The text appended to the history and shown to the model.

        Returns:
            The tool's own string when it returned one; otherwise the payload as JSON
            with non-ASCII kept and unknown types stringified — the serialisation
            `registry.call_tool` has always used, so the model reads the same bytes.
        """
        if self.raw is not None:
            return self.raw
        if isinstance(self.payload, str):
            return self.payload
        return json.dumps(self.payload, ensure_ascii=False, default=str)

ok property

ok: bool

False when the payload carries a truthy error, which is how every shipped tool — and the registry itself — reports a failure.

error property

error: str | None

The error message, or None for a result that is ok.

from_raw classmethod

from_raw(tool: str, obj: Any) -> ToolResult

Wrap whatever a tool returned.

Parameters:

Name	Type	Description	Default
tool	str	The tool's name.	required
obj	Any	Its return value — dict, list, string, anything JSON can carry.	required

Returns:

Type	Description
ToolResult	A result whose payload is the object (a JSON string is parsed) and
ToolResult	whose for_llm() is what the registry used to return for that value.

Source code in helioai/tools/results.py

@classmethod
def from_raw(cls, tool: str, obj: Any) -> ToolResult:
    """Wrap whatever a tool returned.

    Args:
        tool: The tool's name.
        obj: Its return value — dict, list, string, anything JSON can carry.

    Returns:
        A result whose `payload` is the object (a JSON string is parsed) and
        whose `for_llm()` is what the registry used to return for that value.
    """
    if isinstance(obj, str):
        try:
            parsed = json.loads(obj)
        except ValueError:
            parsed = obj
        return cls(tool, parsed, obj)
    return cls(tool, obj)

failure classmethod

failure(tool: str, message: str) -> ToolResult

A result for a tool that never ran, or raised.

Parameters:

Name	Type	Description	Default
tool	str	The tool that was asked for.	required
message	str	Why it failed. An empty message is replaced by the word error rather than kept: str(TimeoutError()) is "", every reader tests the error field for truth, and a tool that never ran was once displayed as ok.	required

Returns:

Type	Description
ToolResult	A result whose for_llm() is the {"error": ...} line the registry has
ToolResult	always returned for these cases, ASCII-escaped as before.

Source code in helioai/tools/results.py

@classmethod
def failure(cls, tool: str, message: str) -> ToolResult:
    """A result for a tool that never ran, or raised.

    Args:
        tool: The tool that was asked for.
        message: Why it failed. An empty message is replaced by the word `error`
            rather than kept: `str(TimeoutError())` is `""`, every reader tests the
            error field for truth, and a tool that never ran was once displayed as ok.

    Returns:
        A result whose `for_llm()` is the `{"error": ...}` line the registry has
        always returned for these cases, ASCII-escaped as before.
    """
    payload = {"error": message or "error"}
    return cls(tool, payload, json.dumps(payload))

with_payload

with_payload(payload: Any) -> ToolResult

The same call with an amended payload — what the figure review returns.

The original text is dropped on purpose: for_llm() must describe the payload the model is about to read, amendments included.

Source code in helioai/tools/results.py

def with_payload(self, payload: Any) -> ToolResult:
    """The same call with an amended payload — what the figure review returns.

    The original text is dropped on purpose: `for_llm()` must describe the payload
    the model is about to read, amendments included.
    """
    return ToolResult(self.tool, payload)

for_llm

for_llm() -> str

The text appended to the history and shown to the model.

Returns:

Type	Description
str	The tool's own string when it returned one; otherwise the payload as JSON
str	with non-ASCII kept and unknown types stringified — the serialisation
str	registry.call_tool has always used, so the model reads the same bytes.

Source code in helioai/tools/results.py

def for_llm(self) -> str:
    """The text appended to the history and shown to the model.

    Returns:
        The tool's own string when it returned one; otherwise the payload as JSON
        with non-ASCII kept and unknown types stringified — the serialisation
        `registry.call_tool` has always used, so the model reads the same bytes.
    """
    if self.raw is not None:
        return self.raw
    if isinstance(self.payload, str):
        return self.payload
    return json.dumps(self.payload, ensure_ascii=False, default=str)

Parameter search

helioai.tools.rag

ChromaDB semantic search over the speasy catalog.

The index is built by indexer.py (run: helioai index). This module provides the read-only search path used at agent query time. Models load lazily and are cached at module scope.

search

search(query: str, top_k: int = 5, *, provider: str | None = None, region: str | None = None, measurement_type: str | None = None, window: tuple[str, str] | None = None) -> list[dict]

Semantic search over speasy catalog (single query).

Optional metadata filters narrow the search at query time (the metadata is already indexed by indexer.py): provider (amda/cda/csa/ssc) is the most useful — it counters CDA's dominance of the catalog.

With hybrid search enabled (default), a BM25 lexical channel is fused with the dense channel via RRF — this recovers exact id/code matches (e.g. BGSEc) that dense embeddings miss.

The returned ORDER is the ranking. score is only the fusion or cross-encoder value, frozen before the domain rerank reorders the list, so it does not decrease monotonically down the results and must not be re-sorted on — reading it as a ranking puts the demoted product back on top. It is kept here to tune the retrieval; search_parameters drops it before a model sees it.

For several queries at once use search_batch (one embedding pass + one Chroma call). Returns a list of dicts: {id, name, description, coverage, score}.

Parameters:

Name	Type	Description	Default
query	str	Free-text English description of ONE parameter.	required
top_k	int	Number of results.	5
provider	str | None	Restrict to one provider (amda/cda/csa/ssc).	None
region	str | None	SPASE region filter (exact indexed string).	None
measurement_type	str | None	Measurement-type filter (exact indexed string).	None
window	tuple[str, str] | None	(start, stop) the caller intends to download — see search_batch.	None

Example

search("ACE solar wind proton density", top_k=2)[0] {'id': 'cda/AC_H2_SWE/Np', 'name': 'Proton No. density', 'description': 'Proton No. density. Solar Wind Proton Number Density, ' 'scalar. ACE/SWEPAM ... Units: #/cc. ...', 'score': 1.0, ...}

Source code in helioai/tools/rag.py

def search(
    query: str,
    top_k: int = 5,
    *,
    provider: str | None = None,
    region: str | None = None,
    measurement_type: str | None = None,
    window: tuple[str, str] | None = None,
) -> list[dict]:
    """Semantic search over speasy catalog (single query).

    Optional metadata filters narrow the search at query time (the metadata is
    already indexed by indexer.py): `provider` (amda/cda/csa/ssc) is the most
    useful — it counters CDA's dominance of the catalog.

    With hybrid search enabled (default), a BM25 lexical channel is fused with
    the dense channel via RRF — this recovers exact id/code matches (e.g.
    `BGSEc`) that dense embeddings miss.

    The returned ORDER is the ranking. `score` is only the fusion or cross-encoder
    value, frozen before the domain rerank reorders the list, so it does not decrease
    monotonically down the results and must not be re-sorted on — reading it as a
    ranking puts the demoted product back on top. It is kept here to tune the
    retrieval; `search_parameters` drops it before a model sees it.

    For several queries at once use `search_batch` (one embedding pass + one
    Chroma call). Returns a list of dicts: {id, name, description, coverage, score}.

    Args:
        query: Free-text English description of ONE parameter.
        top_k: Number of results.
        provider: Restrict to one provider (amda/cda/csa/ssc).
        region: SPASE region filter (exact indexed string).
        measurement_type: Measurement-type filter (exact indexed string).
        window: `(start, stop)` the caller intends to download — see `search_batch`.

    Example:
        >>> search("ACE solar wind proton density", top_k=2)[0]
        {'id': 'cda/AC_H2_SWE/Np', 'name': 'Proton No. density',
         'description': 'Proton No. density. Solar Wind Proton Number Density, '
                        'scalar. ACE/SWEPAM ... Units: #/cc. ...', 'score': 1.0, ...}
    """
    if not query or not query.strip():
        return []
    return search_batch(
        [query],
        top_k,
        provider=provider,
        region=region,
        measurement_type=measurement_type,
        window=window,
    )[0]

search_batch

search_batch(queries: list[str], top_k: int = 5, *, provider: str | None = None, region: str | None = None, measurement_type: str | None = None, window: tuple[str, str] | None = None) -> list[list[dict]]

Resolve several queries in ONE pass — the 'composed RAG'.

Encodes all queries in a single embedding pass and issues a single multi-vector ChromaDB query (Chroma returns one result set per query natively), then fuses each independently. Returns one result list per input query, aligned by index (blank queries map to []).

Parameters:

Name	Type	Description	Default
queries	list[str]	One free-text query per parameter to resolve.	required
top_k	int	Results per query.	5
provider	str | None	Same filter as search().	None
region	str | None	Same filter as search().	None
measurement_type	str | None	Same filter as search().	None
window	tuple[str, str] | None	(start, stop) the caller intends to download; products that cannot cover it are ranked down before the cut, not merely flagged after it.	None

Example

ace, wind = search_batch(["ACE solar wind proton density", ... "Wind magnetic field GSE"], top_k=3) ace[0].get("id"), wind[0].get("id") ('cda/AC_H2_SWE/Np', 'cda/WI_H0_MFI/BGSEa')

Source code in helioai/tools/rag.py

def search_batch(
    queries: list[str],
    top_k: int = 5,
    *,
    provider: str | None = None,
    region: str | None = None,
    measurement_type: str | None = None,
    window: tuple[str, str] | None = None,
) -> list[list[dict]]:
    """Resolve several queries in ONE pass — the 'composed RAG'.

    Encodes all queries in a single embedding pass and issues a single
    multi-vector ChromaDB query (Chroma returns one result set per query
    natively), then fuses each independently. Returns one result list per input
    query, aligned by index (blank queries map to []).

    Args:
        queries: One free-text query per parameter to resolve.
        top_k: Results per query.
        provider: Same filter as `search()`.
        region: Same filter as `search()`.
        measurement_type: Same filter as `search()`.
        window: `(start, stop)` the caller intends to download; products that cannot
            cover it are ranked down before the cut, not merely flagged after it.

    Example:
        >>> ace, wind = search_batch(["ACE solar wind proton density",
        ...                           "Wind magnetic field GSE"], top_k=3)
        >>> ace[0].get("id"), wind[0].get("id")
        ('cda/AC_H2_SWE/Np', 'cda/WI_H0_MFI/BGSEa')
    """
    results: list[list[dict]] = [[] for _ in queries]
    active = [(i, q) for i, q in enumerate(queries) if q and q.strip()]
    if not active:
        return results

    # Check per-query cache; only encode/query Chroma for cache misses.
    def _cache_key(q: str) -> tuple:
        return (q, provider, region, measurement_type, top_k, window)

    uncached = []
    for i, q in active:
        hit = _search_cache.get(_cache_key(q))
        if hit is not None:
            results[i] = hit
        else:
            uncached.append((i, q))

    if not uncached:
        return results

    active = uncached
    model, collection = _load()
    hybrid = settings.rag.hybrid_enabled
    dense_k = settings.rag.hybrid_fetch_k if hybrid else top_k

    vecs = model.encode(
        [q for _, q in active],
        normalize_embeddings=True,
        convert_to_numpy=True,
    ).tolist()

    res = collection.query(
        query_embeddings=vecs,
        n_results=dense_k,
        where=_build_where(provider, region, measurement_type),
        include=["documents", "metadatas", "distances"],
    )
    all_ids = res.get("ids", []) or []
    all_docs = res.get("documents", []) or []
    all_metas = res.get("metadatas", []) or []
    all_dists = res.get("distances", []) or []

    # Second pass without the provider constraint, so an over-eager filter cannot
    # hide a parameter entirely. The embeddings are already computed, so this
    # costs one extra ANN search per batch and no re-encoding.
    open_hits: list[tuple] | None = None
    if provider:
        open_res = collection.query(
            query_embeddings=vecs,
            n_results=dense_k,
            where=_build_where(None, region, measurement_type),
            include=["documents", "metadatas", "distances"],
        )
        open_hits = (
            open_res.get("ids", []) or [],
            open_res.get("documents", []) or [],
            open_res.get("metadatas", []) or [],
            open_res.get("distances", []) or [],
        )

    for j, (i, q) in enumerate(active):
        dense_hit = (
            all_ids[j] if j < len(all_ids) else [],
            all_docs[j] if j < len(all_docs) else [],
            all_metas[j] if j < len(all_metas) else [],
            all_dists[j] if j < len(all_dists) else [],
        )
        res = _fuse_query(
            q,
            dense_hit,
            top_k,
            provider=provider,
            region=region,
            measurement_type=measurement_type,
            hybrid=hybrid,
            window=window,
        )
        if open_hits is not None:
            unfiltered = _fuse_query(
                q,
                tuple(col[j] if j < len(col) else [] for col in open_hits),
                top_k,
                provider=None,
                region=region,
                measurement_type=measurement_type,
                hybrid=hybrid,
                window=window,
            )
            res = _append_cross_provider(res, unfiltered, provider, _CROSS_PROVIDER_EXTRA)
        results[i] = res
        key = _cache_key(q)
        if len(_search_cache) >= _SEARCH_CACHE_MAX:
            _search_cache.pop(next(iter(_search_cache)))
        _search_cache[key] = res
    return results

search_catalogs

search_catalogs(query: str, top_k: int = 5, *, product_type: str | None = None) -> list[dict]

Semantic search over the AMDA catalog/timetable index.

Requires helioai index to have been run at least once.

Parameters:

Name	Type	Description	Default
query	str	Free text describing the events wanted.	required
top_k	int	How many products to return.	5
product_type	str | None	'catalog', 'timetable', or None for both.	None

Returns:

Type	Description
list[dict]	Dicts of {id, name, description, score, nb_events, product_type}, best
list[dict]	first. Empty when the catalog collection is absent — a missing index is
list[dict]	reported as no results rather than as a crash, because the agent can
list[dict]	still answer from parameters.

Source code in helioai/tools/rag.py

def search_catalogs(
    query: str,
    top_k: int = 5,
    *,
    product_type: str | None = None,
) -> list[dict]:
    """Semantic search over the AMDA catalog/timetable index.

    Requires `helioai index` to have been run at least once.

    Args:
        query: Free text describing the events wanted.
        top_k: How many products to return.
        product_type: `'catalog'`, `'timetable'`, or None for both.

    Returns:
        Dicts of `{id, name, description, score, nb_events, product_type}`, best
        first. Empty when the catalog collection is absent — a missing index is
        reported as no results rather than as a crash, because the agent can
        still answer from parameters.
    """
    if not query or not query.strip():
        return []
    try:
        global _catalog_collection
        model, _ = _load()  # reuse the already-loaded embedding model
        if _catalog_collection is None:
            with _lock:
                if _catalog_collection is None:
                    import chromadb

                    client = chromadb.PersistentClient(path=str(settings.rag.chroma_dir))
                    _catalog_collection = client.get_collection(
                        name=settings.rag.catalogs_collection_name
                    )
        col = _catalog_collection
    except Exception as e:
        log.debug("catalog collection unavailable (%s)", e)
        return []

    try:
        vec = model.encode([query], normalize_embeddings=True, convert_to_numpy=True).tolist()
        where: dict | None = {"product_type": product_type} if product_type else None
        res = col.query(
            query_embeddings=vec,
            n_results=min(top_k, col.count() or 1),
            where=where,
            include=["documents", "metadatas", "distances"],
        )
        results: list[dict] = []
        for pid, doc, meta, dist in zip(
            res["ids"][0],
            res["documents"][0],
            res["metadatas"][0],
            res["distances"][0],
            strict=False,
        ):
            score = round(max(0.0, min(1.0, (1.0 - float(dist) + 1.0) / 2.0)), 4)
            results.append(
                {
                    "id": pid,
                    "name": (meta or {}).get("name", pid),
                    "description": _truncate(doc or ""),
                    "score": score,
                    "nb_events": (meta or {}).get("nb_events", 0),
                    "product_type": (meta or {}).get("product_type", ""),
                }
            )
        return results
    except Exception as e:
        log.warning("catalog search failed: %s", e)
        return []

Data access

helioai.tools.speasy_tools

speasy tools: data access layer wrapping the speasy library.

speasy provides unified access to 70+ missions and 65k+ products from CDAWeb, AMDA, CSA, SSC and others. These tools are the helioai equivalent of AMDA's download_timeseries and list_parameters.

get_timeseries async

get_timeseries(param_id: str, start: str, stop: str, max_points: int = 5000, _data_dir: str | None = None) -> dict

Download a time series from any speasy provider.

Parameters:

Name	Type	Description	Default
param_id	str	speasy parameter id (e.g. 'amda/imf', 'cdaweb/AC_H0_MFI/BGSEc')	required
start	str	ISO 8601 start time (e.g. '2024-01-01T00:00:00')	required
stop	str	ISO 8601 stop time	required
max_points	int	max samples to return (downsampled if needed)	5000
_data_dir	str | None	injected by the runtime (tool_exec.trusted_args) — the session's data directory the download is persisted in. Not exposed in the LLM tool schema.	None

The param_id should be in speasy format: "{provider}/{xmlid}" e.g. "amda/ace_epam_ca60_he", "cda/ACE_H0_MFI/BGSEc" (returned by search_parameters).

Returns dict with: param_id, start, stop, units, shape, n_points, preview (first 10 rows as CSV)

Example

await get_timeseries("amda/imf", "2010-01-01T00:00:00", "2010-01-01T06:00:00") {'dataset': 'imf', 'param_id': 'amda/imf', 'units': 'nT', 'components': ['bx', 'by', 'bz'], 'cadence': '16 s', 'shape': [1350, 3], 'n_points': 1350, 'n_valid': 1350, 'quality': {'missing_pct': 0.0, ...}, 'preview': '2010-01-01T00:00:09.000000000 -1.839, 2.308, 0.108\n...', ...}

Source code in helioai/tools/speasy_tools.py

async def get_timeseries(
    param_id: str,
    start: str,
    stop: str,
    max_points: int = 5000,
    _data_dir: str | None = None,
) -> dict:
    """Download a time series from any speasy provider.

    Args:
        param_id: speasy parameter id (e.g. 'amda/imf', 'cdaweb/AC_H0_MFI/BGSEc')
        start: ISO 8601 start time (e.g. '2024-01-01T00:00:00')
        stop:  ISO 8601 stop time
        max_points: max samples to return (downsampled if needed)
        _data_dir: injected by the runtime (`tool_exec.trusted_args`) — the session's data
            directory the download is persisted in. Not exposed in the LLM tool schema.

    The param_id should be in speasy format: "{provider}/{xmlid}"
    e.g. "amda/ace_epam_ca60_he", "cda/ACE_H0_MFI/BGSEc"
    (returned by search_parameters).

    Returns dict with: param_id, start, stop, units, shape, n_points, preview (first 10 rows as CSV)

    Example:
        >>> await get_timeseries("amda/imf", "2010-01-01T00:00:00", "2010-01-01T06:00:00")
        {'dataset': 'imf', 'param_id': 'amda/imf', 'units': 'nT',
         'components': ['bx', 'by', 'bz'], 'cadence': '16 s', 'shape': [1350, 3],
         'n_points': 1350, 'n_valid': 1350, 'quality': {'missing_pct': 0.0, ...},
         'preview': '2010-01-01T00:00:09.000000000  -1.839, 2.308, 0.108\\n...', ...}
    """
    return await run_blocking(
        _get_timeseries_sync,
        param_id=param_id,
        start=start,
        stop=stop,
        max_points=max_points,
        _data_dir=_data_dir,
    )

list_missions async

list_missions() -> dict

List available speasy data providers and their top-level missions.

Returns a summary dict with provider names and approximate product counts.

Example

await list_missions() {'providers': ['amda', 'archive', 'cda', 'csa', 'ssc', 'uiowaephtool'], 'note': 'Use search_parameters to find specific parameters. ...'}

Source code in helioai/tools/speasy_tools.py

async def list_missions() -> dict:
    """List available speasy data providers and their top-level missions.

    Returns a summary dict with provider names and approximate product counts.

    Example:
        >>> await list_missions()
        {'providers': ['amda', 'archive', 'cda', 'csa', 'ssc', 'uiowaephtool'],
         'note': 'Use search_parameters to find specific parameters. ...'}
    """
    return await run_blocking(_list_missions_sync)

search_parameters async

search_parameters(query: str | None = None, top_k: int = 5, provider: str | None = None, queries: list[str] | None = None, start: str | None = None, stop: str | None = None) -> dict

Semantic search over the speasy catalog (83k+ products).

Requires the index to be built first (run: helioai index). Falls back to a direct speasy text match if no index is found.

Every result carries the product's published coverage. Passing the window you intend to download sorts products that cannot cover it to the bottom and flags them, which is cheaper than discovering it one failed download at a time — resolving a 2015 ephemeris used to cost four turns against products that stop in 1997.

Parameters:

Name	Type	Description	Default
query	str | None	free-text English query for a SINGLE parameter.	None
queries	list[str] | None	list of queries to resolve SEVERAL parameters in ONE call (preferred when 2+ parameters are needed — much cheaper).	None
top_k	int	number of results per query.	5
provider	str | None	optional — restrict to one provider (amda/cda/csa/ssc).	None
start	str | None	optional ISO start of the interval you intend to download.	None
stop	str | None	optional ISO stop. Both are needed for the filter to apply.	None

Returns either {query, provider, results} (single) or {provider, groups: [{query, results}]} (batch).

Example

await search_parameters(query="ACE solar wind proton density", top_k=3) {'query': 'ACE solar wind proton density', 'provider': None, 'results': [ {'id': 'cda/AC_H2_SWE/Np', 'description': 'Proton No. density. Solar Wind Proton Number Density, scalar. ' 'ACE/SWEPAM ... 1-Hour Level 2 Data ... Units: #/cc. ...', 'coverage': '1998-02-04 → 2024-07-09'}, ...]}

Source code in helioai/tools/speasy_tools.py

async def search_parameters(
    query: str | None = None,
    top_k: int = 5,
    provider: str | None = None,
    queries: list[str] | None = None,
    start: str | None = None,
    stop: str | None = None,
) -> dict:
    """Semantic search over the speasy catalog (83k+ products).

    Requires the index to be built first (run: helioai index).
    Falls back to a direct speasy text match if no index is found.

    Every result carries the product's published `coverage`. Passing the window you
    intend to download sorts products that cannot cover it to the bottom and flags them,
    which is cheaper than discovering it one failed download at a time — resolving a
    2015 ephemeris used to cost four turns against products that stop in 1997.

    Args:
        query: free-text English query for a SINGLE parameter.
        queries: list of queries to resolve SEVERAL parameters in ONE call
                 (preferred when 2+ parameters are needed — much cheaper).
        top_k: number of results per query.
        provider: optional — restrict to one provider (amda/cda/csa/ssc).
        start: optional ISO start of the interval you intend to download.
        stop: optional ISO stop. Both are needed for the filter to apply.

    Returns either {query, provider, results} (single) or
    {provider, groups: [{query, results}]} (batch).

    Example:
        >>> await search_parameters(query="ACE solar wind proton density", top_k=3)
        {'query': 'ACE solar wind proton density', 'provider': None, 'results': [
         {'id': 'cda/AC_H2_SWE/Np',
          'description': 'Proton No. density. Solar Wind Proton Number Density, scalar. '
                         'ACE/SWEPAM ... 1-Hour Level 2 Data ... Units: #/cc. ...',
          'coverage': '1998-02-04 → 2024-07-09'}, ...]}
    """
    return await run_blocking(
        _search_parameters_sync,
        query=query,
        top_k=top_k,
        provider=provider,
        queries=queries,
        start=start,
        stop=stop,
    )

Event catalogs

helioai.tools.catalog_tools

Catalog and timetable tools for HelioAI.

Exposes the 29 CatalogIndex + 188 TimetableIndex from the AMDA speasy inventory as first-class agent tools. The key capability is get_events_timeseries: download a parameter for every event in a catalog in one speasy call, opening the door to superposed epoch analysis.

list_catalogs async

list_catalogs(type: str = 'all', region: str | None = None, query: str | None = None) -> dict

List available AMDA event catalogs and timetables.

Parameters:

Name	Type	Description	Default
type	str	'catalog', 'timetable', or 'all' (default).	'all'
region	str | None	optional keyword filter on name/description (e.g. 'ICME', 'bow shock', 'MMS').	None
query	str | None	optional free-text description of the events wanted; the list is then ordered by semantic relevance to it instead of by size.	None

Returns a list of entries with id, name, type, nb_events, survey range and description. Use the id field with get_catalog() and get_events_timeseries().

Example

await list_catalogs(type="catalog", region="ICME") {'total': 3, 'type_filter': 'catalog', 'region_filter': 'ICME', 'catalogs': [ {'id': 'amda/sharedcatalog_41', 'name': 'ICME_multi-catalog', 'type': 'catalog', 'nb_events': 2003, 'survey_start': '1975-01-08', 'survey_stop': '2022-10-21', 'description': '...'}, ...]}

Source code in helioai/tools/catalog_tools.py

async def list_catalogs(
    type: str = "all",
    region: str | None = None,
    query: str | None = None,
) -> dict:
    """List available AMDA event catalogs and timetables.

    Args:
        type: 'catalog', 'timetable', or 'all' (default).
        region: optional keyword filter on name/description (e.g. 'ICME', 'bow shock', 'MMS').
        query: optional free-text description of the events wanted; the list is then
            ordered by semantic relevance to it instead of by size.

    Returns a list of entries with id, name, type, nb_events, survey range and description.
    Use the `id` field with get_catalog() and get_events_timeseries().

    Example:
        >>> await list_catalogs(type="catalog", region="ICME")
        {'total': 3, 'type_filter': 'catalog', 'region_filter': 'ICME', 'catalogs': [
         {'id': 'amda/sharedcatalog_41', 'name': 'ICME_multi-catalog', 'type': 'catalog',
          'nb_events': 2003, 'survey_start': '1975-01-08', 'survey_stop': '2022-10-21',
          'description': '...'}, ...]}
    """
    return await run_blocking(_list_catalogs_sync, type=type, region=region, query=query)

get_catalog async

get_catalog(catalog_id: str, start: str | None = None, stop: str | None = None, max_events: int = 10, columns: list[str] | None = None, where: dict | None = None, sort_by: str | None = None, descending: bool = False, offset: int = 0) -> dict

Download and summarize an AMDA event catalog or timetable.

Parameters:

Name	Type	Description	Default
catalog_id	str	speasy uid from list_catalogs (e.g. 'amda/sharedcatalog_41').	required
start	str | None	optional ISO 8601 start — filter events beginning after this time.	None
stop	str | None	optional ISO 8601 stop — filter events beginning before this time.	None
max_events	int	maximum events to include in the sample (default 10).	10
columns	list[str] | None	restrict the metadata columns returned per event.	None
where	dict | None	server-side row filter — {"column": str, "op": "eq|ne|gt|gte|lt|lte|contains", "value": any}.	None
sort_by	str | None	column name to sort events by before slicing.	None
descending	bool	sort direction (default ascending).	False
offset	int	pagination offset into the filtered+sorted events.	0

Returns catalog metadata + a sample of events (start, stop, key columns). Use get_events_timeseries() to download a parameter over all events.

Example

await get_catalog("amda/sharedcatalog_41", start="2015-01-01", stop="2016-01-01", ... max_events=5, sort_by="start") {'_kind': 'catalog_preview', 'catalog_id': 'amda/sharedcatalog_41', 'name': 'ICME_multi-catalog', 'nb_events_total': 2003, 'nb_events_filtered': ..., 'returned': 5, 'columns': [...], 'sample': [{'start': ..., 'stop': ..., ...}, ...], 'survey_start': '1975-01-08', 'survey_stop': '2022-10-21'}

Source code in helioai/tools/catalog_tools.py

async def get_catalog(
    catalog_id: str,
    start: str | None = None,
    stop: str | None = None,
    max_events: int = 10,
    columns: list[str] | None = None,
    where: dict | None = None,
    sort_by: str | None = None,
    descending: bool = False,
    offset: int = 0,
) -> dict:
    """Download and summarize an AMDA event catalog or timetable.

    Args:
        catalog_id: speasy uid from list_catalogs (e.g. 'amda/sharedcatalog_41').
        start:      optional ISO 8601 start — filter events beginning after this time.
        stop:       optional ISO 8601 stop  — filter events beginning before this time.
        max_events: maximum events to include in the sample (default 10).
        columns:    restrict the metadata columns returned per event.
        where:      server-side row filter — {"column": str, "op": "eq|ne|gt|gte|lt|lte|contains", "value": any}.
        sort_by:    column name to sort events by before slicing.
        descending: sort direction (default ascending).
        offset:     pagination offset into the filtered+sorted events.

    Returns catalog metadata + a sample of events (start, stop, key columns).
    Use get_events_timeseries() to download a parameter over all events.

    Example:
        >>> await get_catalog("amda/sharedcatalog_41", start="2015-01-01", stop="2016-01-01",
        ...                   max_events=5, sort_by="start")
        {'_kind': 'catalog_preview', 'catalog_id': 'amda/sharedcatalog_41',
         'name': 'ICME_multi-catalog', 'nb_events_total': 2003, 'nb_events_filtered': ...,
         'returned': 5, 'columns': [...], 'sample': [{'start': ..., 'stop': ..., ...}, ...],
         'survey_start': '1975-01-08', 'survey_stop': '2022-10-21'}
    """
    return await run_blocking(
        _get_catalog_sync,
        catalog_id=catalog_id,
        start=start,
        stop=stop,
        max_events=max_events,
        columns=columns,
        where=where,
        sort_by=sort_by,
        descending=descending,
        offset=offset,
    )

get_events_timeseries async

get_events_timeseries(catalog_id: str, param_id: str, start: str, stop: str, max_events: int = 50, _data_dir: str | None = None) -> dict

Download a parameter for every event in a catalog window (superposed epoch).

This is the core catalog tool: it fetches N time series in a SINGLE speasy call using the native multi-interval API. Use it for: - Superposed epoch analysis (stack-plot across events) - Statistical summaries per event (min/max/mean) - Comparing a parameter across e.g. all ICME crossings in a year

Parameters:

Name	Type	Description	Default
catalog_id	str	speasy uid from list_catalogs (e.g. 'amda/sharedcatalog_41').	required
param_id	str	speasy parameter id (e.g. 'amda/imf_gsm') — resolve via search_parameters first.	required
start	str	ISO 8601 start — restrict to events beginning after this time.	required
stop	str	ISO 8601 stop — restrict to events beginning before this time.	required
max_events	int	cap on events to download (default 20 — each is one speasy call slot).	50
_data_dir	str | None	injected by the runtime (tool_exec.trusted_args) — the session's data directory the collection is persisted in. Not exposed in the LLM tool schema.	None

Returns per-event statistics and saves the raw data to the workspace for run_python.

Example

await get_events_timeseries("amda/sharedcatalog_41", "amda/imf", ... "2015-01-01", "2016-01-01", max_events=10) {'catalog_id': 'amda/sharedcatalog_41', 'param_id': 'amda/imf', 'stats': [ {'event': 0, 'start': '2015-01-03T...', 'stop': '2015-01-04T...', 'n_points': ..., ...}, ...], ...}

Source code in helioai/tools/catalog_tools.py

async def get_events_timeseries(
    catalog_id: str,
    param_id: str,
    start: str,
    stop: str,
    max_events: int = 50,
    _data_dir: str | None = None,
) -> dict:
    """Download a parameter for every event in a catalog window (superposed epoch).

    This is the core catalog tool: it fetches N time series in a SINGLE speasy call
    using the native multi-interval API.  Use it for:
    - Superposed epoch analysis (stack-plot across events)
    - Statistical summaries per event (min/max/mean)
    - Comparing a parameter across e.g. all ICME crossings in a year

    Args:
        catalog_id: speasy uid from list_catalogs (e.g. 'amda/sharedcatalog_41').
        param_id:   speasy parameter id (e.g. 'amda/imf_gsm') — resolve via search_parameters first.
        start:      ISO 8601 start — restrict to events beginning after this time.
        stop:       ISO 8601 stop  — restrict to events beginning before this time.
        max_events: cap on events to download (default 20 — each is one speasy call slot).
        _data_dir: injected by the runtime (`tool_exec.trusted_args`) — the session's data
            directory the collection is persisted in. Not exposed in the LLM tool schema.

    Returns per-event statistics and saves the raw data to the workspace for run_python.

    Example:
        >>> await get_events_timeseries("amda/sharedcatalog_41", "amda/imf",
        ...                             "2015-01-01", "2016-01-01", max_events=10)
        {'catalog_id': 'amda/sharedcatalog_41', 'param_id': 'amda/imf', 'stats': [
         {'event': 0, 'start': '2015-01-03T...', 'stop': '2015-01-04T...',
          'n_points': ..., ...}, ...], ...}
    """
    return await run_blocking(
        _get_events_timeseries_sync,
        catalog_id=catalog_id,
        param_id=param_id,
        start=start,
        stop=stop,
        max_events=max_events,
        _data_dir=_data_dir,
    )

save_catalog async

save_catalog(name: str, events: list[dict], description: str = '', _catalogs_dir: str | None = None) -> dict

Save a list of events as a local catalog under the local/ prefix.

Parameters:

Name	Type	Description	Default
name	str	Catalog name — lowercase letters, digits, hyphens, underscores (1-40 chars).	required
events	list[dict]	List of dicts with 'start' and 'stop' ISO 8601 strings plus optional extra keys.	required
description	str	Short description (optional).	''
_catalogs_dir	str | None	injected by the runtime (tool_exec.trusted_args) — the user's catalogue directory. Not exposed in the LLM tool schema.	None

Returns {"catalog_id": "local/", "nb_events": N, "note": "..."}. Overwrites an existing catalog with the same name. Use list_catalogs() then get_catalog("local/") to inspect it.

Example

await save_catalog("my-shocks", ... [{"start": "2015-03-17T04:01:00", "stop": "2015-03-17T05:00:00", ... "note": "St. Patrick's Day storm shock"}])

Source code in helioai/tools/catalog_tools.py

async def save_catalog(
    name: str,
    events: list[dict],
    description: str = "",
    _catalogs_dir: str | None = None,
) -> dict:
    """Save a list of events as a local catalog under the local/<name> prefix.

    Args:
        name:        Catalog name — lowercase letters, digits, hyphens, underscores (1-40 chars).
        events:      List of dicts with 'start' and 'stop' ISO 8601 strings plus optional extra keys.
        description: Short description (optional).
        _catalogs_dir: injected by the runtime (`tool_exec.trusted_args`) — the user's
            catalogue directory. Not exposed in the LLM tool schema.

    Returns {"catalog_id": "local/<name>", "nb_events": N, "note": "..."}.
    Overwrites an existing catalog with the same name.
    Use list_catalogs() then get_catalog("local/<name>") to inspect it.

    Example:
        >>> await save_catalog("my-shocks",
        ...                    [{"start": "2015-03-17T04:01:00", "stop": "2015-03-17T05:00:00",
        ...                      "note": "St. Patrick's Day storm shock"}])
        {'catalog_id': 'local/my-shocks', 'nb_events': 1, 'overwritten': False, 'note': '...'}
    """
    return await run_blocking(
        _save_catalog_sync,
        name=name,
        events=events,
        description=description,
        _catalogs_dir=_catalogs_dir,
    )

Plasma physics

helioai.tools.plasmapy_tools

PlasmaPy-based plasma physics calculations exposed as agent tools.

Each function accepts plain SI-ish numbers (nT, cm⁻³, eV) and returns a dict with value, unit, and a brief physical context — ready for LLM consumption.

SPASE ParticleQuantity / FieldQuantity mapping: plasma_beta → PlasmaBeta (ActivityIndex) gyrofrequency → Gyrofrequency (FieldQuantity + ParticleQuantity) debye_length → (ParticleQuantity implied) alfven_speed → AlfvenVelocity (ParticleQuantity) inertial_length → (ParticleQuantity implied) power_spectrum → Spectrum (MeasurementType)

plasma_beta async

plasma_beta(B_nT: float, n_cm3: float, T_eV: float) -> dict

Compute plasma beta — ratio of thermal pressure to magnetic pressure.

Parameters:

Name	Type	Description	Default
B_nT	float	Magnetic field magnitude in nT	required
n_cm3	float	Number density in cm⁻³	required
T_eV	float	Temperature in eV	required

Returns dict with beta (dimensionless) and regime interpretation.

Example

await plasma_beta(B_nT=5.0, n_cm3=10.0, T_eV=20.0) {'beta': 3.221367, 'unit': 'dimensionless', 'regime': 'high-β plasma (β ~ 1-10) — typical magnetosheath / plasma sheet', ...}

Source code in helioai/tools/plasmapy_tools.py

async def plasma_beta(B_nT: float, n_cm3: float, T_eV: float) -> dict:
    """Compute plasma beta — ratio of thermal pressure to magnetic pressure.

    Args:
        B_nT:  Magnetic field magnitude in nT
        n_cm3: Number density in cm⁻³
        T_eV:  Temperature in eV

    Returns dict with beta (dimensionless) and regime interpretation.

    Example:
        >>> await plasma_beta(B_nT=5.0, n_cm3=10.0, T_eV=20.0)
        {'beta': 3.221367, 'unit': 'dimensionless',
         'regime': 'high-β plasma (β ~ 1-10) — typical magnetosheath / plasma sheet', ...}
    """
    try:
        import astropy.units as u
        import plasmapy.formulary as pf

        B = B_nT * u.nT
        n = n_cm3 * u.cm**-3
        T = T_eV * u.eV

        beta_val = float(pf.beta(T, n, B).value)

        if beta_val < 0.01:
            regime = "magnetically dominated (β ≪ 1) — typical inner magnetosphere / coronal loop"
        elif beta_val < 1.0:
            regime = "low-β plasma (β < 1) — typical solar wind / outer magnetosphere"
        elif beta_val < 10.0:
            regime = "high-β plasma (β ~ 1-10) — typical magnetosheath / plasma sheet"
        else:
            regime = "pressure-dominated (β ≫ 1) — typical ionosphere / dense plasma"

        return {
            "beta": round(beta_val, 6),
            "unit": "dimensionless",
            "regime": regime,
            "inputs": {"B_nT": B_nT, "n_cm3": n_cm3, "T_eV": T_eV},
        }
    except Exception as e:
        return {"error": str(e)}

gyrofrequency async

gyrofrequency(B_nT: float, particle: str = 'proton') -> dict

Compute particle gyrofrequency (cyclotron frequency).

Parameters:

Name	Type	Description	Default
B_nT	float	Magnetic field magnitude in nT	required
particle	str	'proton', 'electron', 'alpha' (default: proton)	'proton'

Returns dict with frequency in Hz and angular frequency in rad/s.

Example

await gyrofrequency(B_nT=5.0) {'frequency_Hz': 0.0762, 'angular_frequency_rad_s': 0.4789, 'particle': 'proton', 'B_nT': 5.0, 'period_s': 13.118895} await gyrofrequency(B_nT=5.0, particle="electron")

Source code in helioai/tools/plasmapy_tools.py

async def gyrofrequency(B_nT: float, particle: str = "proton") -> dict:
    """Compute particle gyrofrequency (cyclotron frequency).

    Args:
        B_nT:    Magnetic field magnitude in nT
        particle: 'proton', 'electron', 'alpha' (default: proton)

    Returns dict with frequency in Hz and angular frequency in rad/s.

    Example:
        >>> await gyrofrequency(B_nT=5.0)
        {'frequency_Hz': 0.0762, 'angular_frequency_rad_s': 0.4789, 'particle': 'proton',
         'B_nT': 5.0, 'period_s': 13.118895}
        >>> await gyrofrequency(B_nT=5.0, particle="electron")
        {'frequency_Hz': 139.9624, ...}
    """
    try:
        import astropy.units as u
        import plasmapy.formulary as pf

        B = B_nT * u.nT
        p = _parse_particle(particle)

        omega = pf.gyrofrequency(B, particle=p, signed=False)
        f_hz = float((omega / (2 * math.pi * u.rad)).to(u.Hz).value)
        omega_rad_s = float(omega.to(u.rad / u.s).value)

        return {
            "frequency_Hz": round(f_hz, 4),
            "angular_frequency_rad_s": round(omega_rad_s, 4),
            "particle": particle,
            "B_nT": B_nT,
            "period_s": round(1.0 / f_hz, 6) if f_hz > 0 else None,
        }
    except Exception as e:
        return {"error": str(e)}

debye_length async

debye_length(n_cm3: float, T_eV: float) -> dict

Compute electron Debye length.

Parameters:

Name	Type	Description	Default
n_cm3	float	Electron number density in cm⁻³	required
T_eV	float	Electron temperature in eV	required

Returns dict with Debye length in km and meters.

Example

await debye_length(n_cm3=10.0, T_eV=12.0) {'debye_length_m': 8.143475, 'debye_length_km': 0.008143475, 'inputs': {'n_cm3': 10.0, 'T_eV': 12.0}}

Source code in helioai/tools/plasmapy_tools.py

async def debye_length(n_cm3: float, T_eV: float) -> dict:
    """Compute electron Debye length.

    Args:
        n_cm3: Electron number density in cm⁻³
        T_eV:  Electron temperature in eV

    Returns dict with Debye length in km and meters.

    Example:
        >>> await debye_length(n_cm3=10.0, T_eV=12.0)
        {'debye_length_m': 8.143475, 'debye_length_km': 0.008143475,
         'inputs': {'n_cm3': 10.0, 'T_eV': 12.0}}
    """
    try:
        import astropy.units as u
        import plasmapy.formulary as pf

        n = n_cm3 * u.cm**-3
        T = T_eV * u.eV

        lambda_D = pf.Debye_length(T, n)
        lambda_m = float(lambda_D.to(u.m).value)
        lambda_km = float(lambda_D.to(u.km).value)

        return {
            "debye_length_m": round(lambda_m, 6),
            "debye_length_km": round(lambda_km, 9),
            "inputs": {"n_cm3": n_cm3, "T_eV": T_eV},
        }
    except Exception as e:
        return {"error": str(e)}

alfven_speed async

alfven_speed(B_nT: float, n_cm3: float, mass_amu: float = 1.0) -> dict

Compute Alfvén speed.

Parameters:

Name	Type	Description	Default
B_nT	float	Magnetic field magnitude in nT	required
n_cm3	float	Ion number density in cm⁻³	required
mass_amu	float	Ion mass in atomic mass units (default 1.0 = proton)	1.0

Returns dict with Alfvén speed in km/s.

Example

await alfven_speed(B_nT=5.0, n_cm3=5.0) {'alfven_speed_km_s': 48.937, 'alfven_speed_m_s': 48936.9, 'inputs': {'B_nT': 5.0, 'n_cm3': 5.0, 'mass_amu': 1.0}, ...}

Source code in helioai/tools/plasmapy_tools.py

async def alfven_speed(B_nT: float, n_cm3: float, mass_amu: float = 1.0) -> dict:
    """Compute Alfvén speed.

    Args:
        B_nT:      Magnetic field magnitude in nT
        n_cm3:     Ion number density in cm⁻³
        mass_amu:  Ion mass in atomic mass units (default 1.0 = proton)

    Returns dict with Alfvén speed in km/s.

    Example:
        >>> await alfven_speed(B_nT=5.0, n_cm3=5.0)
        {'alfven_speed_km_s': 48.937, 'alfven_speed_m_s': 48936.9,
         'inputs': {'B_nT': 5.0, 'n_cm3': 5.0, 'mass_amu': 1.0}, ...}
    """
    try:
        import astropy.constants as const
        import astropy.units as u
        import plasmapy.formulary as pf

        B = B_nT * u.nT
        n = n_cm3 * u.cm**-3
        from plasmapy.particles import CustomParticle

        ion = CustomParticle(mass=mass_amu * const.u, charge=1 * const.e.si)

        V_A = pf.Alfven_speed(B, n, ion=ion)
        va_km_s = float(V_A.to(u.km / u.s).value)

        return {
            "alfven_speed_km_s": round(va_km_s, 3),
            "alfven_speed_m_s": round(va_km_s * 1000, 1),
            "inputs": {"B_nT": B_nT, "n_cm3": n_cm3, "mass_amu": mass_amu},
            "note": "Typical solar wind: 40-80 km/s. Magnetosphere: 100-1000 km/s.",
        }
    except Exception as e:
        return {"error": str(e)}

inertial_length async

inertial_length(n_cm3: float, particle: str = 'proton') -> dict

Compute ion or electron inertial length (skin depth).

Parameters:

Name	Type	Description	Default
n_cm3	float	Number density in cm⁻³	required
particle	str	'proton' or 'electron' (default: proton)	'proton'

Returns dict with inertial length in km and meters.

Example

await inertial_length(n_cm3=5.0) {'inertial_length_km': 101.8354, 'inertial_length_m': 101835.35, 'particle': 'proton', 'inputs': {'n_cm3': 5.0}}

Source code in helioai/tools/plasmapy_tools.py

async def inertial_length(n_cm3: float, particle: str = "proton") -> dict:
    """Compute ion or electron inertial length (skin depth).

    Args:
        n_cm3:    Number density in cm⁻³
        particle: 'proton' or 'electron' (default: proton)

    Returns dict with inertial length in km and meters.

    Example:
        >>> await inertial_length(n_cm3=5.0)
        {'inertial_length_km': 101.8354, 'inertial_length_m': 101835.35,
         'particle': 'proton', 'inputs': {'n_cm3': 5.0}}
    """
    try:
        import astropy.units as u
        import plasmapy.formulary as pf

        n = n_cm3 * u.cm**-3
        p = _parse_particle(particle)

        d = pf.inertial_length(n, particle=p)
        d_km = float(d.to(u.km).value)
        d_m = float(d.to(u.m).value)

        return {
            "inertial_length_km": round(d_km, 4),
            "inertial_length_m": round(d_m, 2),
            "particle": particle,
            "inputs": {"n_cm3": n_cm3},
        }
    except Exception as e:
        return {"error": str(e)}

power_spectrum async

power_spectrum(values: list[float], dt_s: float, nperseg: int | None = None) -> dict

Compute power spectral density using Welch's method.

Parameters:

Name	Type	Description	Default
values	list[float]	Time series as a list of floats	required
dt_s	float	Sampling interval in seconds	required
nperseg	int | None	Samples per FFT segment (default: min(256, len(values)//4))	None

Returns dict with frequencies (Hz), PSD values, peak frequency, and export-ready summary for LLM interpretation.

Example

import math wave = [math.sin(2 * math.pi * 0.1 * i) for i in range(512)] psd = await power_spectrum(wave, dt_s=1.0) psd["peak_frequency_Hz"], psd["peak_period_s"] (0.101562, 9.846)

Source code in helioai/tools/plasmapy_tools.py

async def power_spectrum(
    values: list[float],
    dt_s: float,
    nperseg: int | None = None,
) -> dict:
    """Compute power spectral density using Welch's method.

    Args:
        values:  Time series as a list of floats
        dt_s:    Sampling interval in seconds
        nperseg: Samples per FFT segment (default: min(256, len(values)//4))

    Returns dict with frequencies (Hz), PSD values, peak frequency,
    and export-ready summary for LLM interpretation.

    Example:
        >>> import math
        >>> wave = [math.sin(2 * math.pi * 0.1 * i) for i in range(512)]
        >>> psd = await power_spectrum(wave, dt_s=1.0)
        >>> psd["peak_frequency_Hz"], psd["peak_period_s"]
        (0.101562, 9.846)
    """
    try:
        import numpy as np
        from scipy import signal

        arr = np.asarray(values, dtype=float)
        arr[~np.isfinite(arr)] = np.nan
        arr[np.abs(arr) >= 1e30] = np.nan
        n_original = len(arr)
        if n_original == 0:
            return {"error": "Input array is empty"}

        valid_mask = np.isfinite(arr)
        n_valid = int(np.sum(valid_mask))
        n_dropped = n_original - n_valid
        if n_valid < 8:
            return {"error": f"Need at least 8 finite samples, got {n_valid}"}

        # Identify contiguous finite runs (telemetry segments)
        finite_indices = np.flatnonzero(valid_mask)
        splits = np.where(np.diff(finite_indices) > 1)[0] + 1
        runs = np.split(finite_indices, splits)

        # Longest gap calculation
        gap_indices = np.flatnonzero(~valid_mask)
        if len(gap_indices) > 0:
            gap_splits = np.where(np.diff(gap_indices) > 1)[0] + 1
            gap_runs = np.split(gap_indices, gap_splits)
            longest_gap_samples = int(max(len(g) for g in gap_runs))
        else:
            longest_gap_samples = 0

        max_run_len = max(len(r) for r in runs)
        if max_run_len < 8:
            return {
                "error": f"Longest contiguous valid segment has {max_run_len} samples, need at least 8"
            }

        fs = 1.0 / dt_s
        seg = nperseg or min(256, n_valid // 4)
        seg = max(seg, 8)
        if seg > max_run_len:
            seg = max(max_run_len, 8)

        # Compute Welch PSD per continuous run and average weighted by window count.
        # Splicing separated runs together creates artificial high-frequency phase
        # jumps at telemetry gaps, destroying the spectral slope.
        usable_runs = 0
        total_windows = 0
        accumulated_psd = None
        freqs = None
        n_samples_used = 0

        step = seg - seg // 2  # default welch 50% overlap step
        for r_idx in runs:
            run_len = len(r_idx)
            if run_len < seg:
                continue
            run_arr = arr[r_idx]
            f, p = signal.welch(run_arr, fs=fs, nperseg=seg)
            n_win = 1 + (run_len - seg) // step
            if accumulated_psd is None:
                accumulated_psd = p * n_win
                freqs = f
            else:
                accumulated_psd += p * n_win
            total_windows += n_win
            n_samples_used += run_len
            usable_runs += 1

        if total_windows == 0 or accumulated_psd is None:
            return {"error": f"No continuous run reached segment length {seg}"}

        psd = accumulated_psd / total_windows

        peak_idx = int(np.argmax(psd[1:])) + 1
        peak_freq = float(freqs[peak_idx])
        peak_power = float(psd[peak_idx])

        return {
            "frequencies_Hz": [round(f, 6) for f in freqs.tolist()],
            "psd": [round(p, 8) for p in psd.tolist()],
            "peak_frequency_Hz": round(peak_freq, 6),
            "peak_period_s": round(1.0 / peak_freq, 3) if peak_freq > 0 else None,
            "peak_power": round(peak_power, 8),
            "n_points": n_valid,
            "n_original": n_original,
            "n_dropped": n_dropped,
            "gap_fraction": round(n_dropped / max(n_original, 1), 4),
            "n_segments": usable_runs,
            "longest_gap_samples": longest_gap_samples,
            "n_samples_used": n_samples_used,
            "fs_Hz": round(fs, 6),
            "freq_resolution_Hz": round(freqs[1] - freqs[0], 8) if len(freqs) > 1 else None,
        }
    except Exception as e:
        return {"error": str(e)}

Sandbox

helioai.tools.sandbox

Python sandbox: execute user/LLM-generated code in an isolated subprocess.

Security model

• Runs in a fresh subprocess (separate memory, no shared globals)
• Hard timeout (default 30s) — kills the process if exceeded
• stdout/stderr captured and returned
• Host credentials (~/.ssh, ~/.gnupg, ~/.config/gh, etc.) and .env masked under bubblewrap
• No network isolation (speasy needs network access) — trust LLM-generated code

Pre-imports available in sandbox: speasy, plasmapy, numpy, scipy, matplotlib, astropy Figures are saved to a temp directory; paths are returned (not base64). Use export(name, array) to share numerical data with the LLM.

run_python async

run_python(code: str, timeout: float = 60.0, _plot_dir: str | None = None, _run_idx: int | None = None, _no_net: bool = False) -> dict

Execute Python code in an isolated subprocess.

Parameters:

Name	Type	Description	Default
code	str	Python source code to execute. Has access to speasy (spz), plasmapy (pf), numpy (np), scipy, matplotlib (Agg — plt.show() saves to disk), astropy units (u). Call export(name, array) to share numerical results with the LLM.	required
timeout	float	maximum execution time in seconds — clamped to _MAX_TIMEOUT_S	60.0
_plot_dir	str | None	injected by the agent loop — workspace dir for this run. Not exposed in the LLM tool schema.	None

Returns dict with

• stdout: captured text output
• stderr: captured errors/warnings
• figure_paths: list of absolute paths to saved PNG files
• exports: dict of named numerical summaries (from export() calls)
• error: error message if execution failed

Example

await run_python( ... "import numpy as np\n" ... "export('rms', np.sqrt(np.mean(np.arange(8) ** 2)))\n" ... "print('done')" ... ) {'stdout': 'done', 'exports': {'rms': {'mean': 4.1833..., 'min': 4.1833..., 'max': 4.1833..., 'std': 0.0, 'n_finite': 1, 'n_nan': 0, ...}}, 'figure_paths': [], 'error': None, ...}

Source code in helioai/tools/sandbox.py

async def run_python(
    code: str,
    timeout: float = 60.0,
    _plot_dir: str | None = None,
    _run_idx: int | None = None,
    _no_net: bool = False,
) -> dict:
    """Execute Python code in an isolated subprocess.

    Args:
        code: Python source code to execute. Has access to speasy (spz), plasmapy (pf),
              numpy (np), scipy, matplotlib (Agg — plt.show() saves to disk),
              astropy units (u).
              Call export(name, array) to share numerical results with the LLM.
        timeout: maximum execution time in seconds — clamped to _MAX_TIMEOUT_S
        _plot_dir: injected by the agent loop — workspace dir for this run.
                   Not exposed in the LLM tool schema.

    Returns dict with:
        - stdout: captured text output
        - stderr: captured errors/warnings
        - figure_paths: list of absolute paths to saved PNG files
        - exports: dict of named numerical summaries (from export() calls)
        - error: error message if execution failed

    Example:
        >>> await run_python(
        ...     "import numpy as np\\n"
        ...     "export('rms', np.sqrt(np.mean(np.arange(8) ** 2)))\\n"
        ...     "print('done')"
        ... )
        {'stdout': 'done', 'exports': {'rms': {'mean': 4.1833..., 'min': 4.1833...,
         'max': 4.1833..., 'std': 0.0, 'n_finite': 1, 'n_nan': 0, ...}},
         'figure_paths': [], 'error': None, ...}
    """
    timeout = min(timeout, _MAX_TIMEOUT_S)
    if _plot_dir is None:
        from helioai.workspace import get_run_dir_for_sandbox

        _plot_dir = get_run_dir_for_sandbox()
    run_idx = _run_idx if _run_idx is not None else 0
    plot_dir = _plot_dir
    from helioai.logging_config import get_logger as _get_logger

    _get_logger(__name__).info("sandbox_plot_dir", plot_dir=plot_dir, run_idx=run_idx)
    code_file = Path(plot_dir, f"code_{run_idx}.py")
    dedented_code = textwrap.dedent(code)
    code_file.write_text(dedented_code, encoding="utf-8")
    n_lines = len(dedented_code.splitlines())
    plot_dir_line = f"__sandbox_plot_dir = {plot_dir!r}\n__sandbox_run_idx = {run_idx!r}\n"
    full_code = (
        plot_dir_line + _SANDBOX_PREAMBLE + textwrap.dedent(code) + "\n" + _SANDBOX_POSTAMBLE
    )

    speasy_seed: Path | None = None
    if _bwrap_works():
        from helioai.workspace import current_user

        speasy_seed = await asyncio.to_thread(_user_speasy_seed, current_user())
    cmd = _build_sandbox_cmd(
        plot_dir,
        no_net=_no_net,
        speasy_seed=str(speasy_seed) if speasy_seed else None,
    )
    program = full_code.encode("utf-8")
    using_bwrap = cmd[0].endswith("bwrap") if cmd else False

    try:
        if using_bwrap:
            sandbox_env = _sandbox_env(home=plot_dir)
            proc = await asyncio.create_subprocess_exec(
                *cmd,
                stdin=asyncio.subprocess.PIPE,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
                env=sandbox_env,
                start_new_session=True,
            )
        else:
            _warn_if_not_isolated()
            sandbox_env = _sandbox_env()
            proc = await asyncio.create_subprocess_exec(
                *cmd,
                stdin=asyncio.subprocess.PIPE,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
                env=sandbox_env,
                start_new_session=True,
                preexec_fn=_preexec_fn(),
                cwd=plot_dir,  # same working directory as the bwrap path's --chdir
            )
        try:
            stdout_bytes, stderr_bytes = await asyncio.wait_for(
                proc.communicate(input=program), timeout=timeout
            )
        except TimeoutError:
            _kill_proc_tree(proc)
            stdout_bytes, stderr_bytes = await proc.communicate()
            return {
                "error": f"Execution timed out after {timeout}s",
                "stdout": stdout_bytes.decode("utf-8", errors="replace")[-2000:],
                "stderr": stderr_bytes.decode("utf-8", errors="replace")[-2000:],
            }
        except asyncio.CancelledError:
            # The caller is gone (a closed SSE stream, a cancelled task): nothing will
            # read this result, so nothing should keep running to produce it.
            _kill_proc_tree(proc)
            raise

        stdout = stdout_bytes.decode("utf-8", errors="replace")
        stderr = stderr_bytes.decode("utf-8", errors="replace")

        figure_paths: list[str] = []
        exports: dict = {}
        cards: list[dict] = []
        clean_stdout_lines: list[str] = []
        for line in stdout.splitlines():
            if line.startswith("__HELIOAI_RESULT__"):
                try:
                    payload = json.loads(line[len("__HELIOAI_RESULT__") :])
                    figure_paths = payload.get("figure_paths", [])
                    exports = payload.get("exports", {})
                    cards = payload.get("cards", [])
                except json.JSONDecodeError:
                    pass
            else:
                clean_stdout_lines.append(line)

        _MAX_STDOUT = 4000
        clean_stdout = "\n".join(clean_stdout_lines).strip()
        if len(clean_stdout) > _MAX_STDOUT:
            clean_stdout = (
                clean_stdout[:_MAX_STDOUT]
                + f"\n[stdout truncated — {len(clean_stdout)} chars total; use export() for numerical data]"
            )

        if proc.returncode != 0:
            agent_stderr = _rewrite_traceback(stderr.strip())
            return {
                "error": _error_summary(agent_stderr, proc.returncode),
                "stdout": clean_stdout,
                "stderr": agent_stderr,
                "figure_paths": figure_paths,
                "exports": exports,
                "cards": cards,
                "code_path": str(code_file),
                "n_lines": n_lines,
            }

        return {
            "stdout": clean_stdout,
            "stderr": stderr.strip() if stderr.strip() else None,
            "figure_paths": figure_paths,
            "n_figures": len(figure_paths),
            "exports": exports,
            "cards": cards,
            "code_path": str(code_file),
            "n_lines": n_lines,
        }

    except Exception as e:
        return {"error": f"Sandbox error: {e}"}

Sandbox helpers

Available inside run_python, and re-emitted into exported notebooks.

helioai.tools.sandbox_helpers

Standalone physics helpers importable inside the sandbox.

MUST NOT import anything from helioai.* — the sandbox masks .env and strips the environment, so helioai.config would fail fast at import time.

Boundary models are clean-room implementations from the published papers: - Shue et al. (1998), JGR 103, 17691, doi:10.1029/98JA01103 - Jelinek et al. (2012), JGR 117, A05208, doi:10.1029/2011JA017252 Coordinate transforms wrap geopack (MIT, Tsyganenko models port).

transform_coords

transform_coords(time: Any, vectors: Any, frm: str = 'gse', to: str = 'gsm') -> np.ndarray

Rotate vectors between geocentric frames: gse, gsm, sm, geo, mag, gei.

Parameters:

Name	Type	Description	Default
time	Any	ISO string(s), datetime(s), numpy datetime64, or epoch seconds, all UTC. One timestamp per vector, or a single one for all.	required
vectors	Any	Shape (3,) or (N, 3).	required
frm	str	Source frame — gse, gsm, sm, geo, mag or gei.	'gse'
to	str	Target frame, same set.	'gsm'

Returns:

Type	Description
ndarray	The rotated vectors, same shape as the input.

The dipole tilt is recomputed per point (geopack.recalc), which is exact but linear in N — comfortable to ~1e4 points, not to a full mission.

Example

t = np.array(["2015-03-17T04:00:00"], dtype="datetime64[s]") transform_coords(t, np.array([[10.0, 0.0, 0.0]]), "gse", "gsm") array([[10., 0., 0.]]) # the X axis is shared by GSE and GSM

Source code in helioai/tools/sandbox_helpers.py

def transform_coords(time: Any, vectors: Any, frm: str = "gse", to: str = "gsm") -> np.ndarray:
    """Rotate vectors between geocentric frames: gse, gsm, sm, geo, mag, gei.

    Args:
        time: ISO string(s), datetime(s), numpy datetime64, or epoch seconds,
            all UTC. One timestamp per vector, or a single one for all.
        vectors: Shape `(3,)` or `(N, 3)`.
        frm: Source frame — gse, gsm, sm, geo, mag or gei.
        to: Target frame, same set.

    Returns:
        The rotated vectors, same shape as the input.

    The dipole tilt is recomputed per point (`geopack.recalc`), which is exact
    but linear in N — comfortable to ~1e4 points, not to a full mission.

    Example:
        >>> t = np.array(["2015-03-17T04:00:00"], dtype="datetime64[s]")
        >>> transform_coords(t, np.array([[10.0, 0.0, 0.0]]), "gse", "gsm")
        array([[10., 0., 0.]])   # the X axis is shared by GSE and GSM
    """
    from geopack import geopack as gp

    frm, to = frm.lower(), to.lower()
    to_gsm = {
        "gsm": lambda x, y, z: (x, y, z),
        "gse": lambda x, y, z: gp.gsmgse(x, y, z, -1),
        "sm": lambda x, y, z: gp.smgsm(x, y, z, 1),
        "geo": lambda x, y, z: gp.geogsm(x, y, z, 1),
        "mag": lambda x, y, z: gp.geogsm(*gp.geomag(x, y, z, -1), 1),
        "gei": lambda x, y, z: gp.geogsm(*gp.geigeo(x, y, z, 1), 1),
    }
    from_gsm = {
        "gsm": lambda x, y, z: (x, y, z),
        "gse": lambda x, y, z: gp.gsmgse(x, y, z, 1),
        "sm": lambda x, y, z: gp.smgsm(x, y, z, -1),
        "geo": lambda x, y, z: gp.geogsm(x, y, z, -1),
        "mag": lambda x, y, z: gp.geomag(*gp.geogsm(x, y, z, -1), 1),
        "gei": lambda x, y, z: gp.geigeo(*gp.geogsm(x, y, z, -1), -1),
    }
    if frm not in to_gsm or to not in from_gsm:
        raise ValueError(f"unknown frame: {frm!r} or {to!r} — use one of {sorted(to_gsm)}")

    vec = np.asarray(vectors, dtype=float)
    single = vec.ndim == 1
    vec = np.atleast_2d(vec)
    if vec.shape[1] != 3:
        raise ValueError(f"vectors must be (N, 3), got {vec.shape}")
    ts = _epoch_seconds(time)
    if ts.size == 1:
        ts = np.full(vec.shape[0], ts[0])
    if ts.size != vec.shape[0]:
        raise ValueError(f"{ts.size} times for {vec.shape[0]} vectors")

    out = np.empty_like(vec)
    for i in range(vec.shape[0]):
        gp.recalc(ts[i])
        out[i] = from_gsm[to](*to_gsm[frm](*vec[i]))
    return out[0] if single else out

mp_shue1998

mp_shue1998(pdyn_nPa: float, bz_nT: float, theta_deg: Any = None) -> tuple[np.ndarray, np.ndarray]

Shue et al. (1998) magnetopause: r = r0 * (2 / (1 + cos(theta)))**alpha.

r0 = (10.22 + 1.29tanh(0.184(Bz + 8.14))) * Pdyn(-1/6.6) alpha = (0.58 - 0.007Bz) * (1 + 0.024ln(Pdyn)) Args: pdyn_nPa: Solar wind dynamic pressure, nPa. bz_nT: IMF Bz in GSM, nT. Southward (negative) erodes the standoff. theta_deg: Angles from the Earth-Sun line. Defaults to 0..170 deg; beyond that the model is extrapolated past where it was fitted.

Returns:

Type	Description
tuple[ndarray, ndarray]	(theta_deg, r_RE) — the boundary in aberrated GSE, radii in R_E.

Reference: Shue et al. (1998), JGR 103, 17691, doi:10.1029/98JA01103.

Example

theta, r = mp_shue1998(2.0, -5.0) # Pdyn=2 nPa, Bz=-5 nT round(float(r[0]), 2), round(float(r[90]), 2) (9.81, 15.13) # standoff and flank, in R_E

Source code in helioai/tools/sandbox_helpers.py

def mp_shue1998(
    pdyn_nPa: float, bz_nT: float, theta_deg: Any = None
) -> tuple[np.ndarray, np.ndarray]:
    """Shue et al. (1998) magnetopause: r = r0 * (2 / (1 + cos(theta)))**alpha.

    r0 = (10.22 + 1.29*tanh(0.184*(Bz + 8.14))) * Pdyn**(-1/6.6)
    alpha = (0.58 - 0.007*Bz) * (1 + 0.024*ln(Pdyn))
    Args:
        pdyn_nPa: Solar wind dynamic pressure, nPa.
        bz_nT: IMF Bz in GSM, nT. Southward (negative) erodes the standoff.
        theta_deg: Angles from the Earth-Sun line. Defaults to 0..170 deg;
            beyond that the model is extrapolated past where it was fitted.

    Returns:
        `(theta_deg, r_RE)` — the boundary in aberrated GSE, radii in R_E.

    Reference: Shue et al. (1998), JGR 103, 17691, doi:10.1029/98JA01103.

    Example:
        >>> theta, r = mp_shue1998(2.0, -5.0)   # Pdyn=2 nPa, Bz=-5 nT
        >>> round(float(r[0]), 2), round(float(r[90]), 2)
        (9.81, 15.13)                            # standoff and flank, in R_E
    """
    theta = (
        np.linspace(0.0, 170.0, 171)
        if theta_deg is None
        else np.atleast_1d(np.asarray(theta_deg, dtype=float))
    )
    r0 = (10.22 + 1.29 * np.tanh(0.184 * (bz_nT + 8.14))) * pdyn_nPa ** (-1.0 / 6.6)
    alpha = (0.58 - 0.007 * bz_nT) * (1.0 + 0.024 * np.log(pdyn_nPa))
    r = r0 * (2.0 / (1.0 + np.cos(np.radians(theta)))) ** alpha
    return theta, r

bs_jelinek2012

bs_jelinek2012(pdyn_nPa: float, theta_deg: Any = None) -> tuple[np.ndarray, np.ndarray]

Jelinek et al. (2012) bow shock: parabola rho^2 = 4R(R - x) / lam^2.

R = 15.02 * Pdyn**(-1/6.55) is the subsolar standoff in R_E, lam = 1.17.

Parameters:

Name	Type	Description	Default
pdyn_nPa	float	Solar wind dynamic pressure, nPa.	required
theta_deg	Any	Angles from the Earth-Sun line. Defaults to 0..120 deg.	None

Returns:

Type	Description
ndarray	(theta_deg, r_RE), NaN where the parabola has no solution rather than
ndarray	a clipped value that would plot as a real boundary.

Reference: Jelinek et al. (2012), JGR 117, A05208, doi:10.1029/2011JA017252.

Example

theta, r = bs_jelinek2012(2.0) # Pdyn=2 nPa round(float(r[0]), 2) 13.51 # subsolar standoff, in R_E

Source code in helioai/tools/sandbox_helpers.py

def bs_jelinek2012(pdyn_nPa: float, theta_deg: Any = None) -> tuple[np.ndarray, np.ndarray]:
    """Jelinek et al. (2012) bow shock: parabola rho^2 = 4*R*(R - x) / lam^2.

    `R = 15.02 * Pdyn**(-1/6.55)` is the subsolar standoff in R_E, `lam = 1.17`.

    Args:
        pdyn_nPa: Solar wind dynamic pressure, nPa.
        theta_deg: Angles from the Earth-Sun line. Defaults to 0..120 deg.

    Returns:
        `(theta_deg, r_RE)`, NaN where the parabola has no solution rather than
        a clipped value that would plot as a real boundary.

    Reference: Jelinek et al. (2012), JGR 117, A05208, doi:10.1029/2011JA017252.

    Example:
        >>> theta, r = bs_jelinek2012(2.0)   # Pdyn=2 nPa
        >>> round(float(r[0]), 2)
        13.51                                 # subsolar standoff, in R_E
    """
    theta = (
        np.linspace(0.0, 120.0, 121)
        if theta_deg is None
        else np.atleast_1d(np.asarray(theta_deg, dtype=float))
    )
    lam = 1.17
    big_r = 15.02 * pdyn_nPa ** (-1.0 / 6.55)
    s, c = np.sin(np.radians(theta)), np.cos(np.radians(theta))
    with np.errstate(divide="ignore", invalid="ignore"):
        r = 2.0 * big_r * (np.sqrt(c**2 + lam**2 * s**2) - c) / (lam**2 * s**2)
    on_axis = np.isclose(s, 0.0)
    r = np.where(on_axis, np.where(c > 0, big_r, np.nan), r)
    return theta, r

Recipes

helioai.tools.recipes

Derived-recipe tools — list, load and run scientific Python recipes.

Recipes live in data/recipes/ as .py files with a YAML comment header: # name: theta_bn # description: Compute the shock normal angle theta_Bn from upstream/downstream B. # inputs: B_up (nT vec), B_dn (nT vec) # outputs: theta_bn_deg

list_recipes() returns the catalogue; load_recipe(name) returns the source code; run_recipe(name, inputs) executes it verbatim in the sandbox on the caller's inputs.

list_recipes async

list_recipes() -> dict

List all available derived recipes with their name, description and the call that runs them.

Each entry carries run_with, the run_recipe(...) call with the recipe's own input names or functions, so a model can go from the catalogue straight to running one: loading a recipe first cost one LLM call per recipe, and the call is where a session pays — every one re-sends the whole context.

Returns dict with 'recipes' list (sorted by name). Each entry has 'name', 'description', 'inputs', 'outputs' (when present in header) and 'run_with'. Returns {"recipes": []} when the recipes directory does not exist.

Example

await list_recipes() {'recipes': [{'name': 'fill_values', 'description': '...', 'run_with': '...'}, {'name': 'mvab', ...}, {'name': 'rankine_hugoniot', ...}, ...]}

Source code in helioai/tools/recipes.py

async def list_recipes() -> dict:
    """List all available derived recipes with their name, description and the call that runs them.

    Each entry carries `run_with`, the `run_recipe(...)` call with the recipe's own input
    names or functions, so a model can go from the catalogue straight to running one:
    loading a recipe first cost one LLM call per recipe, and the call is where a session
    pays — every one re-sends the whole context.

    Returns dict with 'recipes' list (sorted by name). Each entry has
    'name', 'description', 'inputs', 'outputs' (when present in header) and 'run_with'.
    Returns {"recipes": []} when the recipes directory does not exist.

    Example:
        >>> await list_recipes()
        {'recipes': [{'name': 'fill_values', 'description': '...', 'run_with': '...'},
                     {'name': 'mvab', ...}, {'name': 'rankine_hugoniot', ...}, ...]}
    """
    try:
        recipes_dir = settings.recipes.recipes_dir
        if not recipes_dir.exists():
            return {"recipes": []}
        entries = []
        for path in sorted(recipes_dir.glob("*.py")):
            try:
                text = path.read_text(encoding="utf-8")
                meta = _parse_header(text)
                entry = {"name": meta.get("name", path.stem)}
                for field in ("description", "inputs", "outputs"):
                    if field in meta:
                        entry[field] = meta[field]
                entry["run_with"] = run_with(entry["name"], text)
                entries.append(entry)
            except OSError as exc:
                log.warning("recipe_read_error", path=str(path), error=str(exc))
        return {"recipes": entries}
    except Exception as e:
        return {"error": str(e)}

load_recipe async

load_recipe(name: str) -> dict

Load the source code of a named recipe.

Parameters:

Name	Type	Description	Default
name	str	Recipe name without .py extension (e.g. 'theta_bn').	required

Returns dict with 'name' and 'code'. Returns {'error': ...} when not found or when the name contains path-traversal characters.

Example

await load_recipe("theta_bn")

Source code in helioai/tools/recipes.py

async def load_recipe(name: str) -> dict:
    """Load the source code of a named recipe.

    Args:
        name: Recipe name without .py extension (e.g. 'theta_bn').

    Returns dict with 'name' and 'code'. Returns {'error': ...} when not found
    or when the name contains path-traversal characters.

    Example:
        >>> await load_recipe("theta_bn")
        {'name': 'theta_bn', 'code': '# name: theta_bn\\n# description: Compute the shock...'}
    """
    try:
        path = _recipe_path(name)
        if path is None:
            return {"error": f"recipe {name!r} not found"}
        code = path.read_text(encoding="utf-8")
        meta = _parse_header(code)
        return {
            "name": meta.get("name", name),
            "code": code,
            "metadata": meta,
            "run_with": run_with(meta.get("name", name), code),
        }
    except Exception as e:
        return {"error": str(e)}

run_with

run_with(name: str, code: str) -> str

The one line that runs a recipe as shipped, read off its own source.

A model that has just read a recipe's code is one paste away from running a copy of it in run_python — which is how a 54.85° θ_Bn came out of a 12-minute window the recipe would never have chosen. The line names the tool and, exactly, what to bind.

A recipe declares its usual call in its header (# run:), because the names it reads are not a call: rankine_hugoniot reads eighteen, in three alternative bindings, and listed flat, alphabetically, they do not say which to bind. The other names it reads follow it. Without a declaration the line is derived: the variables read with globals().get for a script, the public functions for a library.

Parameters:

Name	Type	Description	Default
name	str	The recipe.	required
code	str	Its source.	required

Returns:

Type	Description
str	The declared call, or a run_recipe(...) template with the recipe's own input
str	names or functions.

Source code in helioai/tools/recipes.py

def run_with(name: str, code: str) -> str:
    """The one line that runs a recipe as shipped, read off its own source.

    A model that has just read a recipe's code is one paste away from running a copy of
    it in `run_python` — which is how a 54.85° θ_Bn came out of a 12-minute window the
    recipe would never have chosen. The line names the tool and, exactly, what to bind.

    A recipe declares its usual call in its header (`# run:`), because the names it reads
    are not a call: `rankine_hugoniot` reads eighteen, in three alternative bindings,
    and listed flat, alphabetically, they do not say which to bind. The other names it
    reads follow it. Without a declaration the line is derived: the
    variables read with `globals().get` for a script, the public functions for a library.

    Args:
        name: The recipe.
        code: Its source.

    Returns:
        The declared call, or a `run_recipe(...)` template with the recipe's own input
        names or functions.
    """
    inputs = _globals_read(code)
    declared = _parse_header(code).get("run")
    if declared:
        named = _declared_inputs(declared)
        others = [i for i in inputs if i not in named]
        if not named:
            return declared
        line = f"{declared} — replace each <...> with yours"
        if others:
            line += f"; other inputs it reads: {', '.join(others)} — its inputs say what each is"
        return line
    if inputs:
        bound = ", ".join(f"{i!r}: ..." for i in inputs)
        return (
            f"run_recipe({name!r}, inputs={{{bound}}}) — bind the inputs you have (each a "
            f"Python expression such as \"load_data('name')\" or a literal); the recipe's "
            f"source then runs verbatim on the session's data"
        )
    functions = [f for f in _PUBLIC_DEF.findall(code) if f != "export"]
    if functions:
        example = f"{functions[-1]}(...)"
        return (
            f"run_recipe({name!r}, inputs={{...}}, call={example!r}) — a library of functions "
            f"({', '.join(functions[:6])}); bind their arguments as inputs and name the call"
        )
    return (
        f"run_recipe({name!r}, inputs={{...}}) runs the recipe's source verbatim on the "
        f"session's data"
    )

recipe_script

recipe_script(name: str, code: str, inputs: dict, call: str | None) -> str

Assemble the script run_recipe executes: inputs, the recipe, the call.

The recipe's source is inserted verbatim — not a function copied out of it, not a formula rewritten from memory — so its calibrated constants, its checks and its own export() calls run as shipped. __name__ is set first so a recipe that guards a demo behind if __name__ == "__main__": runs its functions and not its demo, the way it would if imported. The call comes last, for the recipes that are a library of functions rather than a script.

Parameters:

Name	Type	Description	Default
name	str	The recipe.	required
code	str	Its source, as load_recipe returns it.	required
inputs	dict	{variable: expression | literal} bound before the recipe.	required
call	str | None	An expression evaluated after it, or None.	required

Returns:

Type	Description
str	The Python source, as it is written to the session's code_N.py.

Source code in helioai/tools/recipes.py

def recipe_script(name: str, code: str, inputs: dict, call: str | None) -> str:
    """Assemble the script `run_recipe` executes: inputs, the recipe, the call.

    The recipe's source is inserted verbatim — not a function copied out of it, not a
    formula rewritten from memory — so its calibrated constants, its checks and its own
    `export()` calls run as shipped. `__name__` is set first so a recipe that guards a
    demo behind `if __name__ == "__main__":` runs its functions and not its demo, the way
    it would if imported. The call comes last, for the recipes that are a library of
    functions rather than a script.

    Args:
        name: The recipe.
        code: Its source, as `load_recipe` returns it.
        inputs: `{variable: expression | literal}` bound before the recipe.
        call: An expression evaluated after it, or None.

    Returns:
        The Python source, as it is written to the session's `code_N.py`.
    """
    parts = [
        f"# run_recipe: {name} — inputs, then the recipe verbatim; not __main__, so a demo",
        '# the recipe guards behind `if __name__ == "__main__":` stays off',
        '__name__ = "recipe"',
    ]
    parts += _bindings(inputs)
    parts += ["", f"# ── recipe {name} ──", code.rstrip("\n"), ""]
    if call and call.strip():
        parts += [
            "# ── call ──",
            f"_recipe_result = ({call.strip()})",
            "print(repr(_recipe_result))",
        ]
    return "\n".join(parts) + "\n"

run_recipe async

run_recipe(name: str, inputs: dict | None = None, call: str | None = None, timeout: float = 60.0, _plot_dir: str | None = None, _run_idx: int | None = None, _no_net: bool = False) -> dict

Run a shipped recipe on the session's data, without the model rewriting it.

load_recipe hands the model the source, and the model then pastes a part of it into run_python — or reads the constants and rewrites the computation by hand, which is what the recipe check keeps catching. Here the recipe runs as shipped: the inputs are bound first, the source follows verbatim, and an optional call applies one of its functions. The exports are the recipe's own, the script written to the workspace is the one the export reproduces, and the run is recorded as a use of the recipe (method_used) with its reference.

Parameters:

Name	Type	Description	Default
name	str	Recipe name, as list_recipes lists it.	required
inputs	dict | None	{variable: value} bound before the recipe runs. A string is a Python expression evaluated in the sandbox ("load_data('b').values[:20]"); a number or list is used as the literal it is.	None
call	str | None	An expression evaluated after the recipe, for a recipe that is a library of functions — "rh_jump(n_u, n_d, V_u, V_d, B_u, B_d)"; its value is printed. Not needed for a recipe whose run block reads its inputs itself.	None
timeout	float	Sandbox time budget in seconds.	60.0
_plot_dir	str | None	Session workspace, injected by the runtime.	None
_run_idx	int | None	Script index in that workspace, injected by the runtime.	None
_no_net	bool	Deny the sandbox a network namespace, injected by the runtime.	False

Returns:

Type	Description
dict	The run_python result — stdout, exports, figures, code_path — plus
dict	recipe (name, reference, description), inputs as bound, and a
dict	method_used card. When the run failed, or produced nothing at all — six of
dict	the recipes read their inputs with globals().get and do nothing, silently,
dict	when a name is bound wrong — it also carries recipe_notice: the recipe's
dict	usage, public signatures and run_with, how it is called without its source.
dict	{"error": ...} for an unknown recipe (with the names there are) or an input
dict	that is not a Python name (with the notice).

Source code in helioai/tools/recipes.py

async def run_recipe(
    name: str,
    inputs: dict | None = None,
    call: str | None = None,
    timeout: float = 60.0,
    _plot_dir: str | None = None,
    _run_idx: int | None = None,
    _no_net: bool = False,
) -> dict:
    """Run a shipped recipe on the session's data, without the model rewriting it.

    `load_recipe` hands the model the source, and the model then pastes a part of it
    into `run_python` — or reads the constants and rewrites the computation by hand,
    which is what the recipe check keeps catching. Here the recipe runs as shipped: the
    inputs are bound first, the source follows verbatim, and an optional call applies
    one of its functions. The exports are the recipe's own, the script written to the
    workspace is the one the export reproduces, and the run is recorded as a use of the
    recipe (`method_used`) with its reference.

    Args:
        name: Recipe name, as `list_recipes` lists it.
        inputs: `{variable: value}` bound before the recipe runs. A string is a Python
            expression evaluated in the sandbox (`"load_data('b').values[:20]"`); a
            number or list is used as the literal it is.
        call: An expression evaluated after the recipe, for a recipe that is a library
            of functions — `"rh_jump(n_u, n_d, V_u, V_d, B_u, B_d)"`; its value is
            printed. Not needed for a recipe whose run block reads its inputs itself.
        timeout: Sandbox time budget in seconds.
        _plot_dir: Session workspace, injected by the runtime.
        _run_idx: Script index in that workspace, injected by the runtime.
        _no_net: Deny the sandbox a network namespace, injected by the runtime.

    Returns:
        The `run_python` result — stdout, exports, figures, `code_path` — plus
        `recipe` (`name`, `reference`, `description`), `inputs` as bound, and a
        `method_used` card. When the run failed, or produced nothing at all — six of
        the recipes read their inputs with `globals().get` and do nothing, silently,
        when a name is bound wrong — it also carries `recipe_notice`: the recipe's
        usage, public signatures and `run_with`, how it is called without its source.
        `{"error": ...}` for an unknown recipe (with the names there are) or an input
        that is not a Python name (with the notice).
    """
    from helioai.tools.sandbox import run_python

    path = _recipe_path(name)
    if path is None:
        names = sorted(p.stem for p in settings.recipes.recipes_dir.glob("*.py"))
        return {
            "error": f"recipe {name!r} not found; call list_recipes for the names",
            "recipes": names,
        }
    code = path.read_text(encoding="utf-8")
    meta = _parse_header(code)
    bound = dict(inputs or {})
    try:
        script = recipe_script(name, code, bound, call)
    except ValueError as e:
        return {"error": str(e), "recipe_notice": _describe(meta.get("name", name), code)}

    result = await run_python(
        script, timeout=timeout, _plot_dir=_plot_dir, _run_idx=_run_idx, _no_net=_no_net
    )
    recipe = {
        "name": meta.get("name", name),
        "reference": meta.get("reference", ""),
        "description": meta.get("description", ""),
    }
    result["recipe"] = recipe
    result["inputs"] = bound
    produced = (
        result.get("exports") or result.get("figure_paths") or (result.get("stdout") or "").strip()
    )
    if "error" in result or not produced:
        result["recipe_notice"] = _describe(recipe["name"], code)
    if "error" not in result:
        result.setdefault("cards", []).append(
            {
                "kind": "method_used",
                "name": recipe["name"],
                "reference": recipe["reference"],
                "method": recipe["description"],
            }
        )
    return result

Literature

helioai.tools.literature

NASA ADS literature search — find_papers tool.

find_papers async

find_papers(query: str, max_results: int = 5, year_start: int | None = None, year_end: int | None = None, sort: str = 'relevance', _transport: AsyncBaseTransport | None = None) -> dict

Search NASA ADS for papers relevant to an event, parameter or method.

Requires ADS_API_TOKEN; without it the tool returns an error rather than raising, so the agent can tell the user what is missing.

Parameters:

Name	Type	Description	Default
query	str	Free-text ADS query — event, parameter, method or author.	required
max_results	int	How many papers to return.	5
year_start	int | None	Earliest publication year, inclusive. None leaves it open.	None
year_end	int | None	Latest publication year, inclusive. None leaves it open.	None
sort	str	relevance, citation_count or date, passed to ADS as given.	'relevance'
_transport	AsyncBaseTransport | None	Test seam for injecting an httpx transport. Not part of the tool schema the model sees.	None

Returns:

Type	Description
dict	{"query", "papers", "note"}, where each paper carries title, authors,
dict	year, bibcode, doi, citations and abstract. On a missing token or an ADS
dict	failure, an error key instead — never an exception, because the agent
dict	has to be able to report the cause.

Example

await find_papers("interplanetary shock Rankine-Hugoniot multi-spacecraft", ... max_results=2) {'query': '...', 'papers': [ {'title': 'Multiple spacecraft observations of interplanetary shocks: ...', 'authors': 'Russell, C. T. et al.', 'year': '1983', 'bibcode': '1983JGR....88.9941R', 'doi': '10.1029/JA088iA12p09941', 'citations': 72, 'abstract': '...'}, ...], 'note': '...'}

Source code in helioai/tools/literature.py

async def find_papers(
    query: str,
    max_results: int = 5,
    year_start: int | None = None,
    year_end: int | None = None,
    sort: str = "relevance",
    _transport: httpx.AsyncBaseTransport | None = None,
) -> dict:
    """Search NASA ADS for papers relevant to an event, parameter or method.

    Requires `ADS_API_TOKEN`; without it the tool returns an error rather than
    raising, so the agent can tell the user what is missing.

    Args:
        query: Free-text ADS query — event, parameter, method or author.
        max_results: How many papers to return.
        year_start: Earliest publication year, inclusive. None leaves it open.
        year_end: Latest publication year, inclusive. None leaves it open.
        sort: `relevance`, `citation_count` or `date`, passed to ADS as given.
        _transport: Test seam for injecting an httpx transport. Not part of the
            tool schema the model sees.

    Returns:
        `{"query", "papers", "note"}`, where each paper carries title, authors,
        year, bibcode, doi, citations and abstract. On a missing token or an ADS
        failure, an `error` key instead — never an exception, because the agent
        has to be able to report the cause.

    Example:
        >>> await find_papers("interplanetary shock Rankine-Hugoniot multi-spacecraft",
        ...                   max_results=2)
        {'query': '...', 'papers': [
         {'title': 'Multiple spacecraft observations of interplanetary shocks: ...',
          'authors': 'Russell, C. T. et al.', 'year': '1983',
          'bibcode': '1983JGR....88.9941R', 'doi': '10.1029/JA088iA12p09941',
          'citations': 72, 'abstract': '...'}, ...], 'note': '...'}
    """
    token = settings.literature.ads_token
    if not token:
        return {
            "error": (
                "ADS_API_TOKEN is not set — get a free key at "
                "https://ui.adsabs.harvard.edu/user/settings/token and add it to .env"
            )
        }

    years = f" year:{year_start or ''}-{year_end or ''}" if (year_start or year_end) else ""
    q = f"{query}{years}"
    rows = min(max(max_results, 1), _MAX_ROWS)

    async with httpx.AsyncClient(timeout=15, transport=_transport) as client:
        try:
            resp = await _search(client, token, q, rows, _SORTS.get(sort, _SORTS["relevance"]))
        except httpx.HTTPError as e:
            return {"error": f"ADS request failed: {e}"}
        if resp.status_code != 200:
            return {"error": f"ADS returned HTTP {resp.status_code}: {resp.text[:200]}"}
        docs = resp.json().get("response", {}).get("docs", [])

        widened = _widen(query)
        relaxed_query = f"{widened} {_WIDENED}{years}" if widened else None
        more: list[dict] = []
        if relaxed_query and len(docs) < rows:
            try:
                extra = await _search(client, token, relaxed_query, rows, _SORTS["relevance"])
            except httpx.HTTPError:
                extra = None
            if extra is not None and extra.status_code == 200:
                more = extra.json().get("response", {}).get("docs", [])
            else:
                relaxed_query = None
        else:
            relaxed_query = None

    seen = {d.get("bibcode") for d in docs}
    docs = docs + [d for d in more if d.get("bibcode") not in seen][: rows - len(docs)]
    papers = [_slim(d) for d in docs]
    log.info("find_papers", query=query, n_results=len(papers), relaxed=relaxed_query is not None)
    out = {
        "query": q,
        "papers": papers,
        "note": (
            "Cite as: Authors (year), bibcode. "
            "Full record: https://ui.adsabs.harvard.edu/abs/<bibcode>"
        ),
        "relaxed": relaxed_query is not None,
    }
    if relaxed_query is not None:
        out["relaxed_query"] = relaxed_query
    return out

Registry

helioai.tools.registry

Tool registry: wraps Python async functions with JSON Schema metadata.

The agent loop calls registry.call_tool(name, args) and never imports tool modules directly, which keeps the dependency surface small and makes sub-agent tool whitelisting trivial. Every call comes back as a ToolResult (tools/results.py): the payload once, the model's text once.

Tool dataclass

A registered tool: an async function plus the schema shown to the model.

Source code in helioai/tools/registry.py

@dataclass
class Tool:
    """A registered tool: an async function plus the schema shown to the model."""

    name: str
    description: str
    parameters: dict  # JSON Schema object
    func: Callable[..., Coroutine[Any, Any, Any]]
    read_only: bool = True

ToolRegistry

Maps tool names to async functions and their JSON Schemas.

The agent loop dispatches through here and never imports tool modules, which keeps the dependency surface small and makes per-role whitelisting trivial.

Source code in helioai/tools/registry.py

class ToolRegistry:
    """Maps tool names to async functions and their JSON Schemas.

    The agent loop dispatches through here and never imports tool modules, which
    keeps the dependency surface small and makes per-role whitelisting trivial.
    """

    def __init__(self) -> None:
        self._tools: dict[str, Tool] = {}

    def register(
        self, name: str, description: str, parameters: dict, *, read_only: bool = True
    ) -> Callable:
        """Decorator that registers an async function as a tool.

        Args:
            read_only: False for a tool that changes something a user would care about
                — running arbitrary code, writing a catalogue. MCP clients gate
                auto-approval on it, so the default is the safe-to-repeat majority and
                the two exceptions say so at their registration site. Kept here rather
                than in a table beside the MCP server: the registry is the one place
                that already describes every tool.
        """

        def decorator(func: Callable) -> Callable:
            self._tools[name] = Tool(
                name=name,
                description=description,
                parameters=parameters,
                func=func,
                read_only=read_only,
            )
            return func

        return decorator

    def list_tool_defs(self, only: set[str] | None = None) -> list[ToolDef]:
        """Return tool definitions for the model.

        Args:
        only: Restrict to these names — how sub-agent whitelists are applied.
        None returns every registered tool.
        """
        tools = self._tools.values()
        if only is not None:
            tools = [t for t in tools if t.name in only]
        return [
            ToolDef(name=t.name, description=t.description, parameters=t.parameters) for t in tools
        ]

    async def call_tool(
        self, name: str, arguments: dict | None, *, trusted: dict | None = None
    ) -> ToolResult:
        """Invoke a tool and return its result, typed.

        Never raises: an unknown tool, a rejected private argument and any exception
        the tool lets out all come back as a failed `ToolResult`, whose `for_llm()` is
        the `{"error": ...}` line this method used to return as a string.

        Args:
            name: Registered tool name.
            arguments: Caller-supplied (LLM/MCP) arguments; may not carry private `_*`
                keys.
            trusted: Framework-injected arguments (the sandbox output directory) that
                bypass that guard.

        Returns:
            The tool's payload and the exact text the model will read.
        """
        if name not in self._tools:
            return ToolResult.failure(name, f"unknown tool {name!r}")
        if arguments and any(k.startswith("_") for k in arguments):
            bad = sorted(k for k in arguments if k.startswith("_"))
            return ToolResult.failure(name, f"rejected private argument(s): {bad}")
        try:
            result = await self._tools[name].func(**{**(arguments or {}), **(trusted or {})})
        except Exception as e:
            return ToolResult.failure(name, str(e) or type(e).__name__)
        return ToolResult.from_raw(name, result)

    def is_read_only(self, name: str) -> bool:
        """Whether the tool leaves the user's world unchanged. Unknown names count as not."""
        tool = self._tools.get(name)
        return bool(tool and tool.read_only)

    def __contains__(self, name: str) -> bool:
        return name in self._tools

register

register(name: str, description: str, parameters: dict, *, read_only: bool = True) -> Callable

Decorator that registers an async function as a tool.

Parameters:

Name	Type	Description	Default
read_only	bool	False for a tool that changes something a user would care about — running arbitrary code, writing a catalogue. MCP clients gate auto-approval on it, so the default is the safe-to-repeat majority and the two exceptions say so at their registration site. Kept here rather than in a table beside the MCP server: the registry is the one place that already describes every tool.	True

Source code in helioai/tools/registry.py

def register(
    self, name: str, description: str, parameters: dict, *, read_only: bool = True
) -> Callable:
    """Decorator that registers an async function as a tool.

    Args:
        read_only: False for a tool that changes something a user would care about
            — running arbitrary code, writing a catalogue. MCP clients gate
            auto-approval on it, so the default is the safe-to-repeat majority and
            the two exceptions say so at their registration site. Kept here rather
            than in a table beside the MCP server: the registry is the one place
            that already describes every tool.
    """

    def decorator(func: Callable) -> Callable:
        self._tools[name] = Tool(
            name=name,
            description=description,
            parameters=parameters,
            func=func,
            read_only=read_only,
        )
        return func

    return decorator

list_tool_defs

list_tool_defs(only: set[str] | None = None) -> list[ToolDef]

Return tool definitions for the model.

Args: only: Restrict to these names — how sub-agent whitelists are applied. None returns every registered tool.

Source code in helioai/tools/registry.py

def list_tool_defs(self, only: set[str] | None = None) -> list[ToolDef]:
    """Return tool definitions for the model.

    Args:
    only: Restrict to these names — how sub-agent whitelists are applied.
    None returns every registered tool.
    """
    tools = self._tools.values()
    if only is not None:
        tools = [t for t in tools if t.name in only]
    return [
        ToolDef(name=t.name, description=t.description, parameters=t.parameters) for t in tools
    ]

call_tool async

call_tool(name: str, arguments: dict | None, *, trusted: dict | None = None) -> ToolResult

Invoke a tool and return its result, typed.

Never raises: an unknown tool, a rejected private argument and any exception the tool lets out all come back as a failed ToolResult, whose for_llm() is the {"error": ...} line this method used to return as a string.

Parameters:

Name	Type	Description	Default
name	str	Registered tool name.	required
arguments	dict | None	Caller-supplied (LLM/MCP) arguments; may not carry private _* keys.	required
trusted	dict | None	Framework-injected arguments (the sandbox output directory) that bypass that guard.	None

Returns:

Type	Description
ToolResult	The tool's payload and the exact text the model will read.

Source code in helioai/tools/registry.py

async def call_tool(
    self, name: str, arguments: dict | None, *, trusted: dict | None = None
) -> ToolResult:
    """Invoke a tool and return its result, typed.

    Never raises: an unknown tool, a rejected private argument and any exception
    the tool lets out all come back as a failed `ToolResult`, whose `for_llm()` is
    the `{"error": ...}` line this method used to return as a string.

    Args:
        name: Registered tool name.
        arguments: Caller-supplied (LLM/MCP) arguments; may not carry private `_*`
            keys.
        trusted: Framework-injected arguments (the sandbox output directory) that
            bypass that guard.

    Returns:
        The tool's payload and the exact text the model will read.
    """
    if name not in self._tools:
        return ToolResult.failure(name, f"unknown tool {name!r}")
    if arguments and any(k.startswith("_") for k in arguments):
        bad = sorted(k for k in arguments if k.startswith("_"))
        return ToolResult.failure(name, f"rejected private argument(s): {bad}")
    try:
        result = await self._tools[name].func(**{**(arguments or {}), **(trusted or {})})
    except Exception as e:
        return ToolResult.failure(name, str(e) or type(e).__name__)
    return ToolResult.from_raw(name, result)

is_read_only

is_read_only(name: str) -> bool

Whether the tool leaves the user's world unchanged. Unknown names count as not.

Source code in helioai/tools/registry.py

def is_read_only(self, name: str) -> bool:
    """Whether the tool leaves the user's world unchanged. Unknown names count as not."""
    tool = self._tools.get(name)
    return bool(tool and tool.read_only)

MCP client

helioai.tools.mcp_client

Generic MCP client — mounts remote MCP server tools into the ToolRegistry.

Configured via HELIOAI_MCP_SERVERS, a JSON object keyed by server alias: {"ads": {"url": "https://.../mcp", "headers": {"Authorization": "Bearer x"}}, "alphaxiv": {"command": "npx", "args": ["-y", "mcp-remote", "https://api.alphaxiv.org/mcp/v1"]}}

Remote tools are registered as "_" and proxied with a fresh connection per call: the CLI runs one asyncio.run() per query, so a persistent session would not outlive a single request anyway.

discover_and_register async

discover_and_register() -> list[str]

Connect to every configured MCP server and register its tools.

Idempotent: only the first call does any work. Servers that are unreachable are logged and skipped rather than fatal — a dead remote must not stop HelioAI from starting.

Returns:

Type	Description
list[str]	The names under which remote tools were registered, prefixed by alias.

Source code in helioai/tools/mcp_client.py

async def discover_and_register() -> list[str]:
    """Connect to every configured MCP server and register its tools.

    Idempotent: only the first call does any work. Servers that are unreachable
    are logged and skipped rather than fatal — a dead remote must not stop
    HelioAI from starting.

    Returns:
        The names under which remote tools were registered, prefixed by alias.
    """
    global _discovered
    if _discovered:
        return []
    _discovered = True

    registered: list[str] = []
    for alias, spec in _server_specs().items():
        try:
            # `spec` is bound as a default: the closure would otherwise read the
            # loop variable at await time, so every server after the first would
            # be discovered against the last spec if this were ever deferred.
            async def _list(spec=spec):
                async with _session(spec) as session:
                    return await session.list_tools()

            tools = (await asyncio.wait_for(_list(), timeout=_DISCOVER_TIMEOUT_S)).tools
        except Exception as e:
            log.warning("mcp_server_unreachable", server=alias, error=str(e))
            continue

        for t in tools:
            name = _safe_name(alias, t.name)
            if name in registry:
                log.warning("mcp_tool_collision_skipped", server=alias, tool=name)
                continue
            schema = t.input_schema or {"type": "object", "properties": {}}
            props = schema.get("properties") or {}
            if any(k.startswith("_") for k in props):
                log.warning("mcp_tool_private_args", server=alias, tool=name)
            registry.register(
                name=name,
                description=f"[{alias} MCP] {t.description or t.name}",
                parameters=schema,
            )(_make_proxy(alias, spec, t.name))
            registered.append(name)
        log.info("mcp_server_mounted", server=alias, n_tools=len(tools))
    return registered