Coverage for src / mcp_server_langgraph / execution / sandbox.py: 88%

1""" 

2Abstract sandbox interface for secure code execution 

3 

4Defines the contract for all sandbox implementations (Docker, Kubernetes, Process). 

5""" 

6 

7import time 

8from abc import ABC, abstractmethod 

9from dataclasses import dataclass 

10 

11from mcp_server_langgraph.execution.resource_limits import ResourceLimits 

12 

13 

14class SandboxError(Exception): 

15 """Raised when sandbox operations fail""" 

16 

17 

18@dataclass 

19class ExecutionResult: 

20 """ 

21 Result of code execution in a sandbox. 

22 

23 Attributes: 

24 success: True if code executed without errors 

25 stdout: Standard output from execution 

26 stderr: Standard error from execution 

27 exit_code: Process exit code (0 = success) 

28 execution_time: Time taken in seconds 

29 timed_out: True if execution was terminated due to timeout 

30 memory_used_mb: Peak memory usage in MB (if available) 

31 error_message: Human-readable error message (if failed) 

32 """ 

33 

34 success: bool 

35 stdout: str = "" 

36 stderr: str = "" 

37 exit_code: int = 0 

38 execution_time: float = 0.0 

39 timed_out: bool = False 

40 memory_used_mb: float | None = None 

41 error_message: str = "" 

42 

43 def __repr__(self) -> str: 

44 status = "SUCCESS" if self.success else "FAILED" 

45 if self.timed_out: 

46 status = "TIMEOUT" 

47 return f"ExecutionResult(status={status}, exit_code={self.exit_code}, time={self.execution_time:.2f}s)" 

48 

49 

50class Sandbox(ABC): 

51 """ 

52 Abstract base class for code execution sandboxes. 

53 

54 All sandbox implementations (Docker, Kubernetes, Process) must inherit from this 

55 class and implement the execute() method. 

56 

57 Example: 

58 >>> sandbox = DockerSandbox(limits=ResourceLimits.testing()) 

59 >>> result = sandbox.execute("print('Hello, World!')") 

60 >>> assert result.success 

61 >>> assert "Hello, World!" in result.stdout 

62 """ 

63 

64 def __init__(self, limits: ResourceLimits): 

65 """ 

66 Initialize sandbox with resource limits. 

67 

68 Args: 

69 limits: Resource limits to enforce (timeout, memory, CPU, etc.) 

70 """ 

71 self.limits = limits 

72 

73 @abstractmethod 

74 def execute(self, code: str) -> ExecutionResult: 

75 """ 

76 Execute Python code in the sandbox. 

77 

78 This method must be implemented by all sandbox subclasses. 

79 

80 Args: 

81 code: Python source code to execute 

82 

83 Returns: 

84 ExecutionResult with execution status and outputs 

85 

86 Raises: 

87 SandboxError: If sandbox setup or execution fails 

88 """ 

89 

90 def _create_success_result( 

91 self, 

92 stdout: str, 

93 stderr: str, 

94 execution_time: float, 

95 memory_used_mb: float | None = None, 

96 ) -> ExecutionResult: 

97 """Helper to create successful execution result""" 

98 return ExecutionResult( 

99 success=True, 

100 stdout=stdout, 

101 stderr=stderr, 

102 exit_code=0, 

103 execution_time=execution_time, 

104 timed_out=False, 

105 memory_used_mb=memory_used_mb, 

106 ) 

107 

108 def _create_failure_result( 

109 self, 

110 stdout: str, 

111 stderr: str, 

112 exit_code: int, 

113 execution_time: float, 

114 timed_out: bool = False, 

115 error_message: str = "", 

116 ) -> ExecutionResult: 

117 """Helper to create failed execution result""" 

118 return ExecutionResult( 

119 success=False, 

120 stdout=stdout, 

121 stderr=stderr, 

122 exit_code=exit_code, 

123 execution_time=execution_time, 

124 timed_out=timed_out, 

125 error_message=error_message, 

126 ) 

127 

128 def _measure_time(self, start_time: float) -> float: 

129 """Helper to measure execution time""" 

130 return time.time() - start_time