Interfaces

See Interfaces for how to use each surface.

Command line

helioai.interfaces.cli

Interactive CLI for HelioAI.

Usage

helioai # interactive readline session helioai "your query" # one-shot query helioai index # rebuild speasy catalog index helioai index --rebuild # force full reindex helioai export [id] # export a session as a reproducible .ipynb

main

main() -> None

Entry point for the helioai command.

Routes subcommands (index, export, history, delete, profile, serve, ...) and otherwise runs either a one-shot query or the interactive prompt.

Source code in helioai/interfaces/cli.py

def main() -> None:
    """Entry point for the `helioai` command.

    Routes subcommands (index, export, history, delete, profile, serve, ...) and
    otherwise runs either a one-shot query or the interactive prompt.
    """
    global _SESSION_ID
    from helioai.config import dev_unlock, settings
    from helioai.workspace import cleanup_old_runs, set_user

    set_user(_USER_ID)
    cleanup_old_runs()
    args = sys.argv[1:]

    # --dev: supply the configured dev token to bypass the scope guardrail
    dev_flag = "--dev" in args
    if dev_flag:
        args = [a for a in args if a != "--dev"]
    restricted = not dev_unlock(settings.dev.token if dev_flag else None)

    if "--session" in args:
        idx = args.index("--session")
        if idx + 1 < len(args):
            _SESSION_ID = args[idx + 1]
            args = [a for i, a in enumerate(args) if i not in (idx, idx + 1)]

    if not args:
        _interactive(restricted=restricted)
        return

    if args[0] == "history":
        if len(args) >= 3 and args[1] == "delete":
            _delete_session(args[2])
        else:
            _show_history()
        return

    if args[0] == "index":
        _run_index(rebuild="--rebuild" in args)
        return

    if args[0] == "profile":
        _run_profile()
        return

    if args[0] == "export":
        _run_export(args[1] if len(args) > 1 else None)
        return

    if args[0] == "migrate-storage":
        _run_migrate_storage()
        return

    if args[0] == "serve":
        if "--web" in args:
            serve_args = args[1:]
            host = "127.0.0.1"
            port = 7890
            if "--host" in serve_args:
                idx = serve_args.index("--host")
                host = serve_args[idx + 1]
            if "--port" in serve_args:
                idx = serve_args.index("--port")
                port = int(serve_args[idx + 1])
            from helioai.interfaces.web.app import serve_web

            serve_web(host=host, port=port)
        else:
            from helioai.mcp_server import main as mcp_main

            sys.argv = [sys.argv[0]] + args[1:]
            mcp_main()
        return

    if "--resume" in args:
        session_id = _pick_session()
        if session_id:
            _SESSION_ID = session_id
        _interactive(restricted=restricted)
        return

    query = " ".join(args)
    asyncio.run(_run_query(query, restricted=restricted))

Jupyter magic

helioai.interfaces.jupyter_magic

Jupyter IPython magics for HelioAI.

Load with

%load_ext helioai.interfaces.jupyter_magic

Cell magic

%%helioai solar wind density ACE 2005-01-17

Line magics

%helioai_session reset %helioai_provider groq|gemini|azure %helioai_history %helioai_resume %helioai_export [session_id] %helioai_dev on|off — toggle dev mode (bypasses helio-only scope guardrail)

HelioAIMagics

Bases: Magics

IPython magics exposing the agent inside a notebook.

Source code in helioai/interfaces/jupyter_magic.py

