Add ability to disable networking (#5)

Author: dmontagu

mcp_run_python/_cli.py

@@ -23,6 +23,9 @@ def cli_logic(args_list: Sequence[str] | None = None) -> int:
 
     parser.add_argument('--port', type=int, help='Port to run the server on, default 3001.')
     parser.add_argument('--deps', '--dependencies', help='Comma separated list of dependencies to install')
+    parser.add_argument(
+        '--disable-networking', action='store_true', help='Disable networking during execution of python code'
+    )
     parser.add_argument('--verbose', action='store_true', help='Enable verbose logging')
     parser.add_argument('--version', action='store_true', help='Show version and exit')
     parser.add_argument(
@@ -46,6 +49,7 @@ def cli_logic(args_list: Sequence[str] | None = None) -> int:
         deps: list[str] = args.deps.split(',') if args.deps else []
         return_code = run_mcp_server(
             args.mode.replace('-', '_'),
+            allow_networking=not args.disable_networking,
             http_port=args.port,
             dependencies=deps,
             deps_log_handler=deps_log_handler,

mcp_run_python/code_sandbox.py

@@ -43,6 +43,7 @@ async def code_sandbox(
     dependencies: list[str] | None = None,
     log_handler: LogHandler | None = None,
     logging_level: mcp_types.LoggingLevel | None = None,
+    allow_networking: bool = True,
 ) -> AsyncIterator['CodeSandbox']:
     """Run code in a secure sandbox.
 
@@ -51,9 +52,14 @@ async def code_sandbox(
         log_handler: A callback function to handle print statements when code is running.
         logging_level: The logging level to use for the print handler, defaults to `info` if `log_handler` is provided.
         deps_log_handler: A callback function to run on log statements during initial install of dependencies.
+        allow_networking: Whether to allow networking or not while executing python code.
     """
     async with async_prepare_deno_env(
-        'stdio', dependencies=dependencies, deps_log_handler=log_handler, return_mode='json'
+        'stdio',
+        dependencies=dependencies,
+        deps_log_handler=log_handler,
+        return_mode='json',
+        allow_networking=allow_networking,
     ) as deno_env:
         server_params = StdioServerParameters(command='deno', args=deno_env.args, cwd=deno_env.cwd)
 

mcp_run_python/main.py

@@ -25,6 +25,7 @@ def run_mcp_server(
     dependencies: list[str] | None = None,
     return_mode: Literal['json', 'xml'] = 'xml',
     deps_log_handler: LogHandler | None = None,
+    allow_networking: bool = True,
 ) -> int:
     """Install dependencies then run the mcp-run-python server.
 
@@ -34,9 +35,15 @@ def run_mcp_server(
         dependencies: The dependencies to install.
         return_mode: The mode to return tool results in.
         deps_log_handler: Optional function to receive logs emitted while installing dependencies.
+        allow_networking: Whether to allow networking when running provided python code.
     """
     with prepare_deno_env(
-        mode, dependencies=dependencies, http_port=http_port, return_mode=return_mode, deps_log_handler=deps_log_handler
+        mode,
+        dependencies=dependencies,
+        http_port=http_port,
+        return_mode=return_mode,
+        deps_log_handler=deps_log_handler,
+        allow_networking=allow_networking,
     ) as env:
         if mode == 'streamable_http':
             logger.info('Running mcp-run-python via %s on port %d...', mode, http_port)
@@ -66,6 +73,7 @@ def prepare_deno_env(
     dependencies: list[str] | None = None,
     return_mode: Literal['json', 'xml'] = 'xml',
     deps_log_handler: LogHandler | None = None,
+    allow_networking: bool = True,
 ) -> Iterator[DenoEnv]:
     """Prepare the deno environment for running the mcp-run-python server with Deno.
 
@@ -79,6 +87,8 @@ def prepare_deno_env(
         dependencies: The dependencies to install.
         return_mode: The mode to return tool results in.
         deps_log_handler: Optional function to receive logs emitted while installing dependencies.
+        allow_networking: Whether the prepared DenoEnv should allow networking when running code.
+            Note that we always allow networking during environment initialization to install dependencies.
 
     Returns:
         Yields the deno environment details.
@@ -105,7 +115,13 @@ def prepare_deno_env(
         if p.returncode != 0:
             raise RuntimeError(f'`deno run ...` returned a non-zero exit code {p.returncode}: {"".join(stdout)}')
 
-        args = _deno_run_args(mode, http_port=http_port, dependencies=dependencies, return_mode=return_mode)
+        args = _deno_run_args(
+            mode,
+            http_port=http_port,
+            dependencies=dependencies,
+            return_mode=return_mode,
+            allow_networking=allow_networking,
+        )
         yield DenoEnv(cwd, args)
 
     finally:
@@ -120,6 +136,7 @@ async def async_prepare_deno_env(
     dependencies: list[str] | None = None,
     return_mode: Literal['json', 'xml'] = 'xml',
     deps_log_handler: LogHandler | None = None,
+    allow_networking: bool = True,
 ) -> AsyncIterator[DenoEnv]:
     """Async variant of `prepare_deno_env`."""
     ct = await _asyncify(
@@ -129,6 +146,7 @@ async def async_prepare_deno_env(
         dependencies=dependencies,
         return_mode=return_mode,
         deps_log_handler=deps_log_handler,
+        allow_networking=allow_networking,
     )
     try:
         yield await _asyncify(ct.__enter__)
@@ -157,10 +175,12 @@ def _deno_run_args(
     http_port: int | None = None,
     dependencies: list[str] | None = None,
     return_mode: Literal['json', 'xml'] = 'xml',
+    allow_networking: bool = True,
 ) -> list[str]:
-    args = [
-        'run',
-        '--allow-net',
+    args = ['run']
+    if allow_networking:
+        args += ['--allow-net']
+    args += [
         '--allow-read=./node_modules',
         '--node-modules-dir=auto',
         'src/main.ts',

tests/test_sandbox.py

@@ -109,3 +109,18 @@ def log_handler(level: str, message: str):
         ),
     )
     assert [(level, msg) for level, msg in logs if level == 'info'][-1] == snapshot(('info', 'hello 123'))
+
+
+async def test_disallow_networking():
+    code = """
+import httpx
+async with httpx.AsyncClient() as client:
+    await client.get('http://localhost')
+"""
+    async with code_sandbox(dependencies=['httpx'], allow_networking=False) as sandbox:
+        result = await sandbox.eval(code)
+
+    assert 'error' in result
+    assert result['error'].strip().splitlines()[-1] == snapshot(
+        'httpx.ConnectError: Requires net access to "localhost:80", run again with the --allow-net flag'
+    )