MCP integration of Repl into Claude Desktop

misc/repl-mcp-demo/.gitignore

@@ -0,0 +1,11 @@
+# Python-generated files
+__pycache__/
+*.py[oc]
+build/
+dist/
+wheels/
+*.egg-info
+
+# Virtual environments
+.venv
+.env

misc/repl-mcp-demo/.python-version

@@ -0,0 +1 @@
+3.11

misc/repl-mcp-demo/README.md

@@ -0,0 +1,78 @@
+# Modal REPL MCP Integration 
+
+
+This example shows how to set up a Python REPL built off Modal sandboxes that can be connected to Claude via MCP.
+
+
+## Setup
+
+In your root directory, run
+```bash
+git clone https://github.com/modal-labs/modal-examples.git
+cd modal-examples/misc/repl-mcp-demo
+uv sync
+touch .env
+```
+
+Then, open the `.env` file and paste in the following line: 
+```txt
+SNAPSHOT_ID_FILE_PATH=~/.modal_repl_snapshot.txt
+```
+
+Be sure to setup your Modal account if you haven't yet with `modal setup`.
+
+If you do not have [Claude Desktop](https://claude.ai/download) installed, please install it. 
+
+Create the file `~/Library/Application Support/Claude/claude_desktop_config.json` if it does not yet exist.
+
+Add the following MCP server to the `claude_desktop_config.json` file:
+
+
+```json
+...
+"modalPythonREPL": {
+      "command": "uv",
+      "args": [
+        "--directory",
+        "{YourUserRootDirectory}/modal-examples/misc/repl-mcp-demo",
+        "run",
+        "main.py"
+      ]
+    }
+...
+
+```
+
+For more information on how to configure this file, see the [MCP Docs](https://modelcontextprotocol.info/docs/quickstart/user/).
+
+
+
+## Using the MCP
+
+Open your Claude desktop app and ensure `modalPythonREPL` is toggled on in the "Search and Tools menu:
+
+![Claude Desktop Menu](./README_utils/claude_menu.png)
+
+
+Claude will have access to 4 tools. These are 
+
+- **`create_repl`**: This creates a new repl with the specified timeout and packages.
+- **`exec_cmd`**: Executes a command in the current sandbox. Will return an error if no sandbox has been created or is currently active.
+- **`end_repl_and_save_snapshot`**: Shuts down the current Modal REPL and saves a snapshot of it. The ID of the snapshot is saved to the `SNAPSHOT_ID_FILE_PATH` environment variable.
+- **`get_repl_from_snapshot`**: Retrives the snapshot ID from the `SNAPSHOT_ID_FILE_PATH` environment variable and restores a repl with the snapshotted memory intact.
+
+
+
+Go ahead and try prompting Claude with the activated tool! You can check your Modal account dashboard to track when the REPL's backend sandboxes are active or terminated.
+
+Here's an example snippet of a response on an analysis of airline safety performance on a public dataset:
+
+![Example Response](./README_utils/example_response.png).
+
+
+In the Modal Repl, Claude was able to execute a series of complex steps leveraging both the `pandas` and `requests` libraries to analyze airline incident rates.
+
+
+Here is another example of a response of Claude using the REPL environment with `pandas` and `requests` to do a keyword analysis on shopping sites.
+
+![Example Response 2](./README_utils/shopping_keyword.png).

misc/repl-mcp-demo/main.py

@@ -0,0 +1,80 @@
+import contextlib
+import os
+from typing import List, Optional
+
+import dotenv
+from mcp.server.fastmcp import FastMCP
+from repl import CommandResponse, Repl
+
+dotenv.load_dotenv()
+
+
+"""
+This file specifies the MCP server and its tool with the Claude Desktop app.
+It is automatically ran when the Claude Desktop app is open provided one appropriately
+configures their `claude_desktop_config.json` file as specified in the README.
+"""
+
+sessionRepl: Optional[Repl] = None
+snapshot_id_store_file = os.path.expanduser(os.getenv("SNAPSHOT_ID_FILE_PATH"))
+
+
+mcp = FastMCP("modalrepl")
+
+
+# This tool creates a new repl with the specified timeout and packages.
+@mcp.tool()
+async def create_repl(timeout: int = 600, packages: List[str] = []) -> None:
+    # default timeout is 10 minute
+    try:
+        packages.extend(["fastapi", "uvicorn", "pydantic"])
+        with (
+            contextlib.redirect_stdout(open(os.devnull, "w")),
+            contextlib.redirect_stderr(open(os.devnull, "w")),
+        ):
+            repl = await Repl.create(packages=packages, timeout=timeout)
+        global sessionRepl
+        sessionRepl = repl
+    except Exception as exc:
+        raise RuntimeError(f"Error creating REPL. {exc}")
+
+
+# This tool executes a command in the current repl.
+@mcp.tool()
+async def exec_cmd(command: str) -> CommandResponse:
+    try:
+        if sessionRepl is None:
+            raise RuntimeError("REPL not created")
+        commands = Repl.parse_command(command)
+        res = await sessionRepl.run(commands)
+        return res
+    except Exception as exc:
+        raise RuntimeError(f"Error executing command: {exc}")
+
+
+# This tool restores a repl from a snapshot.
+@mcp.tool()
+async def get_repl_from_snapshot() -> None:
+    try:
+        with open(snapshot_id_store_file, "r") as f:
+            snapshot_id = f.read()
+        repl = await Repl.from_snapshot(snapshot_id)
+        global sessionRepl
+        sessionRepl = repl
+    except Exception as exc:
+        raise RuntimeError(f"Error getting REPL from snapshot: {exc}")
+
+
+@mcp.tool()  # This tool saves the snapshot id to a file
+def end_repl_and_save_snapshot():
+    try:
+        if sessionRepl:
+            snapshot_id = sessionRepl.kill()
+            with open(snapshot_id_store_file, "w") as f:
+                f.write(snapshot_id)
+    except Exception as exc:
+        raise RuntimeError(f"Error shutting down REPL: {exc}")
+
+
+if __name__ == "__main__":
+    mcp.run()

misc/repl-mcp-demo/pyproject.toml

@@ -0,0 +1,14 @@
+[project]
+name = "repl-mcp-demo"
+version = "0.1.0"
+description = "Add your description here"
+readme = "README.md"
+requires-python = ">=3.10"
+dependencies = [
+    "asyncio>=4.0.0",
+    "httpx>=0.28.1",
+    "mcp[cli]>=1.14.0",
+    "modal>=1.1.4",
+    "requests>=2.32.5",
+    "typing>=3.10.0.0",
+]

misc/repl-mcp-demo/repl.py

@@ -0,0 +1,155 @@
+import ast
+import os
+import uuid
+from typing import List, Literal, Optional, Tuple
+
+import httpx
+from modal.app import App
+from modal.image import Image
+from modal.output import enable_output
+from modal.sandbox import Sandbox
+from modal.snapshot import SandboxSnapshot
+from pydantic import BaseModel
+
+"""
+This file specifies the Repl class and it's interface with sandboxes via HTTP requests.
+"""
+
+
+class CommandResponse(BaseModel):
+    output: Optional[str]
+    stdout: Optional[str]
+    error: Optional[str]
+
+
+class Repl:
+    def __init__(self, sandbox: Sandbox, sb_url: str, id: Optional[str] = None):
+        self.sb = sandbox
+        self.sb_url = sb_url
+        self.id = id or str(uuid.uuid4())
+
+    @staticmethod  # Uses AST parsing to distinguish between exec and eval commands
+    def parse_command(code: str) -> List[Tuple[str, Literal["exec", "eval"]]]:
+        try:
+            tree = ast.parse(code, mode="exec")
+            if (
+                tree.body and len(tree.body) > 0 and isinstance(tree.body[-1], ast.Expr)
+            ):  # ast.Expr should be eval()'d
+                last_expr = tree.body[-1]
+                lines = code.splitlines(keepends=True)
+                start_line = getattr(last_expr, "lineno", None)
+                start_col = getattr(last_expr, "col_offset", None)
+                end_line = getattr(last_expr, "end_lineno", None)
+                end_col = getattr(last_expr, "end_col_offset", None)
+                # print(start_line, start_col, end_line, end_col)
+                if (
+                    end_line is None
+                    or end_col is None
+                    or start_line is None
+                    or start_col is None
+                ):
+                    return [(code, "exec")]
+                start_line -= 1
+                end_line -= 1  # ast parser returns 1-indexed lines.our list of strings is 0-indexed
+                prefix_parts = []
+                if start_line > 0:
+                    prefix_parts.append("".join(lines[:start_line]))
+                prefix_parts.append(lines[start_line][:start_col])
+                prefix_code = "".join(prefix_parts)
+                # puts everything before last expression into one str. this is all exec()'d
+                last_expr_parts = []
+                if start_line == end_line:
+                    last_expr_parts.append(lines[start_line][start_col:end_col])
+                else:
+                    last_expr_parts.append(lines[start_line][start_col:])
+                    if end_line - start_line > 1:
+                        last_expr_parts.append(
+                            "\n".join(lines[start_line + 1 : end_line])
+                        )
+                    last_expr_parts.append(lines[end_line][:end_col])
+                last_expr_code = "".join(last_expr_parts)
+
+                commands = []
+                if prefix_code.strip():
+                    commands.append((prefix_code, "exec"))
+                commands.append((last_expr_code, "eval"))
+            else:
+                commands = [(code, "exec")]  # whole thing exec()'d
+            returnCommands = []
+            for cmd in commands:
+                if cmd[0].strip():
+                    returnCommands.append(cmd)
+            return returnCommands
+        except Exception as e:
+            print(str(e))
+            return []
+
+    @staticmethod  # Creates a new sandbox and starts the repl server
+    async def create(
+        python_version: str = "3.13",
+        port: int = 8000,
+        packages: List[str] = [],
+        timeout: int = 600,
+    ) -> "Repl":
+        try:
+            image = Image.debian_slim(python_version=python_version)
+            image = image.pip_install(*packages)
+            repl_server_path = os.path.join(os.path.dirname(__file__), "repl_server.py")
+            image = image.add_local_file(
+                local_path=repl_server_path, remote_path="/root/repl_server.py"
+            )
+            app = App.lookup(name="repl", create_if_missing=True)
+            with enable_output():
+                start_cmd = ["bash", "-c", "cd /root && python repl_server.py"]
+                sb = await Sandbox.create.aio(
+                    *start_cmd,
+                    app=app,
+                    image=image,
+                    encrypted_ports=[port],
+                    _experimental_enable_snapshot=True,
+                    timeout=timeout,
+                )
+                sb_url = (await sb.tunnels.aio())[port].url
+            return Repl(sb, sb_url)
+        except Exception as e:
+            raise Exception(f"Error creating REPL: {e}")
+
+    @staticmethod  # Restores a repl from a snapshot
+    async def from_snapshot(snapshot_id: str, id: Optional[str] = None) -> "Repl":
+        try:
+            snapshot = await SandboxSnapshot.from_id.aio(snapshot_id)
+            sb = await Sandbox._experimental_from_snapshot.aio(snapshot)
+            sb_url = (await sb.tunnels.aio())[8000].url
+            return Repl(sb, sb_url, id)
+        except Exception as e:
+            raise Exception(f"Error getting REPL from snapshot: {e}")
+
+    async def run(  # Runs a list of commands in the repl
+        self, commands: List[Tuple[str, Literal["exec", "eval"]]]
+    ) -> CommandResponse:
+        try:
+            async with httpx.AsyncClient() as client:
+                repl_output = await client.post(self.sb_url, json={"code": commands})
+                if repl_output.status_code != 200:
+                    return CommandResponse(
+                        output=None, stdout=None, error=repl_output.json()["detail"]
+                    )
+                output = repl_output.json()["result"]
+                stdout = repl_output.json()["stdout"]
+                stdout_lines = stdout.splitlines()
+                stdout_lines = [
+                    line for line in stdout_lines if not line.startswith("INFO:")
+                ]  # bad sol to ignore uvicorn logs
+                return CommandResponse(
+                    output=output, stdout="\n".join(stdout_lines), error=None
+                )
+
+        except Exception as e:
+            raise Exception(f"Error running commands: {e}")
+
+    def kill(self) -> str:  # Kills the repl and returns the snapshot id
+        if self.sb:
+            snapshot = self.sb._experimental_snapshot()
+            self.sb.terminate()
+            return snapshot.object_id
+        raise ValueError("repl not found")