@magics_class
class HelioAIMagics(Magics):
    """IPython magics exposing the agent inside a notebook."""

    @cell_magic
    def helioai(self, line: str, cell: str) -> None:
        """`%%helioai` — send a natural-language query to the agent.

        Figures render inline; parameter cards and catalog previews render as HTML.

        Example:
            %load_ext helioai.interfaces.jupyter_magic

            %%helioai
            Download ACE IMF for the 2015-03-17 storm, plot Bz and mark
            the shock arrival.
        """
        import helioai.tools.setup  # noqa: F401
        from helioai.core.agent_loop import stream_chat
        from helioai.logging_config import setup_logging

        setup_logging("WARNING")

        async def _run():
            llm = _get_llm()
            try:
                async for ev in stream_chat(
                    llm, _USER_ID, _SESSION_ID, cell.strip(), restricted=_dev_restricted
                ):
                    _render_jupyter_event(ev)
            finally:
                # Must happen inside this loop: the pool is bound to it, and
                # `_run_async` closes the loop the moment this returns.
                await llm.aclose()

        _run_async(_run())

    @line_magic
    def helioai_session(self, line: str) -> None:
        """`%helioai_session [id]` — show or switch the active session."""
        global _SESSION_ID
        parts = line.strip().split(maxsplit=1)
        cmd = parts[0] if parts else ""
        arg = parts[1] if len(parts) > 1 else ""

        if cmd == "reset":
            from helioai.core.session import store

            store.reset(_USER_ID, _SESSION_ID)
            _SESSION_ID = str(uuid.uuid4())
            print(f"Session reset. New id: {_SESSION_ID[:8]}")
        elif cmd == "delete":
            from helioai.core.session import store
            from helioai.workspace import _root

            if not arg:
                print("Usage: %helioai_session delete <session_id_prefix>")
                return
            all_ids = store.all_sessions(_USER_ID)
            matches = [s for s in all_ids if s.startswith(arg)]
            if not matches:
                print(f"No session matching {arg!r}.")
                return
            sid = matches[0]
            wdir = store.get_workspace_dir(_USER_ID, sid)
            store.reset(_USER_ID, sid)
            if wdir:
                import shutil

                ws_path = _root() / wdir
                if ws_path.exists():
                    shutil.rmtree(ws_path, ignore_errors=True)
            if sid == _SESSION_ID:
                _SESSION_ID = str(uuid.uuid4())
                print(f"Current session deleted. New id: {_SESSION_ID[:8]}")
            else:
                print(f"Session {sid[:8]} deleted.")
        else:
            print(f"Unknown command: {cmd!r}. Use 'reset' or 'delete <id>'.")

    @line_magic
    def helioai_provider(self, line: str) -> None:
        """`%helioai_provider [name]` — show or switch the LLM provider."""
        provider = line.strip().lower()
        if provider not in ("groq", "gemini", "azure", "opencode", "ollama"):
            print(f"Unknown provider {provider!r}. Use: groq | gemini | azure | opencode | ollama")
            return
        import os

        os.environ["HELIOAI_LLM_PROVIDER"] = provider
        print(f"Provider switched to {provider!r}.")

    @line_magic
    def helioai_history(self, line: str) -> None:
        """`%helioai_history` — list recent sessions."""
        from helioai.core.session import store

        summaries = store.list_summaries(_USER_ID)
        if not summaries:
            print("No history found.")
            return
        rows = "".join(
            f"<tr>"
            f"<td><code>{s['session_id'][:8]}</code></td>"
            f"<td>{s['updated_at'][:16].replace('T', ' ')}</td>"
            f"<td style='text-align:center'>{s['n_messages']}</td>"
            f"<td>{s['first_message']}</td>"
            f"</tr>"
            for s in summaries
        )
        display(
            HTML(
                "<table><thead><tr>"
                "<th>Session</th><th>Updated</th><th>Msgs</th><th>First message</th>"
                "</tr></thead><tbody>" + rows + "</tbody></table>"
            )
        )

    @line_magic
    def helioai_profile(self, line: str) -> None:
        """`%helioai_profile` — show or edit the user profile."""
        from helioai.workspace import user_home

        parts = line.strip().split(maxsplit=1)
        cmd = parts[0] if parts else ""
        arg = parts[1].strip().strip("\"'") if len(parts) > 1 else ""
        p = user_home(_USER_ID) / "profile.md"

        if cmd == "show":
            content = p.read_text(encoding="utf-8").strip() if p.exists() else ""
            display(Markdown(content if content else "_(profil vide)_"))
        elif cmd == "set":
            if not arg:
                print('Usage: %helioai_profile set "your preferences here"')
                return
            p.parent.mkdir(parents=True, exist_ok=True)
            with p.open("a", encoding="utf-8") as f:
                f.write(("\n" if p.stat().st_size > 0 else "") + arg + "\n")
            print(f"Profile updated ({p}).")
        else:
            print('Usage: %helioai_profile show | set "<text>"')

    @line_magic
    def helioai_export(self, line: str) -> None:
        """`%helioai_export` — export the session as a standalone notebook."""
        from helioai.core.session import store
        from helioai.export import export_session_notebook

        prefix = line.strip()
        session_id = _SESSION_ID
        if prefix:
            matches = [s for s in store.all_sessions(_USER_ID) if s.startswith(prefix)]
            if not matches:
                print(f"No session matching {prefix!r}.")
                return
            session_id = matches[0]
        path = export_session_notebook(_USER_ID, session_id)
        from IPython.display import FileLink

        display(FileLink(str(path), result_html_prefix="📓 Exported notebook: "))

    @line_magic
    def helioai_resume(self, line: str) -> None:
        """`%helioai_resume` — pick a previous session to continue."""
        global _SESSION_ID
        prefix = line.strip()
        if not prefix:
            print("Usage: %helioai_resume <session_id>")
            return
        from helioai.core.session import store

        all_ids = store.all_sessions(_USER_ID)
        matches = [s for s in all_ids if s.startswith(prefix)]
        if not matches:
            print(f"No session found matching {prefix!r}.")
            return
        _SESSION_ID = matches[0]
        msgs = store.get_or_create(_USER_ID, _SESSION_ID)
        print(f"Resumed session {_SESSION_ID[:8]} ({len(msgs)} messages).")

    @line_magic
    def helioai_dev(self, line: str) -> None:
        """`%helioai_dev <token>` — unlock unrestricted mode for this session."""
        global _dev_restricted
        from helioai.config import dev_unlock, settings

        cmd = line.strip().lower()
        if cmd == "on":
            if not dev_unlock(settings.dev.token):
                print("Dev token not configured or incorrect. Set HELIOAI_DEV_TOKEN in .env.")
                return
            _dev_restricted = False
            print("Dev mode ON — scope guardrail disabled.")
        elif cmd == "off":
            _dev_restricted = True
            print("Dev mode OFF — helio-only scope guardrail active.")
        else:
            status = "OFF (restricted)" if _dev_restricted else "ON (unrestricted)"
            print(f"Dev mode: {status}. Usage: %helioai_dev on | off")

