Install dependencies separately (#1)

Author: samuelcolvin

.github/workflows/ci.yml

@@ -57,6 +57,7 @@ jobs:
       - run: make test
         env:
           COVERAGE_FILE: coverage/.coverage.py${{ matrix.python-version }}
+      - run: uv run mcp-run-python --deps numpy example
       - name: store coverage files
         uses: actions/upload-artifact@v4
         with:
@@ -78,7 +79,7 @@ jobs:
         with:
           enable-cache: true
       - run: uvx coverage combine coverage
-      - run: uvx coverage report --fail-under 90
+      - run: uvx coverage report --fail-under 95
 
   check:
     if: always()

Makefile

@@ -58,7 +58,7 @@ typecheck: typecheck-ts typecheck-py ## Typecheck all code
 
 .PHONY: test
 test: build ## Run tests and collect coverage data
-	uv run coverage run -m pytest
+	uv run coverage run -m pytest -v
 	@uv run coverage report
 
 .PHONY: all

README.md

@@ -32,7 +32,7 @@ To use this server, you must have both Python and [Deno](https://deno.com/) inst
 The server can be run with `deno` installed using `uvx`:
 
 ```bash
-uvx mcp-run-python [stdio|streamable-http|warmup]
+uvx mcp-run-python [-h] [--version] [--port PORT] [--deps DEPS] {stdio,streamable-http,example}
 ```
 
 where:
@@ -44,8 +44,8 @@ where:
   [Streamable HTTP MCP transport](https://modelcontextprotocol.io/specification/2025-06-18/basic/transports#streamable-http) -
   suitable for running the server as an HTTP server to connect locally or remotely. This supports stateful requests, but
   does not require the client to hold a stateful connection like SSE
-- `warmup` will run a minimal Python script to download and cache the Python standard library. This is also useful to
-  check the server is running correctly.
+- `example` will run a minimal Python script using `numpy`, useful for checking that the package is working, for the code
+  to run successfully, you'll need to install `numpy` using `uvx mcp-run-python --deps numpy example`
 
 ## Usage in codes
 
@@ -60,15 +60,15 @@ Then you can use `mcp-run-python` with Pydantic AI:
 ```python
 from pydantic_ai import Agent
 from pydantic_ai.mcp import MCPServerStdio
-from mcp_run_python import deno_args
+from mcp_run_python import deno_args_prepare
 
 import logfire
 
 logfire.configure()
 logfire.instrument_mcp()
 logfire.instrument_pydantic_ai()
 
-server = MCPServerStdio('deno', args=deno_args('stdio'))
+server = MCPServerStdio('deno', args=deno_args_prepare('stdio'))
 agent = Agent('claude-3-5-haiku-latest', toolsets=[server])
 
 
@@ -83,10 +83,75 @@ if __name__ == '__main__':
     asyncio.run(main())
 ```
 
+**Note**: `deno_args_prepare` can take `deps` as a keyword argument to install dependencies.
+As well as returning the args needed to run `mcp_run_python`, `deno_args_prepare` installs the dependencies
+so they can be used by the server.
+
 ## Logging
 
 MCP Run Python supports emitting stdout and stderr from the python execution as [MCP logging messages](https://github.com/modelcontextprotocol/specification/blob/eb4abdf2bb91e0d5afd94510741eadd416982350/docs/specification/draft/server/utilities/logging.md?plain=1).
 
 For logs to be emitted you must set the logging level when connecting to the server. By default, the log level is set to the highest level, `emergency`.
 
 Currently, it's not possible to demonstrate this due to a bug in the Python MCP Client, see [modelcontextprotocol/python-sdk#201](https://github.com/modelcontextprotocol/python-sdk/issues/201#issuecomment-2727663121).
+
+## Dependencies
+
+`mcp_run_python` uses a two step process to install dependencies while avoiding any risk that sandboxed code can
+edit the filesystem.
+
+* `deno` is first run with write permissions to the `node_modules` directory and dependencies are installed, causing wheels to be written to ``
+* `deno` is then run with read-only permissions to the `node_modules` directory to run untrusted code.
+
+Dependencies must be provided when initializing the server so they can be installed in the first step.
+
+Here's an example of manually running code with `mcp-run-python`:
+
+```python
+from mcp import ClientSession, StdioServerParameters
+from mcp.client.stdio import stdio_client
+
+from mcp_run_python import deno_args_prepare
+
+code = """
+import numpy
+a = numpy.array([1, 2, 3])
+print(a)
+a
+"""
+server_params = StdioServerParameters(
+    command='deno',
+    args=deno_args_prepare('stdio', deps=['numpy'])
+)
+
+
+async def main():
+    async with stdio_client(server_params) as (read, write):
+        async with ClientSession(read, write) as session:
+            await session.initialize()
+            tools = await session.list_tools()
+            print(len(tools.tools))
+            #> 1
+            print(repr(tools.tools[0].name))
+            #> 'run_python_code'
+            print(repr(tools.tools[0].inputSchema))
+            """
+            {'type': 'object', 'properties': {'python_code': {'type': 'string', 'description': 'Python code to run'}}, 'required': ['python_code'], 'additionalProperties': False, '$schema': 'http://json-schema.org/draft-07/schema#'}
+            """
+            result = await session.call_tool('run_python_code', {'python_code': code})
+            print(result.content[0].text)
+            """
+            <status>success</status>
+            <dependencies>["numpy"]</dependencies>
+            <output>
+            [1 2 3]
+            </output>
+            <return_value>
+            [
+              1,
+              2,
+              3
+            ]
+            </return_value>
+            """
+```

build/prepare_env.py

@@ -7,29 +7,18 @@
 
 import importlib
 import logging
-import re
 import sys
 import traceback
-from collections.abc import Iterable, Iterator
+from collections.abc import Iterator
 from contextlib import contextmanager
 from dataclasses import dataclass
-from pathlib import Path
-from typing import Any, Literal, TypedDict, cast
+from typing import Any, Literal, cast
 
 import micropip
-import pyodide_js
-import tomllib
-from pyodide.code import find_imports
 
 __all__ = 'prepare_env', 'dump_json'
 
 
-class File(TypedDict):
-    name: str
-    content: str
-    active: bool
-
-
 @dataclass
 class Success:
     dependencies: list[str] | None
@@ -42,22 +31,9 @@ class Error:
     kind: Literal['error'] = 'error'
 
 
-async def prepare_env(files: list[File]) -> Success | Error:
+async def prepare_env(dependencies: list[str] | None) -> Success | Error:
     sys.setrecursionlimit(400)
 
-    cwd = Path.cwd()
-    for file in files:
-        (cwd / file['name']).write_text(file['content'])
-
-    active: File | None = next((f for f in files if f['active']), None)
-
-    dependencies: list[str] | None = None
-    if active:
-        python_code = active['content']
-        dependencies = _find_pep723_dependencies(python_code)
-        if dependencies is None:
-            dependencies = await _find_import_dependencies(python_code)
-
     if dependencies:
         dependencies = _add_extra_dependencies(dependencies)
 
@@ -141,60 +117,3 @@ def _micropip_logging() -> Iterator[str]:
         yield file_name
     finally:
         logger.removeHandler(handler)
-
-
-def _find_pep723_dependencies(code: str) -> list[str] | None:
-    """Extract dependencies from a script with PEP 723 metadata."""
-    metadata = _read_pep723_metadata(code)
-    dependencies: list[str] | None = metadata.get('dependencies')
-    if dependencies is None:
-        return None
-    else:
-        assert isinstance(dependencies, list), 'dependencies must be a list'
-        assert all(isinstance(dep, str) for dep in dependencies), 'dependencies must be a list of strings'
-        return dependencies
-
-
-def _read_pep723_metadata(code: str) -> dict[str, Any]:
-    """Read PEP 723 script metadata.
-
-    Copied from https://packaging.python.org/en/latest/specifications/inline-script-metadata/#reference-implementation
-    """
-    name = 'script'
-    magic_comment_regex = r'(?m)^# /// (?P<type>[a-zA-Z0-9-]+)$\s(?P<content>(^#(| .*)$\s)+)^# ///$'
-    matches = list(filter(lambda m: m.group('type') == name, re.finditer(magic_comment_regex, code)))
-    if len(matches) > 1:
-        raise ValueError(f'Multiple {name} blocks found')
-    elif len(matches) == 1:
-        content = ''.join(
-            line[2:] if line.startswith('# ') else line[1:]
-            for line in matches[0].group('content').splitlines(keepends=True)
-        )
-        return tomllib.loads(content)
-    else:
-        return {}
-
-
-async def _find_import_dependencies(code: str) -> list[str] | None:
-    """Find dependencies in imports."""
-    try:
-        imports: list[str] = find_imports(code)
-    except SyntaxError:
-        return None
-    else:
-        return list(_find_imports_to_install(imports))
-
-
-TO_PACKAGE_NAME: dict[str, str] = pyodide_js._api._import_name_to_package_name.to_py()  # pyright: ignore[reportPrivateUsage]
-
-
-def _find_imports_to_install(imports: list[str]) -> Iterable[str]:
-    """Given a list of module names being imported, return packages that are not installed."""
-    for module in imports:
-        try:
-            importlib.import_module(module)
-        except ModuleNotFoundError:
-            if package_name := TO_PACKAGE_NAME.get(module):
-                yield package_name
-            elif '.' not in module:
-                yield module

build/stubs/pyodide/code.pyi

@@ -0,0 +1,30 @@
+from mcp import ClientSession, StdioServerParameters
+from mcp.client.stdio import stdio_client
+
+from mcp_run_python import deno_args_prepare
+
+code = """
+import numpy
+a = numpy.array([1, 2, 3])
+print(a)
+a
+"""
+server_params = StdioServerParameters(command='deno', args=deno_args_prepare('stdio', deps=['numpy']))
+
+
+async def main():
+    async with stdio_client(server_params) as (read, write):
+        async with ClientSession(read, write) as session:
+            await session.initialize()
+            tools = await session.list_tools()
+            print(len(tools.tools))
+            print(repr(tools.tools[0].name))
+            print(repr(tools.tools[0].inputSchema))
+            result = await session.call_tool('run_python_code', {'python_code': code})
+            print(result.content[0].text)
+
+
+if __name__ == '__main__':
+    import asyncio
+
+    asyncio.run(main())

build/stubs/pyodide_js.pyi

@@ -13,13 +13,13 @@
 from pydantic_ai import Agent
 from pydantic_ai.mcp import MCPServerStdio
 
-from mcp_run_python import deno_args
+from mcp_run_python import deno_args_prepare
 
 logfire.configure()
 logfire.instrument_mcp()
 logfire.instrument_pydantic_ai()
 
-server = MCPServerStdio('deno', args=deno_args('stdio'))
+server = MCPServerStdio('deno', args=deno_args_prepare('stdio'))
 agent_with_python = Agent('claude-3-5-haiku-latest', toolsets=[server])
 
 

examples/direct.py

@@ -2,7 +2,7 @@
 
 from importlib.metadata import version as _metadata_version
 
-from .main import deno_args, run_deno_server
+from .main import deno_args_prepare, deno_run_server
 
 __version__ = _metadata_version('mcp_run_python')
-__all__ = '__version__', 'deno_args', 'run_deno_server'
+__all__ = '__version__', 'deno_args_prepare', 'deno_run_server'

pydantic_ai_example.py examples/pydantic_ai_ex.pypydantic_ai_example.py renamed to examples/pydantic_ai_ex.py

@@ -1,19 +1,14 @@
 from __future__ import annotations as _annotations
 
 import argparse
-import sys
 from collections.abc import Sequence
 
 from . import __version__
-from .main import run_deno_server
+from .main import deno_run_server
 
 
-def cli() -> int:  # pragma: no cover
+def cli(args_list: Sequence[str] | None = None):  # pragma: no cover
     """Run the CLI."""
-    sys.exit(cli_logic())
-
-
-def cli_logic(args_list: Sequence[str] | None = None) -> int:
     parser = argparse.ArgumentParser(
         prog='mcp-run-python',
         description=f'mcp-run-python CLI v{__version__}\n\nMCP server for running untrusted Python code.\n',
@@ -22,16 +17,19 @@ def cli_logic(args_list: Sequence[str] | None = None) -> int:
     parser.add_argument('--version', action='store_true', help='Show version and exit')
 
     parser.add_argument('--port', type=int, help='Port to run the server on, default 3001.')
+    parser.add_argument('--deps', help='Comma separated list of dependencies to install')
     parser.add_argument(
         'mode',
-        choices=['stdio', 'streamable-http', 'warmup'],
+        choices=['stdio', 'streamable-http', 'example'],
+        nargs='?',
         help='Mode to run the server in.',
     )
 
     args = parser.parse_args(args_list)
     if args.version:
         print(f'mcp-run-python {__version__}')
-        return 0
+    elif args.mode:
+        deps: list[str] = args.deps.split(',') if args.deps else []
+        deno_run_server(args.mode.replace('-', '_'), port=args.port, deps=deps, install_log_handler=print)
     else:
-        run_deno_server(args.mode.replace('-', '_'), port=args.port)
-        return 0
+        parser.error('Mode is required')