InterpreterService
class InterpreterService()InterpreterService.__init__
def __init__(sandbox_id: str,
transport: AsyncTransport,
process: ProcessService) -> NoneInterpreterService.create_context
async def create_context(*,
language: str,
context_id: Optional[str] = None,
cwd: Optional[str] = None,
env: Optional[Mapping[str, str]] = None,
request_timeout: Optional[float] = None) -> InterpreterContextInterpreterService.connect_context
def connect_context(context: str | InterpreterContext | InterpreterContextJSON,
*,
language: str | None = None) -> InterpreterContextInterpreterService.get_context
def get_context(context: str | InterpreterContext | InterpreterContextJSON,
*,
language: str | None = None) -> InterpreterContextInterpreterService.list_contexts
async def list_contexts() -> list[InterpreterContext]InterpreterService.delete_context
async def delete_context(context: InterpreterContext | InterpreterContextJSON | str) -> NoneInterpreterService.run
async def run(code: str,
*,
language: str = 'python',
context: Optional[InterpreterContext] = None,
isolated: bool = False,
timeout: Optional[float] = None,
cwd: Optional[str] = None,
env: Optional[Mapping[str, str]] = None) -> InterpreterResultExecute code and return the parsed interpreter result.
A rehydrated kind='code' process handle can still replay logs and
wait/kill, but it does not reconstruct this decomposed result on
another worker. A vanished explicit context surfaces the daemon's typed
SESSION_NOT_FOUND error; callers decide whether to recreate the
context and what to rerun. timeout is the caller wait budget in
seconds; 0 waits unbounded.
InterpreterContextJSON
class InterpreterContextJSON(TypedDict)Attributes:
sandbox_idstrcontext_idstrlanguagestr
InterpreterDisplay
@dataclass(frozen=True)
class InterpreterDisplay()One rich interpreter output bundle.
Attributes:
mimestrdatastr
InterpreterProcess
@dataclass(frozen=True)
class InterpreterProcess()Attributes:
process_idstrexit_codeOptional[int]signalOptional[str]reasonOptional[WaitReason]raw_reasonOptional[str] = None
InterpreterLogs
@dataclass(frozen=True)
class InterpreterLogs()Attributes:
stdoutstr = ''stderrstr = ''
InterpreterError
@dataclass(frozen=True)
class InterpreterError()Attributes:
namestrvaluestrtracebackList[str] = field(default_factory=list)
InterpreterOutput
@dataclass(frozen=True)
class InterpreterOutput()Attributes:
dataJsonObjectis_main_resultbooltextOptional[str] = NonehtmlOptional[str] = NonepngOptional[str] = NonesvgOptional[str] = NonejsonOptional[JsonValue] = None
InterpreterResult
@dataclass(frozen=True)
class InterpreterResult()Interpreter run result.
Attributes:
processInterpreterProcessresultsList[InterpreterOutput] = field(default_factory=list)logsInterpreterLogs = field(default_factory=InterpreterLogs)errorOptional[InterpreterError] = None
InterpreterContext
@dataclass(frozen=True)
class InterpreterContext()Serializable interpreter context handle.
Attributes:
sandbox_idstrcontext_idstrlanguagestr
InterpreterContext.bind
def bind(runner: Callable[[str, Optional[str], Optional[Mapping[str, str]], Optional[float]], Awaitable[InterpreterResult]],
closer: Callable[[], Awaitable[None]]) -> SelfInterpreterContext.run
async def run(code: str,
*,
cwd: Optional[str] = None,
env: Optional[Mapping[str, str]] = None,
timeout: Optional[float] = None) -> InterpreterResultExecute code inside this explicit interpreter context.
If the named context vanished, the daemon's typed
SESSION_NOT_FOUND error surfaces unchanged so the caller can
recreate the context and decide what code, if any, is safe to rerun.
timeout is the caller wait budget in seconds; 0 waits
unbounded.
InterpreterContext.close
async def close() -> NoneSyncInterpreterService
class SyncInterpreterService()Synchronous InterpreterService.
SyncInterpreterService.__init__
def __init__(impl: InterpreterService, bridge: SyncBridge) -> NoneSyncInterpreterService.connect_context
def connect_context(context: str | SyncInterpreterContext | InterpreterContextJSON,
*,
language: str | None = None) -> SyncInterpreterContextSyncInterpreterService.create_context
def create_context(*,
language: str,
context_id: Optional[str] = None,
cwd: Optional[str] = None,
env: Optional[Mapping[str, str]] = None,
request_timeout: Optional[float] = None) -> SyncInterpreterContextSyncInterpreterService.delete_context
def delete_context(context: SyncInterpreterContext | InterpreterContextJSON | str) -> NoneSyncInterpreterService.get_context
def get_context(context: str | SyncInterpreterContext | InterpreterContextJSON,
*,
language: str | None = None) -> SyncInterpreterContextSyncInterpreterService.list_contexts
def list_contexts() -> list[SyncInterpreterContext]SyncInterpreterService.run
def run(code: str,
*,
language: str = 'python',
context: Optional[SyncInterpreterContext] = None,
isolated: bool = False,
timeout: Optional[float] = None,
cwd: Optional[str] = None,
env: Optional[Mapping[str, str]] = None) -> InterpreterResultExecute code and return the parsed interpreter result.
SyncInterpreterContext
class SyncInterpreterContext()Serializable interpreter context handle.
SyncInterpreterContext.__init__
def __init__(impl: InterpreterContext, bridge: SyncBridge) -> NoneSyncInterpreterContext.sandbox_id
@property
def sandbox_id() -> strSyncInterpreterContext.context_id
@property
def context_id() -> strSyncInterpreterContext.language
@property
def language() -> strSyncInterpreterContext.close
def close() -> NoneSyncInterpreterContext.run
def run(code: str,
*,
cwd: Optional[str] = None,
env: Optional[Mapping[str, str]] = None,
timeout: Optional[float] = None) -> InterpreterResultExecute code inside this explicit interpreter context.