helioai

helioai(line: str, cell: str) -> None

%%helioai — send a natural-language query to the agent.

Figures render inline; parameter cards and catalog previews render as HTML.

Example

%load_ext helioai.interfaces.jupyter_magic

%%helioai Download ACE IMF for the 2015-03-17 storm, plot Bz and mark the shock arrival.

Source code in helioai/interfaces/jupyter_magic.py

@cell_magic
def helioai(self, line: str, cell: str) -> None:
    """`%%helioai` — send a natural-language query to the agent.

    Figures render inline; parameter cards and catalog previews render as HTML.

    Example:
        %load_ext helioai.interfaces.jupyter_magic

        %%helioai
        Download ACE IMF for the 2015-03-17 storm, plot Bz and mark
        the shock arrival.
    """
    import helioai.tools.setup  # noqa: F401
    from helioai.core.agent_loop import stream_chat
    from helioai.logging_config import setup_logging

    setup_logging("WARNING")

    async def _run():
        llm = _get_llm()
        try:
            async for ev in stream_chat(
                llm, _USER_ID, _SESSION_ID, cell.strip(), restricted=_dev_restricted
            ):
                _render_jupyter_event(ev)
        finally:
            # Must happen inside this loop: the pool is bound to it, and
            # `_run_async` closes the loop the moment this returns.
            await llm.aclose()

    _run_async(_run())

helioai_session

helioai_session(line: str) -> None

%helioai_session [id] — show or switch the active session.

Source code in helioai/interfaces/jupyter_magic.py

@line_magic
def helioai_session(self, line: str) -> None:
    """`%helioai_session [id]` — show or switch the active session."""
    global _SESSION_ID
    parts = line.strip().split(maxsplit=1)
    cmd = parts[0] if parts else ""
    arg = parts[1] if len(parts) > 1 else ""

    if cmd == "reset":
        from helioai.core.session import store

        store.reset(_USER_ID, _SESSION_ID)
        _SESSION_ID = str(uuid.uuid4())
        print(f"Session reset. New id: {_SESSION_ID[:8]}")
    elif cmd == "delete":
        from helioai.core.session import store
        from helioai.workspace import _root

        if not arg:
            print("Usage: %helioai_session delete <session_id_prefix>")
            return
        all_ids = store.all_sessions(_USER_ID)
        matches = [s for s in all_ids if s.startswith(arg)]
        if not matches:
            print(f"No session matching {arg!r}.")
            return
        sid = matches[0]
        wdir = store.get_workspace_dir(_USER_ID, sid)
        store.reset(_USER_ID, sid)
        if wdir:
            import shutil

            ws_path = _root() / wdir
            if ws_path.exists():
                shutil.rmtree(ws_path, ignore_errors=True)
        if sid == _SESSION_ID:
            _SESSION_ID = str(uuid.uuid4())
            print(f"Current session deleted. New id: {_SESSION_ID[:8]}")
        else:
            print(f"Session {sid[:8]} deleted.")
    else:
        print(f"Unknown command: {cmd!r}. Use 'reset' or 'delete <id>'.")

helioai_provider

helioai_provider(line: str) -> None

%helioai_provider [name] — show or switch the LLM provider.

Source code in helioai/interfaces/jupyter_magic.py

@line_magic
def helioai_provider(self, line: str) -> None:
    """`%helioai_provider [name]` — show or switch the LLM provider."""
    provider = line.strip().lower()
    if provider not in ("groq", "gemini", "azure", "opencode", "ollama"):
        print(f"Unknown provider {provider!r}. Use: groq | gemini | azure | opencode | ollama")
        return
    import os

    os.environ["HELIOAI_LLM_PROVIDER"] = provider
    print(f"Provider switched to {provider!r}.")

helioai_history

helioai_history(line: str) -> None

%helioai_history — list recent sessions.

Source code in helioai/interfaces/jupyter_magic.py

@line_magic
def helioai_history(self, line: str) -> None:
    """`%helioai_history` — list recent sessions."""
    from helioai.core.session import store

    summaries = store.list_summaries(_USER_ID)
    if not summaries:
        print("No history found.")
        return
    rows = "".join(
        f"<tr>"
        f"<td><code>{s['session_id'][:8]}</code></td>"
        f"<td>{s['updated_at'][:16].replace('T', ' ')}</td>"
        f"<td style='text-align:center'>{s['n_messages']}</td>"
        f"<td>{s['first_message']}</td>"
        f"</tr>"
        for s in summaries
    )
    display(
        HTML(
            "<table><thead><tr>"
            "<th>Session</th><th>Updated</th><th>Msgs</th><th>First message</th>"
            "</tr></thead><tbody>" + rows + "</tbody></table>"
        )
    )

helioai_profile

helioai_profile(line: str) -> None

%helioai_profile — show or edit the user profile.

Source code in helioai/interfaces/jupyter_magic.py

