Class AsyncCodeModeContext

marimo/_code_mode/_context.py:677–1857

Async programmatic control of a running marimo notebook. Use as an async context manager — mutations are queued during the block and applied atomically on exit:: async with cm.get_context() as ctx: ctx.create_cell("x = 1") ctx.edit_cell("my_cell", code="x =

@helpable
class AsyncCodeModeContext:
    """Async programmatic control of a running marimo notebook.

    Use as an async context manager — mutations are queued during the
    block and applied atomically on exit::

        async with cm.get_context() as ctx:
            ctx.create_cell("x = 1")
            ctx.edit_cell("my_cell", code="x = 42")
            ctx.delete_cell("old_cell")

    Read cells via `ctx.cells[key]` where *key* is an integer index,
    cell ID string, or cell name.
    """

    def __init__(
        self,
        kernel: Kernel,
        cell_manager: CellManager | None = None,
        *,
        skip_validation: bool = False,
        skip_staleness_check: bool = False,
    ) -> None:
        from marimo._messaging.notebook.document import get_current_document

        document = get_current_document()
        if document is None:
            raise RuntimeError(
                "NotebookDocument not available — code_mode must be invoked "
                "via the /api/execute endpoint which sets the document "
                "context variable"
            )
        self._kernel = kernel
        self._document = document
        # Output snapshot is optional — callers that don't pass one
        # (e.g. the MCP code server) get ``cell.output is None`` and
        # ``cell.console_outputs == []`` for every cell, same as cells
        # that genuinely produced no output.
        self._outputs: CellOutputs | None = get_current_outputs()
        self._cell_manager = cell_manager
        self._skip_validation = skip_validation
        self._skip_staleness_check = skip_staleness_check
        self._ops: list[_Op] = []
        # Track cell IDs added during this batch so subsequent ops
        # can reference them before they exist in the graph.
        self._pending_adds: dict[CellId_t, _AddOp] = {}
        # ID generator for new cells — seed=None uses OS entropy so we
        # never replay a prior session's ID sequence.  seen_ids from the
        # document prevents collisions with cells already on disk.
        self._id_generator = CellIdGenerator(seed=None)
        self._id_generator.seen_ids = set(document.cell_ids)
        self._packages = Packages(self)
        self._ui_updates: list[tuple[UIElementId, Any]] = []
        self._cells_to_run: set[CellId_t] = set()
        self._entered = False
        self._screenshot_session: _ScreenshotSession | None = None

    def _require_entered(self) -> None: