Minimal interactive REPL for the public SageCell server.

README.md

SageCell REPL

Install dependencies with python3 -m pip install requests websocket-client, put sagecell.py on your PATH, make it executable with chmod +x sagecell.py, then run sagecell.py.

sagecell.py

#!/usr/bin/env python3
import json
import os
import readline
import sys
import threading
import warnings
from uuid import uuid4

warnings.filterwarnings("ignore", message="urllib3 v2 only supports OpenSSL.*")

import requests
import websocket


class SageCell:
    def __init__(self, url="https://sagecell.sagemath.org", timeout=30):
        self.url = url.rstrip("/") + "/"
        self.timeout = timeout
        self.ws = None
        self.stop = threading.Event()
        self.connect()
        threading.Thread(target=self._heartbeat, daemon=True).start()

    def _heartbeat(self):
        while not self.stop.wait(10):
            try:
                self.ws.ping()
            except Exception:
                pass

    def connect(self):
        response = requests.post(self.url + "kernel", data={"accepted_tos": "true"}, headers={"Accept": "application/json"}, timeout=self.timeout)
        response.raise_for_status()
        kernel = response.json()
        self.ws = websocket.create_connection(f"{kernel['ws_url']}kernel/{kernel['id']}/channels", timeout=self.timeout, header={"Jupyter-Kernel-ID": kernel["id"]})

    def execute(self, code):
        try:
            self._execute(code)
        except (websocket.WebSocketException, OSError):
            self.disconnect()
            self.connect()
            self._execute(code)

    def _execute(self, code):
        msg_id = str(uuid4())
        message = {
            "channel": "shell",
            "header": {"msg_type": "execute_request", "msg_id": msg_id, "username": "", "session": str(uuid4()), "version": "5.3"},
            "parent_header": {},
            "metadata": {},
            "content": {"code": code, "silent": False, "store_history": True, "user_expressions": {}, "allow_stdin": False, "stop_on_error": True},
        }
        self.ws.send(json.dumps(message))
        replied = idle = False
        while not (replied and idle):
            msg = json.loads(self.ws.recv())
            if msg.get("parent_header", {}).get("msg_id") != msg_id:
                continue
            kind = msg.get("header", {}).get("msg_type")
            content = msg.get("content", {})
            if msg.get("channel") == "shell" and kind == "execute_reply":
                replied = True
            elif kind == "status" and content.get("execution_state") == "idle":
                idle = True
            elif kind == "stream":
                print(content.get("text", ""), end="")
            elif kind in ("execute_result", "display_data"):
                value = content.get("data", {}).get("text/plain")
                if value is not None:
                    print(value)
            elif kind == "error":
                trace = content.get("traceback")
                print("\n".join(trace) if trace else f"{content.get('ename', 'Error')}: {content.get('evalue', '')}", file=sys.stderr)

    def disconnect(self):
        if self.ws:
            try:
                self.ws.close()
            except Exception:
                pass

    def close(self):
        self.stop.set()
        self.disconnect()


def main():
    url = sys.argv[1] if len(sys.argv) > 1 else "https://sagecell.sagemath.org"
    history = os.path.expanduser("~/.sagecell_history")
    try:
        readline.read_history_file(history)
    except OSError:
        pass
    readline.set_history_length(1000)
    try:
        cell = SageCell(url)
    except Exception as error:
        raise SystemExit(f"connection failed: {error}")
    print(f"SageCell REPL · {url} · exit with quit, exit, or Ctrl-D")
    try:
        while True:
            try:
                line = input("sage> ")
                if line.strip() in ("quit", "exit"):
                    break
                lines = [line]
                if line.rstrip().endswith(":") or line.rstrip().endswith("\\"):
                    while True:
                        more = input(".... ")
                        if not more:
                            break
                        lines.append(more)
                if any(part.strip() for part in lines):
                    cell.execute("\n".join(lines))
            except KeyboardInterrupt:
                print("\nKeyboardInterrupt")
            except EOFError:
                print()
                break
            except Exception as error:
                print(f"error: {error}", file=sys.stderr)
    finally:
        cell.close()
        try:
            readline.write_history_file(history)
        except OSError:
            pass


if __name__ == "__main__":
    main()