@line_magic
def helioai_profile(self, line: str) -> None:
    """`%helioai_profile` — show or edit the user profile."""
    from helioai.workspace import user_home

    parts = line.strip().split(maxsplit=1)
    cmd = parts[0] if parts else ""
    arg = parts[1].strip().strip("\"'") if len(parts) > 1 else ""
    p = user_home(_USER_ID) / "profile.md"

    if cmd == "show":
        content = p.read_text(encoding="utf-8").strip() if p.exists() else ""
        display(Markdown(content if content else "_(profil vide)_"))
    elif cmd == "set":
        if not arg:
            print('Usage: %helioai_profile set "your preferences here"')
            return
        p.parent.mkdir(parents=True, exist_ok=True)
        with p.open("a", encoding="utf-8") as f:
            f.write(("\n" if p.stat().st_size > 0 else "") + arg + "\n")
        print(f"Profile updated ({p}).")
    else:
        print('Usage: %helioai_profile show | set "<text>"')

helioai_export

helioai_export(line: str) -> None

%helioai_export — export the session as a standalone notebook.

Source code in helioai/interfaces/jupyter_magic.py

@line_magic
def helioai_export(self, line: str) -> None:
    """`%helioai_export` — export the session as a standalone notebook."""
    from helioai.core.session import store
    from helioai.export import export_session_notebook

    prefix = line.strip()
    session_id = _SESSION_ID
    if prefix:
        matches = [s for s in store.all_sessions(_USER_ID) if s.startswith(prefix)]
        if not matches:
            print(f"No session matching {prefix!r}.")
            return
        session_id = matches[0]
    path = export_session_notebook(_USER_ID, session_id)
    from IPython.display import FileLink

    display(FileLink(str(path), result_html_prefix="📓 Exported notebook: "))

helioai_resume

helioai_resume(line: str) -> None

%helioai_resume — pick a previous session to continue.

Source code in helioai/interfaces/jupyter_magic.py

@line_magic
def helioai_resume(self, line: str) -> None:
    """`%helioai_resume` — pick a previous session to continue."""
    global _SESSION_ID
    prefix = line.strip()
    if not prefix:
        print("Usage: %helioai_resume <session_id>")
        return
    from helioai.core.session import store

    all_ids = store.all_sessions(_USER_ID)
    matches = [s for s in all_ids if s.startswith(prefix)]
    if not matches:
        print(f"No session found matching {prefix!r}.")
        return
    _SESSION_ID = matches[0]
    msgs = store.get_or_create(_USER_ID, _SESSION_ID)
    print(f"Resumed session {_SESSION_ID[:8]} ({len(msgs)} messages).")

helioai_dev

helioai_dev(line: str) -> None

%helioai_dev <token> — unlock unrestricted mode for this session.

Source code in helioai/interfaces/jupyter_magic.py

@line_magic
def helioai_dev(self, line: str) -> None:
    """`%helioai_dev <token>` — unlock unrestricted mode for this session."""
    global _dev_restricted
    from helioai.config import dev_unlock, settings

    cmd = line.strip().lower()
    if cmd == "on":
        if not dev_unlock(settings.dev.token):
            print("Dev token not configured or incorrect. Set HELIOAI_DEV_TOKEN in .env.")
            return
        _dev_restricted = False
        print("Dev mode ON — scope guardrail disabled.")
    elif cmd == "off":
        _dev_restricted = True
        print("Dev mode OFF — helio-only scope guardrail active.")
    else:
        status = "OFF (restricted)" if _dev_restricted else "ON (unrestricted)"
        print(f"Dev mode: {status}. Usage: %helioai_dev on | off")

load_ipython_extension

load_ipython_extension(ipython) -> None

Register the HelioAI magics. Called by %load_ext.

Source code in helioai/interfaces/jupyter_magic.py

def load_ipython_extension(ipython) -> None:
    """Register the HelioAI magics. Called by `%load_ext`."""
    ipython.register_magics(HelioAIMagics)

Web application

helioai.interfaces.web.app

FastAPI web interface for HelioAI.

Single-user, no auth. Streams agent events as SSE. Figures from the sandbox are served via /figure?path=.

require_user async

require_user(x_helio_token: str | None = Header(default=None)) -> str

Resolve the caller's user_id from the X-Helio-Token header.

No users configured (local dev) → single shared user, no auth. Once HELIOAI_USERS is set (deployment), a valid nominative token is required.

Source code in helioai/interfaces/web/app.py

async def require_user(x_helio_token: str | None = Header(default=None)) -> str:
    """Resolve the caller's user_id from the X-Helio-Token header.

    No users configured (local dev) → single shared user, no auth. Once
    HELIOAI_USERS is set (deployment), a valid nominative token is required.
    """
    users = settings.web_auth.users
    if not users:
        return _DEFAULT_USER
    if not x_helio_token or x_helio_token not in users:
        raise HTTPException(status_code=401, detail="Invalid or missing token")
    return users[x_helio_token]

index async

index()

Serve the single-page web UI.

Source code in helioai/interfaces/web/app.py

@app.get("/")
async def index():
    """Serve the single-page web UI."""
    return FileResponse(_STATIC / "index.html")

health async

health()

Liveness probe. Returns {"status": "ok"}.

Source code in helioai/interfaces/web/app.py

@app.get("/health")
async def health():
    """Liveness probe. Returns `{"status": "ok"}`."""
    return {"status": "ok"}

chat_stream async

