Skip to content
View as Markdown
class InterpreterService()

Synchronous InterpreterService.

def __init__(impl: AsyncInterpreterService, bridge: SyncBridge) -> None
def connect_context(context: str | InterpreterContext | InterpreterContextJSON,
*,
language: str | None = None) -> InterpreterContext
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) -> InterpreterContext
def delete_context(context: InterpreterContext | InterpreterContextJSON | str) -> None
def get_context(context: str | InterpreterContext | InterpreterContextJSON,
*,
language: str | None = None) -> InterpreterContext
def list_contexts() -> list[InterpreterContext]
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) -> InterpreterResult

Execute code and return the parsed interpreter result.

class AsyncInterpreterService()
def __init__(sandbox_id: str,
transport: AsyncTransport,
process: AsyncProcessService) -> None
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) -> AsyncInterpreterContext
def connect_context(context: str | AsyncInterpreterContext | InterpreterContextJSON,
*,
language: str | None = None) -> AsyncInterpreterContext
def get_context(context: str | AsyncInterpreterContext | InterpreterContextJSON,
*,
language: str | None = None) -> AsyncInterpreterContext
async def list_contexts() -> list[AsyncInterpreterContext]
async def delete_context(context: AsyncInterpreterContext | InterpreterContextJSON | str) -> None
async def run(code: str,
*,
language: str = 'python',
context: Optional[AsyncInterpreterContext] = None,
isolated: bool = False,
timeout: Optional[float] = None,
cwd: Optional[str] = None,
env: Optional[Mapping[str, str]] = None) -> InterpreterResult

Execute 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.

class InterpreterContext()

Serializable interpreter context handle.

def __init__(impl: AsyncInterpreterContext, bridge: SyncBridge) -> None
@property
def sandbox_id() -> str
@property
def context_id() -> str
@property
def language() -> str
def close() -> None
def run(code: str,
*,
cwd: Optional[str] = None,
env: Optional[Mapping[str, str]] = None,
timeout: Optional[float] = None) -> InterpreterResult

Execute code inside this explicit interpreter context.

@dataclass(frozen=True)
class AsyncInterpreterContext()

Serializable interpreter context handle.

Attributes:

  • sandbox_id str
  • context_id str
  • language str
def bind(runner: Callable[[str, Optional[str], Optional[Mapping[str, str]], Optional[float]], Awaitable[InterpreterResult]],
closer: Callable[[], Awaitable[None]]) -> Self
async def run(code: str,
*,
cwd: Optional[str] = None,
env: Optional[Mapping[str, str]] = None,
timeout: Optional[float] = None) -> InterpreterResult

Execute 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.

async def close() -> None
class InterpreterContextJSON(TypedDict)

Attributes:

  • sandbox_id str
  • context_id str
  • language str
@dataclass(frozen=True)
class InterpreterDisplay()

One rich interpreter output bundle.

Attributes:

  • mime str
  • data str
@dataclass(frozen=True)
class InterpreterProcess()

Attributes:

  • process_id str
  • exit_code Optional[int]
  • signal Optional[str]
  • reason Optional[WaitReason]
  • raw_reason Optional[str] = None
@dataclass(frozen=True)
class InterpreterLogs()

Attributes:

  • stdout str = ”
  • stderr str = ”
@dataclass(frozen=True)
class InterpreterError()

Attributes:

  • name str
  • value str
  • traceback List[str] = field(default_factory=list)
@dataclass(frozen=True)
class InterpreterOutput()

Attributes:

  • data JsonObject
  • is_main_result bool
  • text Optional[str] = None
  • html Optional[str] = None
  • png Optional[str] = None
  • svg Optional[str] = None
  • json Optional[JsonValue] = None
@dataclass(frozen=True)
class InterpreterResult()

Interpreter run result.

Attributes:

  • process InterpreterProcess
  • results List[InterpreterOutput] = field(default_factory=list)
  • logs InterpreterLogs = field(default_factory=InterpreterLogs)
  • error Optional[InterpreterError] = None