feat(box/mcp): integrate MCP stdio with Box sandbox — auto-isolation, dep install, security

## Summary

  When Podman/Docker is available, all stdio-mode MCP servers now automatically
  run inside Box containers with dependency installation, path rewriting, and
  lifecycle management. When no container runtime exists, LangBot starts normally
  and stdio MCP falls back to host-direct execution.

  ## What changed

  ### MCP stdio → Box integration (mcp.py)
  - Add `MCPServerBoxConfig` pydantic model for structured box configuration
    with validation and defaults (network, host_path_mode, timeouts, resources)
  - Auto-infer `host_path` from command/args with venv detection: recognizes
    `.venv/bin/python` patterns and walks up to the project root
  - Rewrite host paths to container `/workspace` paths transparently
  - Replace venv python commands with container-native `python`
  - Auto-detect `pyproject.toml`/`setup.py`/`requirements.txt` and run
    `pip install` inside the container before starting the MCP server
  - Copy project to `/tmp` before install to handle read-only mounts
  - Add retry with exponential backoff (3 retries, 2s/4s/8s delays)
  - Add Box managed process health monitoring (poll every 5s)
  - Fix session leak: `_cleanup_box_stdio_session()` now runs in `finally`
    block of `_lifecycle_loop`, covering all exit paths
  - Fix retry logic: `_ready_event` is only set after all retries exhaust
    or on success, not on first failure
  - Enhance `get_runtime_info_dict()` with `box_session_id` and `box_enabled`

  ### Box security (security.py — new)
  - `validate_sandbox_security()` blocks dangerous host paths:
    `/etc`, `/proc`, `/sys`, `/dev`, `/root`, `/boot`, `/run`,
    docker.sock, podman socket
  - Called at the start of `CLISandboxBackend.start_session()`

  ### Box models (models.py)
  - Add `BoxHostMountMode.NONE` — skips volume mount entirely
  - Adjust `validate_host_mount_consistency` to allow arbitrary workdir
    when `host_path_mode=NONE`

  ### Box backend (backend.py)
  - Add `validate_sandbox_security()` call in `start_session()`
  - Add `langbot.box.config_hash` label on containers for drift detection
  - Handle `BoxHostMountMode.NONE` — skip `-v` mount arg
  - Add `cleanup_orphaned_containers()` to base class (no-op default) and
    CLI implementation (single batched `rm -f` command)

  ### Box runtime (runtime.py)
  - Call `cleanup_orphaned_containers()` during `initialize()` to remove
    lingering containers from previous runs

  ### Box service (service.py)
  - Graceful degradation: `initialize()` catches runtime errors and sets
    `available=False` instead of crashing LangBot startup
  - Add `available` property and guard on `execute_sandbox_tool()`
  - Add `skip_host_mount_validation` parameter to `build_spec()` and
    `create_session()` — MCP paths are admin-configured and trusted,
    bypassing `allowed_host_mount_roots` restrictions meant for
    LLM-generated sandbox_exec commands

  ### Default behavior
  - stdio MCP servers automatically use Box when `box_service.available`
    is True (Podman/Docker detected); no explicit `box` config needed
  - When no container runtime exists, falls back to host-direct stdio
  - MCP Box defaults: `network=on` (for pip install), `read_only_rootfs=false`
    (for site-packages), `host_path_mode=ro`, `startup_timeout=120s`

  ### Tests
  - `test_box_security.py`: blocked paths, safe paths, subpath rejection
  - `test_mcp_box_integration.py`: config model, path rewriting, venv
    unwrap, host_path inference, payload building, runtime info, box
    availability check
  - `test_box_service.py`: `BoxHostMountMode.NONE` validation tests
This commit is contained in:
youhuanghe
2026-03-21 04:34:08 +00:00
committed by WangCham
parent c802dc8029
commit e8aa7b2e6d
13 changed files with 1543 additions and 41 deletions

View File

@@ -0,0 +1,103 @@
from __future__ import annotations
import asyncio
import datetime as dt
from unittest.mock import Mock
import pytest
from langbot.pkg.box.backend import BaseSandboxBackend
from langbot.pkg.box.models import BoxManagedProcessSpec, BoxManagedProcessStatus, BoxSessionInfo, BoxSpec
from langbot.pkg.box.runtime import BoxRuntime
_UTC = dt.timezone.utc
class FakeManagedProcessBackend(BaseSandboxBackend):
name = 'fake-managed'
def __init__(self, logger: Mock):
super().__init__(logger)
async def is_available(self) -> bool:
return True
async def start_session(self, spec: BoxSpec) -> BoxSessionInfo:
now = dt.datetime.now(_UTC)
return BoxSessionInfo(
session_id=spec.session_id,
backend_name=self.name,
backend_session_id=f'backend-{spec.session_id}',
image=spec.image,
network=spec.network,
host_path=spec.host_path,
host_path_mode=spec.host_path_mode,
cpus=spec.cpus,
memory_mb=spec.memory_mb,
pids_limit=spec.pids_limit,
read_only_rootfs=spec.read_only_rootfs,
created_at=now,
last_used_at=now,
)
async def exec(self, session: BoxSessionInfo, spec: BoxSpec):
raise NotImplementedError
async def stop_session(self, session: BoxSessionInfo):
return None
async def start_managed_process(self, session: BoxSessionInfo, spec: BoxManagedProcessSpec) -> asyncio.subprocess.Process:
return await asyncio.create_subprocess_exec(
'sh',
'-lc',
'cat',
stdin=asyncio.subprocess.PIPE,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
@pytest.mark.asyncio
async def test_runtime_start_managed_process_tracks_status():
logger = Mock()
runtime = BoxRuntime(logger=logger, backends=[FakeManagedProcessBackend(logger)], session_ttl_sec=300)
await runtime.initialize()
session_spec = BoxSpec.model_validate({'cmd': 'echo bootstrap', 'session_id': 'mcp-session'})
await runtime.create_session(session_spec)
process_info = await runtime.start_managed_process(
'mcp-session',
BoxManagedProcessSpec(command='python', args=['-m', 'demo'], cwd='/workspace'),
)
assert process_info['session_id'] == 'mcp-session'
assert process_info['status'] == BoxManagedProcessStatus.RUNNING.value
assert process_info['command'] == 'python'
assert process_info['args'] == ['-m', 'demo']
queried = runtime.get_managed_process('mcp-session')
assert queried['status'] == BoxManagedProcessStatus.RUNNING.value
await runtime.shutdown()
@pytest.mark.asyncio
async def test_runtime_does_not_reap_session_with_running_managed_process():
logger = Mock()
runtime = BoxRuntime(logger=logger, backends=[FakeManagedProcessBackend(logger)], session_ttl_sec=1)
await runtime.initialize()
session_spec = BoxSpec.model_validate({'cmd': 'echo bootstrap', 'session_id': 'mcp-session'})
await runtime.create_session(session_spec)
await runtime.start_managed_process(
'mcp-session',
BoxManagedProcessSpec(command='python', args=['-m', 'demo'], cwd='/workspace'),
)
runtime._sessions['mcp-session'].info.last_used_at = dt.datetime.now(_UTC) - dt.timedelta(seconds=120)
await runtime._reap_expired_sessions_locked()
assert 'mcp-session' in runtime._sessions
await runtime.shutdown()

View File

@@ -0,0 +1,59 @@
from __future__ import annotations
import pytest
from langbot.pkg.box.errors import BoxValidationError
from langbot.pkg.box.models import BoxHostMountMode, BoxNetworkMode, BoxSpec
from langbot.pkg.box.security import BLOCKED_HOST_PATHS, validate_sandbox_security
def _make_spec(**overrides) -> BoxSpec:
defaults = {
'session_id': 'test-session',
'cmd': 'echo hi',
'image': 'python:3.11-slim',
}
defaults.update(overrides)
return BoxSpec(**defaults)
class TestValidateSandboxSecurity:
def test_no_host_path_passes(self):
spec = _make_spec(host_path=None)
validate_sandbox_security(spec) # should not raise
def test_safe_host_path_passes(self):
spec = _make_spec(host_path='/home/user/my-project')
validate_sandbox_security(spec) # should not raise
@pytest.mark.parametrize('blocked', [
'/etc',
'/proc',
'/sys',
'/dev',
'/root',
'/boot',
'/run',
'/var/run',
'/run/docker.sock',
'/var/run/docker.sock',
'/run/podman',
'/var/run/podman',
])
def test_blocked_paths_rejected(self, blocked):
spec = _make_spec(host_path=blocked)
with pytest.raises(BoxValidationError, match='blocked for security'):
validate_sandbox_security(spec)
def test_blocked_subpath_rejected(self):
spec = _make_spec(host_path='/etc/nginx')
with pytest.raises(BoxValidationError, match='blocked for security'):
validate_sandbox_security(spec)
def test_path_starting_with_blocked_prefix_but_different_dir_passes(self):
# /etcetera is NOT /etc
spec = _make_spec(host_path='/etcetera/data')
validate_sandbox_security(spec) # should not raise
def test_blocked_host_paths_is_frozenset(self):
assert isinstance(BLOCKED_HOST_PATHS, frozenset)

View File

@@ -19,6 +19,7 @@ from langbot.pkg.box.models import (
BoxExecutionResult,
BoxExecutionStatus,
BoxHostMountMode,
BoxManagedProcessSpec,
BoxNetworkMode,
BoxProfile,
BoxSessionInfo,
@@ -60,6 +61,12 @@ class _InProcessBoxRuntimeClient(BoxRuntimeClient):
async def create_session(self, spec):
return await self._runtime.create_session(spec)
async def start_managed_process(self, session_id: str, spec: BoxManagedProcessSpec):
return await self._runtime.start_managed_process(session_id, spec)
async def get_managed_process(self, session_id: str):
return self._runtime.get_managed_process(session_id)
def _can_open_test_socket() -> bool:
try:
@@ -1191,3 +1198,46 @@ async def test_remote_client_exec_raises_conflict_error():
await client.shutdown()
finally:
await server.close()
# ── BoxHostMountMode.NONE tests ─────────────────────────────────────
class TestBoxHostMountModeNone:
def test_none_mode_is_valid_enum(self):
assert BoxHostMountMode.NONE.value == 'none'
def test_spec_with_none_mode_skips_workdir_check(self):
"""When host_path_mode is NONE, workdir validation is skipped."""
spec = BoxSpec(
session_id='test',
cmd='echo hi',
host_path='/home/user/data',
host_path_mode=BoxHostMountMode.NONE,
workdir='/opt/custom', # Not under /workspace, should be allowed
)
assert spec.host_path_mode == BoxHostMountMode.NONE
assert spec.workdir == '/opt/custom'
def test_spec_with_rw_mode_requires_workspace_workdir(self):
"""When host_path_mode is RW, workdir must be under /workspace."""
with pytest.raises(Exception):
BoxSpec(
session_id='test',
cmd='echo hi',
host_path='/home/user/data',
host_path_mode=BoxHostMountMode.READ_WRITE,
workdir='/opt/custom',
)
def test_spec_with_ro_mode_requires_workspace_workdir(self):
"""When host_path_mode is RO, workdir must be under /workspace."""
with pytest.raises(Exception):
BoxSpec(
session_id='test',
cmd='echo hi',
host_path='/home/user/data',
host_path_mode=BoxHostMountMode.READ_ONLY,
workdir='/opt/custom',
)

View File

@@ -0,0 +1,421 @@
"""Tests for MCP Box integration: path rewriting, host_path inference, config model, payloads.
Uses importlib.util.spec_from_file_location to load mcp.py directly without
triggering the circular import chain through the app module.
"""
from __future__ import annotations
import importlib
import importlib.util
import os
import sys
import tempfile
import types
from unittest.mock import Mock
import pytest
# ---------------------------------------------------------------------------
# Load mcp.py directly from file path, with stub dependencies
# ---------------------------------------------------------------------------
def _stub_module(fqn: str, attrs: dict | None = None, is_package: bool = False):
"""Create or return a stub module and register it in sys.modules."""
if fqn in sys.modules:
mod = sys.modules[fqn]
else:
mod = types.ModuleType(fqn)
mod.__spec__ = importlib.machinery.ModuleSpec(fqn, None, is_package=is_package)
if is_package:
mod.__path__ = []
sys.modules[fqn] = mod
parts = fqn.rsplit('.', 1)
if len(parts) == 2 and parts[0] in sys.modules:
setattr(sys.modules[parts[0]], parts[1], mod)
if attrs:
for k, v in attrs.items():
setattr(mod, k, v)
return mod
@pytest.fixture(scope='module', autouse=True)
def mcp_module():
"""Load mcp.py with minimal stubs to avoid circular imports."""
saved = {}
def _save_and_stub(name, attrs=None, is_package=False):
saved[name] = sys.modules.get(name)
# Don't overwrite modules that already exist (from other test modules)
if name in sys.modules:
return
_stub_module(name, attrs, is_package)
# Stub entire dependency chains as packages / modules
_save_and_stub('langbot_plugin', is_package=True)
_save_and_stub('langbot_plugin.api', is_package=True)
_save_and_stub('langbot_plugin.api.entities', is_package=True)
_save_and_stub('langbot_plugin.api.entities.events', is_package=True)
_save_and_stub('langbot_plugin.api.entities.events.pipeline_query', {})
_save_and_stub('langbot_plugin.api.entities.builtin', is_package=True)
_save_and_stub('langbot_plugin.api.entities.builtin.resource', is_package=True)
_save_and_stub('langbot_plugin.api.entities.builtin.resource.tool', {
'LLMTool': type('LLMTool', (), {}),
})
_save_and_stub('langbot_plugin.api.entities.builtin.provider', is_package=True)
_save_and_stub('langbot_plugin.api.entities.builtin.provider.message', {})
_save_and_stub('sqlalchemy', {'select': Mock()})
_save_and_stub('httpx', {'AsyncClient': Mock()})
_save_and_stub('mcp', {'ClientSession': Mock, 'StdioServerParameters': Mock}, is_package=True)
_save_and_stub('mcp.client', is_package=True)
_save_and_stub('mcp.client.stdio', {'stdio_client': Mock()})
_save_and_stub('mcp.client.sse', {'sse_client': Mock()})
_save_and_stub('mcp.client.streamable_http', {'streamable_http_client': Mock()})
_save_and_stub('mcp.client.websocket', {'websocket_client': Mock()})
# Stub the provider.tools.loader (source of circular import)
_save_and_stub('langbot', is_package=True)
_save_and_stub('langbot.pkg', is_package=True)
_save_and_stub('langbot.pkg.provider', is_package=True)
_save_and_stub('langbot.pkg.provider.tools', is_package=True)
_save_and_stub('langbot.pkg.provider.tools.loader', {
'ToolLoader': type('ToolLoader', (), {'__init__': lambda self, ap: None}),
})
_save_and_stub('langbot.pkg.provider.tools.loaders', is_package=True)
_save_and_stub('langbot.pkg.core', is_package=True)
_save_and_stub('langbot.pkg.core.app', {'Application': type('Application', (), {})})
_save_and_stub('langbot.pkg.entity', is_package=True)
_save_and_stub('langbot.pkg.entity.persistence', is_package=True)
_save_and_stub('langbot.pkg.entity.persistence.mcp', {})
# box models
import enum as _enum
class _BPS(str, _enum.Enum):
RUNNING = 'running'
EXITED = 'exited'
_save_and_stub('langbot.pkg.box', is_package=True)
_save_and_stub('langbot.pkg.box.models', {'BoxManagedProcessStatus': _BPS})
# Now load mcp.py via spec_from_file_location
mod_fqn = 'langbot.pkg.provider.tools.loaders.mcp'
sys.modules.pop(mod_fqn, None)
mcp_path = os.path.join(
os.path.dirname(__file__), '..', '..', '..',
'src', 'langbot', 'pkg', 'provider', 'tools', 'loaders', 'mcp.py',
)
mcp_path = os.path.normpath(mcp_path)
spec = importlib.util.spec_from_file_location(mod_fqn, mcp_path)
mod = importlib.util.module_from_spec(spec)
sys.modules[mod_fqn] = mod
spec.loader.exec_module(mod)
yield mod
# Cleanup
sys.modules.pop(mod_fqn, None)
for name in reversed(list(saved)):
if saved[name] is None:
sys.modules.pop(name, None)
else:
sys.modules[name] = saved[name]
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_ap():
ap = Mock()
ap.logger = Mock()
ap.box_service = Mock()
return ap
def _make_session(mcp_module, server_config: dict, ap=None):
if ap is None:
ap = _make_ap()
return mcp_module.RuntimeMCPSession(
server_name=server_config.get('name', 'test-server'),
server_config=server_config,
enable=True,
ap=ap,
)
# ── MCPServerBoxConfig ──────────────────────────────────────────────
class TestMCPServerBoxConfig:
def test_default_values(self, mcp_module):
cfg = mcp_module.MCPServerBoxConfig.model_validate({})
assert cfg.image is None
assert cfg.network == 'on'
assert cfg.host_path is None
assert cfg.host_path_mode == 'ro'
assert cfg.env == {}
assert cfg.startup_timeout_sec == 120
assert cfg.cpus is None
assert cfg.memory_mb is None
assert cfg.pids_limit is None
assert cfg.read_only_rootfs is None
def test_custom_values(self, mcp_module):
cfg = mcp_module.MCPServerBoxConfig.model_validate({
'image': 'node:20',
'network': 'on',
'host_path': '/home/user/mcp',
'host_path_mode': 'rw',
'env': {'FOO': 'bar'},
'startup_timeout_sec': 60,
'cpus': 2.0,
'memory_mb': 1024,
'pids_limit': 256,
'read_only_rootfs': False,
})
assert cfg.image == 'node:20'
assert cfg.network == 'on'
assert cfg.cpus == 2.0
assert cfg.memory_mb == 1024
def test_extra_fields_ignored(self, mcp_module):
cfg = mcp_module.MCPServerBoxConfig.model_validate({
'image': 'node:20',
'unknown_field': 'whatever',
})
assert cfg.image == 'node:20'
assert not hasattr(cfg, 'unknown_field')
# ── Path Rewriting ──────────────────────────────────────────────────
class TestRewritePath:
def test_no_host_path_returns_unchanged(self, mcp_module):
s = _make_session(mcp_module, {
'name': 'test', 'uuid': 'u1', 'mode': 'sse',
'command': 'python', 'args': [],
})
assert s._rewrite_path('/some/path', None) == '/some/path'
def test_empty_path_returns_empty(self, mcp_module):
s = _make_session(mcp_module, {
'name': 'test', 'uuid': 'u1', 'mode': 'sse',
'command': 'python', 'args': [],
})
assert s._rewrite_path('', '/home/user/mcp') == ''
def test_prefix_match_rewrites(self, mcp_module):
s = _make_session(mcp_module, {
'name': 'test', 'uuid': 'u1', 'mode': 'sse',
'command': 'python', 'args': [],
})
result = s._rewrite_path('/home/user/mcp/server.py', '/home/user/mcp')
assert result == '/workspace/server.py'
def test_exact_match_rewrites_to_workspace(self, mcp_module):
s = _make_session(mcp_module, {
'name': 'test', 'uuid': 'u1', 'mode': 'sse',
'command': 'python', 'args': [],
})
result = s._rewrite_path('/home/user/mcp', '/home/user/mcp')
assert result == '/workspace'
def test_non_matching_path_unchanged(self, mcp_module):
s = _make_session(mcp_module, {
'name': 'test', 'uuid': 'u1', 'mode': 'sse',
'command': 'python', 'args': [],
})
result = s._rewrite_path('/opt/other/server.py', '/home/user/mcp')
assert result == '/opt/other/server.py'
def test_similar_prefix_not_rewritten(self, mcp_module):
s = _make_session(mcp_module, {
'name': 'test', 'uuid': 'u1', 'mode': 'sse',
'command': 'python', 'args': [],
})
result = s._rewrite_path('/home/user/mcp-other/file.py', '/home/user/mcp')
assert result == '/home/user/mcp-other/file.py'
def test_nested_subpath_rewrites(self, mcp_module):
s = _make_session(mcp_module, {
'name': 'test', 'uuid': 'u1', 'mode': 'sse',
'command': 'python', 'args': [],
})
result = s._rewrite_path('/home/user/mcp/src/lib/main.py', '/home/user/mcp')
assert result == '/workspace/src/lib/main.py'
# ── host_path Inference ─────────────────────────────────────────────
class TestInferHostPath:
def test_no_absolute_paths_returns_none(self, mcp_module):
s = _make_session(mcp_module, {
'name': 'test', 'uuid': 'u1', 'mode': 'sse',
'command': 'python', 'args': ['server.py'],
})
assert s._infer_host_path() is None
def test_nonexistent_path_returns_none(self, mcp_module):
s = _make_session(mcp_module, {
'name': 'test', 'uuid': 'u1', 'mode': 'sse',
'command': '/nonexistent/path/to/python', 'args': [],
})
assert s._infer_host_path() is None
def test_existing_absolute_path_infers_directory(self, mcp_module):
with tempfile.NamedTemporaryFile(suffix='.py') as f:
s = _make_session(mcp_module, {
'name': 'test', 'uuid': 'u1', 'mode': 'sse',
'command': 'python', 'args': [f.name],
})
result = s._infer_host_path()
assert result is not None
assert result == os.path.dirname(os.path.realpath(f.name))
# ── Build Box Session Payload ───────────────────────────────────────
class TestBuildBoxSessionPayload:
def test_minimal_config(self, mcp_module):
s = _make_session(mcp_module, {
'name': 'test', 'uuid': 'u1', 'mode': 'sse',
'command': 'python', 'args': [],
})
payload = s._build_box_session_payload('session-123')
assert payload['session_id'] == 'session-123'
assert payload['workdir'] == '/workspace'
assert payload['env'] == {}
assert 'host_path' not in payload
def test_with_host_path(self, mcp_module):
s = _make_session(mcp_module, {
'name': 'test', 'uuid': 'u1', 'mode': 'sse',
'command': 'python', 'args': [],
'box': {'host_path': '/home/user/mcp', 'host_path_mode': 'ro'},
})
payload = s._build_box_session_payload('session-123')
assert payload['host_path'] == '/home/user/mcp'
assert payload['host_path_mode'] == 'ro'
def test_optional_fields_included_when_set(self, mcp_module):
s = _make_session(mcp_module, {
'name': 'test', 'uuid': 'u1', 'mode': 'sse',
'command': 'python', 'args': [],
'box': {'image': 'node:20', 'cpus': 2.0, 'memory_mb': 1024, 'pids_limit': 256},
})
payload = s._build_box_session_payload('session-123')
assert payload['image'] == 'node:20'
assert payload['cpus'] == 2.0
assert payload['memory_mb'] == 1024
assert payload['pids_limit'] == 256
def test_none_fields_excluded(self, mcp_module):
s = _make_session(mcp_module, {
'name': 'test', 'uuid': 'u1', 'mode': 'sse',
'command': 'python', 'args': [],
})
payload = s._build_box_session_payload('session-123')
assert 'image' not in payload
assert 'cpus' not in payload
# ── Build Box Process Payload ───────────────────────────────────────
class TestBuildBoxProcessPayload:
def test_basic_payload(self, mcp_module):
s = _make_session(mcp_module, {
'name': 'test', 'uuid': 'u1', 'mode': 'sse',
'command': 'python', 'args': ['server.py'], 'env': {'KEY': 'val'},
})
payload = s._build_box_process_payload()
assert payload['command'] == 'python'
assert payload['args'] == ['server.py']
assert payload['env'] == {'KEY': 'val'}
assert payload['cwd'] == '/workspace'
def test_path_rewriting_applied(self, mcp_module):
s = _make_session(mcp_module, {
'name': 'test', 'uuid': 'u1', 'mode': 'sse',
'command': '/home/user/mcp/venv/bin/python',
'args': ['/home/user/mcp/server.py', '--config', '/home/user/mcp/config.json'],
'env': {},
'box': {'host_path': '/home/user/mcp'},
})
payload = s._build_box_process_payload()
# venv python is replaced with plain 'python' (deps installed in-container)
assert payload['command'] == 'python'
assert payload['args'] == ['/workspace/server.py', '--config', '/workspace/config.json']
def test_non_matching_args_not_rewritten(self, mcp_module):
s = _make_session(mcp_module, {
'name': 'test', 'uuid': 'u1', 'mode': 'sse',
'command': 'python',
'args': ['/opt/other/server.py', '--flag'],
'env': {},
'box': {'host_path': '/home/user/mcp'},
})
payload = s._build_box_process_payload()
assert payload['command'] == 'python'
assert payload['args'] == ['/opt/other/server.py', '--flag']
# ── get_runtime_info_dict ───────────────────────────────────────────
class TestGetRuntimeInfoDict:
def test_non_stdio_session(self, mcp_module):
s = _make_session(mcp_module, {
'name': 'test', 'uuid': 'test-uuid', 'mode': 'sse',
'command': 'python', 'args': [],
})
info = s.get_runtime_info_dict()
assert info['status'] == 'connecting'
assert 'box_session_id' not in info
def test_stdio_session_includes_box_info(self, mcp_module):
ap = _make_ap()
ap.box_service.available = True
s = _make_session(mcp_module, {
'name': 'test', 'uuid': 'test-uuid', 'mode': 'stdio',
'command': 'python', 'args': [],
}, ap=ap)
info = s.get_runtime_info_dict()
assert info['box_session_id'] == 'mcp-test-uuid'
assert info['box_enabled'] is True
def test_stdio_session_without_box_runtime(self, mcp_module):
ap = _make_ap()
ap.box_service.available = False
s = _make_session(mcp_module, {
'name': 'test', 'uuid': 'test-uuid', 'mode': 'stdio',
'command': 'python', 'args': [],
}, ap=ap)
info = s.get_runtime_info_dict()
assert 'box_session_id' not in info
# ── Box config parsing ──────────────────────────────────────────────
class TestBoxConfigParsing:
def test_box_config_parsed_from_server_config(self, mcp_module):
s = _make_session(mcp_module, {
'name': 'test', 'uuid': 'u1', 'mode': 'sse',
'command': 'python', 'args': [],
'box': {'image': 'node:20', 'host_path': '/home/user/mcp'},
})
assert isinstance(s.box_config, mcp_module.MCPServerBoxConfig)
assert s.box_config.image == 'node:20'
assert s.box_config.host_path == '/home/user/mcp'
def test_missing_box_key_uses_defaults(self, mcp_module):
s = _make_session(mcp_module, {
'name': 'test', 'uuid': 'u1', 'mode': 'sse',
'command': 'python', 'args': [],
})
assert isinstance(s.box_config, mcp_module.MCPServerBoxConfig)
assert s.box_config.image is None
assert s.box_config.host_path_mode == 'ro'