chat_stream(req: _ChatRequest, x_helio_dev_token: str | None = Header(default=None), user_id: str = Depends(require_user))

Stream one agent turn as Server-Sent Events.

Each agent event — tool calls, results, artifacts, sub-agent activity — is forwarded as it happens, which is what drives the live activity dock.

Source code in helioai/interfaces/web/app.py

@app.post("/chat/stream")
async def chat_stream(
    req: _ChatRequest,
    x_helio_dev_token: str | None = Header(default=None),
    user_id: str = Depends(require_user),
):
    """Stream one agent turn as Server-Sent Events.

    Each agent event — tool calls, results, artifacts, sub-agent activity — is
    forwarded as it happens, which is what drives the live activity dock.
    """
    # Authenticated nominative users are trusted → unrestricted; the legacy dev
    # token still unlocks scope when no users are configured (local dev).
    restricted = not (bool(settings.web_auth.users) or dev_unlock(x_helio_dev_token))

    async def gen():
        llm = None
        try:
            llm = build_llm_client(req.provider)
            async for ev in stream_chat(
                llm, user_id, req.session_id, req.message, restricted=restricted
            ):
                yield f"data: {json.dumps(ev)}\n\n"
        except Exception as e:
            yield f"data: {json.dumps({'event': 'error', 'data': {'message': str(e)}})}\n\n"
        finally:
            # One client per request, so the pool has to be released per request —
            # including when the browser disconnects mid-stream and this generator
            # is closed early.
            if llm is not None:
                await llm.aclose()

    return StreamingResponse(
        gen(),
        media_type="text/event-stream",
        headers={"X-Accel-Buffering": "no", "Cache-Control": "no-cache"},
    )

list_sessions async

list_sessions(user_id: str = Depends(require_user))

List the calling user's sessions, most recent first.

Source code in helioai/interfaces/web/app.py

@app.get("/api/sessions")
async def list_sessions(user_id: str = Depends(require_user)):
    """List the calling user's sessions, most recent first."""
    return store.list_summaries(user_id)

get_session_messages async

get_session_messages(session_id: str, user_id: str = Depends(require_user))

Replay a session: its messages plus any figures and figure reviews.

Source code in helioai/interfaces/web/app.py

@app.get("/api/sessions/{session_id}/messages")
async def get_session_messages(session_id: str, user_id: str = Depends(require_user)):
    """Replay a session: its messages plus any figures and figure reviews."""
    history = store.get_or_create(user_id, session_id)
    out: list[dict] = []
    pending_figures: list[str] = []
    pending_cards: list[dict] = []
    pending_catalogs: list[dict] = []
    pending_code: list[dict] = []
    pending_recipes: list[dict] = []
    for m in history:
        if m.role == "user":
            out.append({"role": "user", "content": m.content})
        elif m.role == "assistant" and m.content:
            entry: dict = {"role": "assistant", "content": m.content}
            if pending_figures:
                entry["figures"] = pending_figures[:]
                pending_figures = []
            if pending_cards:
                entry["cards"] = pending_cards[:]
                pending_cards = []
            if pending_catalogs:
                entry["catalogs"] = pending_catalogs[:]
                pending_catalogs = []
            if pending_code:
                entry["code"] = pending_code[:]
                pending_code = []
            if pending_recipes:
                entry["recipes"] = pending_recipes[:]
                pending_recipes = []
            out.append(entry)
        elif m.role == "tool" and m.content:
            try:
                data = json.loads(m.content)
                if isinstance(data, dict):
                    if data.get("figure_paths"):  # run_python direct
                        pending_figures.extend(data["figure_paths"])
                    for card in data.get(
                        "cards", []
                    ):  # param_card()/document_method() in run_python
                        if not isinstance(card, dict):
                            continue
                        if card.get("kind") == "parameter_card":
                            pending_cards.append(card)
                        elif card.get("kind") == "method_used":
                            pending_recipes.append(
                                {
                                    "kind": "recipe_used",
                                    "name": card.get("name", ""),
                                    "reference": card.get("reference", ""),
                                    "description": card.get("method", ""),
                                }
                            )
                    if data.get("code_path"):  # run_python direct — artifact code
                        pending_code.append(
                            {
                                "kind": "code",
                                "code_path": data["code_path"],
                                "name": Path(data["code_path"]).name,
                                "n_lines": data.get("n_lines"),
                            }
                        )
                    if "metadata" in data and data.get("name") and data.get("code"):  # load_recipe
                        _meta = data.get("metadata") or {}
                        pending_recipes.append(
                            {
                                "kind": "recipe_used",
                                "name": data["name"],
                                "reference": _meta.get("reference", ""),
                                "description": _meta.get("description", ""),
                            }
                        )
                    if data.get("_kind") == "catalog_preview":  # get_catalog
                        pending_catalogs.append(
                            {
                                "kind": "catalog_preview",
                                "catalog_id": data.get("catalog_id"),
                                "name": data.get("name"),
                                "type": data.get("type"),
                                "nb_events_total": data.get("nb_events_total"),
                                "columns": data.get("columns", []),
                                "sample": (data.get("sample") or [])[:5],
                                "survey_start": data.get("survey_start"),
                                "survey_stop": data.get("survey_stop"),
                            }
                        )
                    if data.get("param_id") and "preview" in data:  # get_timeseries direct
                        pending_cards.append(
                            {
                                "kind": "parameter_card",
                                "param_id": data.get("param_id"),
                                "name": data.get("name"),
                                "mission": data.get("mission"),
                                "instrument": data.get("instrument"),
                                "units": data.get("units"),
                                "cadence": data.get("cadence"),
                                "components": data.get("components"),
                                "n_points": data.get("n_points"),
                                "start": data.get("start"),
                                "stop": data.get("stop"),
                            }
                        )
                    for art in data.get("artifacts", []):  # résultat sous-agent
                        if not isinstance(art, dict):
                            continue
                        if art.get("figure_paths"):
                            pending_figures.extend(art["figure_paths"])
                        if art.get("kind") == "parameter_card":
                            pending_cards.append(art)
                        if art.get("kind") == "catalog_preview":
                            pending_catalogs.append(art)
                        if art.get("kind") == "code":
                            pending_code.append(art)
                        if art.get("kind") == "recipe_used":
                            pending_recipes.append(art)
            except (ValueError, TypeError):
                pass
    return {"messages": out}

get_profile async

get_profile(user_id: str = Depends(require_user))

Return the caller's profile markdown.

Source code in helioai/interfaces/web/app.py

@app.get("/api/profile")
async def get_profile(user_id: str = Depends(require_user)):
    """Return the caller's profile markdown."""
    p = _profile_path(user_id)
    content = p.read_text(encoding="utf-8").strip() if p.exists() else ""
    return {"content": content}

put_profile async

put_profile(body: _ProfileBody, user_id: str = Depends(require_user))

Replace the caller's profile markdown.

Source code in helioai/interfaces/web/app.py

@app.put("/api/profile")
async def put_profile(body: _ProfileBody, user_id: str = Depends(require_user)):
    """Replace the caller's profile markdown."""
    p = _profile_path(user_id)
    p.parent.mkdir(parents=True, exist_ok=True)
    p.write_text(body.content, encoding="utf-8")
    return {"ok": True}

delete_session async

delete_session(session_id: str, user_id: str = Depends(require_user))

Delete one of the caller's sessions and its workspace.

Source code in helioai/interfaces/web/app.py

@app.delete("/api/sessions/{session_id}")
async def delete_session(session_id: str, user_id: str = Depends(require_user)):
    """Delete one of the caller's sessions and its workspace."""
    wdir = store.get_workspace_dir(user_id, session_id)
    store.reset(user_id, session_id)
    if wdir:
        # Containment, not trust: the label is persisted data, and a row written by
        # an older build (before session ids were sanitised) would walk this rmtree
        # straight out of the user's home.
        ws_root = (user_home(user_id) / "workspace").resolve()
        ws_path = (ws_root / wdir).resolve()
        if ws_path.is_relative_to(ws_root) and ws_path.exists():
            shutil.rmtree(ws_path, ignore_errors=True)
    return {"deleted": session_id}

export_notebook async

export_notebook(session_id: str, user_id: str = Depends(require_user))

Export a session as a standalone .ipynb and return it.

Source code in helioai/interfaces/web/app.py

@app.get("/api/export")
async def export_notebook(session_id: str, user_id: str = Depends(require_user)):
    """Export a session as a standalone `.ipynb` and return it."""
    from helioai.export import export_session_notebook

    if session_id not in store.all_sessions(user_id):
        raise HTTPException(status_code=404, detail="Unknown session")
    path = export_session_notebook(user_id, session_id)
    return FileResponse(
        path,
        media_type="application/x-ipynb+json",
        filename=path.name,
    )

serve_code async

serve_code(path: str, user_id: str = Depends(require_user))

Return a generated script, rewritten to standalone form.

Ownership is checked against the caller before anything is read, so a path outside the caller's workspace is a 404 rather than a leak.

Source code in helioai/interfaces/web/app.py

@app.get("/code")
async def serve_code(path: str, user_id: str = Depends(require_user)):
    """Return a generated script, rewritten to standalone form.

    Ownership is checked against the caller before anything is read, so a path
    outside the caller's workspace is a 404 rather than a leak.
    """
    path = path.strip()
    if not is_under_workspace(path) or not _owns_path(user_id, path):
        log.warning("code_rejected", path=path, reason="outside workspace or not owner")
        raise HTTPException(status_code=404, detail="Not found")
    p = Path(path).resolve()
    if p.suffix != ".py" or not p.is_file():
        log.warning("code_rejected", path=path, reason="file not found or not .py")
        raise HTTPException(status_code=404, detail="Not found")
    from helioai.datastore import read_manifest
    from helioai.export import to_standalone

    manifest = read_manifest(p.parent)
    standalone = to_standalone(p.read_text(encoding="utf-8"), manifest, with_header=True)
    return PlainTextResponse(standalone)

serve_figure async

serve_figure(path: str, user_id: str = Depends(require_user))

Serve a figure (PNG or PDF) from the caller's workspace.

Source code in helioai/interfaces/web/app.py

@app.get("/figure")
async def serve_figure(path: str, user_id: str = Depends(require_user)):
    """Serve a figure (PNG or PDF) from the caller's workspace."""
    path = path.strip()
    if not is_under_workspace(path) or not _owns_path(user_id, path):
        log.warning("figure_rejected", path=path, reason="outside workspace or not owner")
        raise HTTPException(status_code=404, detail="Not found")
    p = Path(path).resolve()
    media_type = _FIGURE_TYPES.get(p.suffix.lower())
    if media_type is None:
        log.warning("figure_rejected", path=path, reason="unsupported type")
        raise HTTPException(status_code=404, detail="Not found")
    if not p.is_file():
        log.warning("figure_rejected", path=path, reason="file not found")
        raise HTTPException(status_code=404, detail="Not found")
    return FileResponse(p, media_type=media_type)

serve_web

serve_web(host: str = '127.0.0.1', port: int = 7890) -> None

Run the web UI with uvicorn.

Binds to localhost by default. The open-source build ships no authentication and run_python executes model-written code, so do not expose this on a network without putting auth in front of it.

Source code in helioai/interfaces/web/app.py

def serve_web(host: str = "127.0.0.1", port: int = 7890) -> None:
    """Run the web UI with uvicorn.

    Binds to localhost by default. The open-source build ships no authentication
    and `run_python` executes model-written code, so do not expose this on a
    network without putting auth in front of it.
    """
    import uvicorn
    from starlette.middleware.trustedhost import TrustedHostMiddleware

    from helioai.workspace import cleanup_old_runs

    if host in {"127.0.0.1", "localhost", "::1"}:
        # A loopback bind is not a boundary: any web page can resolve its own
        # domain to 127.0.0.1 and reach this server (DNS rebinding). Pinning Host
        # costs nothing here and CORS does not cover it.
        app.add_middleware(TrustedHostMiddleware, allowed_hosts=["localhost", "127.0.0.1"])
    else:
        log.warning("web_exposed_beyond_loopback", host=host)
    cleanup_old_runs()
    uvicorn.run(app, host=host, port=port)

MCP server

helioai.mcp_server

MCP server for HelioAI — exposes all registered tools via stdio or HTTP streamable transport.

Usage

helioai serve # stdio (Claude Desktop / claude CLI) helioai serve --http # HTTP streamable on 127.0.0.1:8765 helioai serve --http --host 0.0.0.0 --port 9000 helioai-mcp # direct entry point (stdio only)

serve_stdio async

serve_stdio() -> None

Run the MCP server over stdio, for clients like Claude Desktop.

Blocks until the client closes the pipe. All 17 registry tools are exposed.

Example

Claude Desktop config: {"command": "helioai-mcp"} — stdio is the default transport, no flags needed.

Source code in helioai/mcp_server.py

async def serve_stdio() -> None:
    """Run the MCP server over stdio, for clients like Claude Desktop.

    Blocks until the client closes the pipe. All 17 registry tools are exposed.

    Example:
        Claude Desktop config: {"command": "helioai-mcp"} — stdio is the default
        transport, no flags needed.
    """
    from mcp.server.stdio import stdio_server

    async with stdio_server() as (read, write):
        await server.run(read, write, _init_options())

build_http_app

build_http_app()

Build the streamable-HTTP ASGI app exposing the MCP server.

Returns a Starlette app mounting the MCP session manager at /mcp, suitable for any ASGI server (serve_http wraps it in uvicorn).

Source code in helioai/mcp_server.py

def build_http_app():
    """Build the streamable-HTTP ASGI app exposing the MCP server.

    Returns a Starlette app mounting the MCP session manager at `/mcp`,
    suitable for any ASGI server (`serve_http` wraps it in uvicorn).
    """
    from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
    from starlette.applications import Starlette
    from starlette.routing import Mount

    manager = StreamableHTTPSessionManager(app=server, json_response=False, stateless=False)

    @contextlib.asynccontextmanager
    async def lifespan(app):
        async with manager.run():
            yield

    return Starlette(routes=[Mount("/mcp", app=manager.handle_request)], lifespan=lifespan)

serve_http

serve_http(host: str, port: int) -> None

Run the MCP server over streamable HTTP.

Parameters:

Name	Type	Description	Default
host	str	Bind address. Anything but loopback logs a warning — run_python would be reachable from that network without authentication.	required
port	int	TCP port.	required

Source code in helioai/mcp_server.py

def serve_http(host: str, port: int) -> None:
    """Run the MCP server over streamable HTTP.

    Args:
        host: Bind address. Anything but loopback logs a warning — `run_python`
            would be reachable from that network without authentication.
        port: TCP port.
    """
    import uvicorn

    uvicorn.run(build_http_app(), host=host, port=port)

main

main() -> None

Entry point for the helioai-mcp command.

Example

helioai-mcp # stdio (Claude Desktop, claude CLI) helioai-mcp --http --port 8765 # streamable HTTP on 127.0.0.1:8765

Source code in helioai/mcp_server.py

def main() -> None:
    """Entry point for the `helioai-mcp` command.

    Example:
        helioai-mcp                        # stdio (Claude Desktop, claude CLI)
        helioai-mcp --http --port 8765     # streamable HTTP on 127.0.0.1:8765
    """
    setup_logging("WARNING")
    args = sys.argv[1:]
    if "--http" in args:
        host = _arg(args, "--host", "127.0.0.1")
        port = int(_arg(args, "--port", "8765"))
        if host not in {"127.0.0.1", "localhost", "::1"}:
            # Over stdio the client owns the process, so exposing run_python is the
            # normal contract. Over HTTP there is no authentication and no scope
            # guardrail: anything that reaches this port runs Python on this host.
            get_logger(__name__).warning(
                "mcp_http_exposed_without_auth",
                host=host,
                port=port,
                detail="every registered tool, run_python included, is reachable unauthenticated",
            )
        serve_http(host, port)
    else:
        asyncio.run(serve_stdio())

Indexer

helioai.indexer

Build the speasy catalog ChromaDB index.

Usage

helioai index # incremental (skip existing) helioai index --rebuild # wipe and rebuild

build_index

build_index(rebuild: bool = False, batch_size: int = 128, verbose: bool = True) -> int

Walk the speasy inventory and index all parameters into ChromaDB.

Backs helioai index and must run once before search_parameters works; the index persists under settings.rag.chroma_dir.

Parameters:

Name	Type	Description	Default
rebuild	bool	Drop and re-create the collection instead of appending.	False
batch_size	int	Documents per ChromaDB insert.	128
verbose	bool	Print per-provider progress to stdout.	True

Returns:

Type	Description
int	Number of parameters indexed (0 when speasy or chromadb is missing).

Example

build_index(rebuild=True) # equivalent to: helioai index --rebuild 82433

Source code in helioai/indexer.py

def build_index(rebuild: bool = False, batch_size: int = 128, verbose: bool = True) -> int:
    """Walk the speasy inventory and index all parameters into ChromaDB.

    Backs `helioai index` and must run once before `search_parameters` works;
    the index persists under `settings.rag.chroma_dir`.

    Args:
        rebuild: Drop and re-create the collection instead of appending.
        batch_size: Documents per ChromaDB insert.
        verbose: Print per-provider progress to stdout.

    Returns:
        Number of parameters indexed (0 when speasy or chromadb is missing).

    Example:
        >>> build_index(rebuild=True)   # equivalent to: helioai index --rebuild
        82433
    """
    try:
        import chromadb
        import speasy as spz
        from sentence_transformers import SentenceTransformer
        from speasy.core.inventory.indexes import SpeasyIndex
    except ImportError as e:
        print(f"[indexer] Missing dependency: {e}")
        print("[indexer] Run: pip install speasy chromadb sentence-transformers")
        return 0

    from helioai.config import settings

    chroma_dir = settings.rag.chroma_dir
    collection_name = settings.rag.collection_name
    embed_model = settings.rag.embed_model

    if rebuild and chroma_dir.exists():
        if verbose:
            print(f"[indexer] wiping {chroma_dir}")
        shutil.rmtree(chroma_dir)

    chroma_dir.mkdir(parents=True, exist_ok=True)

    if verbose:
        print(f"[indexer] loading embedding model {embed_model}…")
    model = SentenceTransformer(embed_model)

    client = chromadb.PersistentClient(path=str(chroma_dir))
    collection = client.get_or_create_collection(
        name=collection_name,
        metadata={"hnsw:space": "cosine"},
    )

    existing_ids: set[str] = set()
    if not rebuild:
        try:
            existing_ids = set(collection.get(include=[])["ids"])
            if verbose and existing_ids:
                print(f"[indexer] {len(existing_ids)} existing entries — skipping")
        except Exception:
            pass

    if verbose:
        print("[indexer] walking speasy inventory…")

    docs: list[dict] = []
    tree = spz.inventories.tree

    for provider_attr, prefix in _PROVIDER_PREFIXES.items():
        provider_node = getattr(tree, provider_attr, None)
        if provider_node is None:
            continue
        before = len(docs)
        _walk(provider_node, prefix, docs, existing_ids, SpeasyIndex)
        if verbose:
            print(f"[indexer]   {prefix}: {len(docs) - before} new params")

    if verbose:
        print(f"[indexer] total new params to index: {len(docs)}")

    if not docs:
        if verbose:
            print("[indexer] up to date — nothing to index")
        return 0

    t0 = time.perf_counter()
    total = 0

    for i in range(0, len(docs), batch_size):
        batch = docs[i : i + batch_size]
        ids = [d["id"] for d in batch]
        texts = [d["text"] for d in batch]
        metas = [d["meta"] for d in batch]

        embeddings = model.encode(
            texts,
            batch_size=batch_size,
            show_progress_bar=False,
            convert_to_numpy=True,
            normalize_embeddings=True,
        ).tolist()

        collection.upsert(ids=ids, embeddings=embeddings, documents=texts, metadatas=metas)
        total += len(ids)
        if verbose:
            print(f"[indexer]   {total}/{len(docs)} indexed…", end="\r", flush=True)

    elapsed = time.perf_counter() - t0
    if verbose:
        print()
        print(f"[indexer] done: {total} params in {elapsed:.1f}s")
        print(f"[indexer] collection total: {collection.count()}")

    # Index catalogs + timetables into a separate collection
    cat_total = _build_catalog_index(model, client, settings, rebuild=rebuild, verbose=verbose)

    return total + cat_total