Code Executor

ipybox.CodeExecutor

CodeExecutor(
    tool_server_host: str = "localhost",
    tool_server_port: int | None = None,
    kernel_gateway_host: str = "localhost",
    kernel_gateway_port: int | None = None,
    kernel_env: dict[str, str] | None = None,
    images_dir: Path | None = None,
    approval_timeout: float | None = None,
    connect_timeout: float = 30,
    sandbox: bool = False,
    sandbox_config: Path | None = None,
    log_level: str = "WARNING",
)

Executes Python code in an IPython kernel with programmatic MCP tool call support.

CodeExecutor launches an embedded KernelGateway for running Python code and an embedded ToolServer for executing MCP tools. Code is executed in an IPython kernel, providing a stateful environment where variables and definitions persist across executions.

MCP tools can be called from executed code using the API generated by generate_mcp_sources. When code calls an MCP tool, the tool server receives the request and emits an approval request. The client must accept or reject the request before the tool executes.

Example

Generate a Python tool API and execute code that uses it:

from pathlib import Path

from ipybox import ApprovalRequest, CodeExecutionResult, CodeExecutor
from ipybox import generate_mcp_sources

# Generate a Python tool API for the fetch MCP server
server_params = {"command": "uvx", "args": ["mcp-server-fetch"]}
await generate_mcp_sources("fetch", server_params, Path("mcptools"))

# Execute code that calls the generated API
code = """
from mcptools.fetch import fetch

result = fetch.run(fetch.Params(url="https://example.com"))
print(result)
"""

async with CodeExecutor() as executor:
    async for item in executor.stream(code):
        match item:
            case ApprovalRequest():
                print(f"Tool call: {item}")
                await item.accept()
            case CodeExecutionResult():
                print(item.text)

Configure a code executor with optional sandboxing.

Parameters:

Name	Type	Description	Default
`tool_server_host`	`str`	Hostname for the `ToolServer`.	`'localhost'`
`tool_server_port`	`int \| None`	Port for the tool server. If `None`, a free port is selected automatically.	`None`
`kernel_gateway_host`	`str`	Hostname for the `KernelGateway`.	`'localhost'`
`kernel_gateway_port`	`int \| None`	Port for the kernel gateway. If `None`, a free port is selected automatically.	`None`
`kernel_env`	`dict[str, str] \| None`	Environment variables to set for the IPython kernel. Kernels do not inherit environment variables from the parent process.	`None`
`images_dir`	`Path \| None`	Directory for saving images generated during code execution. Defaults to `images` in the current directory.	`None`
`approval_timeout`	`float \| None`	Timeout in seconds for approval requests. If an approval request is not accepted or rejected within this time, the tool call fails. If `None`, no timeout is applied.	`None`
`connect_timeout`	`float`	Timeout in seconds for starting MCP servers.	`30`
`sandbox`	`bool`	Whether to run the kernel gateway inside Anthropic's sandbox-runtime. When enabled, IPython kernels run in a secure sandbox with no network access except to the local tool server.	`False`
`sandbox_config`	`Path \| None`	Path to a JSON file with sandbox configuration. See the Configuration section of the sandbox-runtime README for available options.	`None`
`log_level`	`str`	Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL).	`'WARNING'`

cancel

cancel()

Cancel the currently running execution.

Signals the active stream or execute call to interrupt the kernel and return cleanly without raising an exception. Safe to call from any coroutine in the same event loop.

Has no effect if no execution is in progress.

execute `async`

execute(
    code: str, timeout: float | None = None
) -> CodeExecutionResult

Execute Python code with automatic approval of all MCP tool calls.

Convenience method that executes code, auto-approves any MCP tool calls, and returns the final result directly.

Parameters:

Name	Type	Description	Default
`code`	`str`	Python code to execute.	required
`timeout`	`float \| None`	Maximum time in seconds to wait for execution to complete. If `None`, no timeout is applied. Approval wait time is excluded from this timeout budget.	`None`

Returns:

Type	Description
`CodeExecutionResult`	The execution result containing output text and generated images. If the execution was cancelled, returns a result with `text=None` and empty `images`.

Raises:

Type	Description
`CodeExecutionError`	If code execution raises an error.
`TimeoutError`	If code execution exceeds the timeout (excluding approval wait time).

reset `async`

reset()

Reset execution state.

Restarts the IPython kernel and stops all started MCP servers. Kernel state (variables, definitions, imports) is lost. MCP servers are lazily restarted on their next tool call.

start `async`

start()

Start the executor.

Starts the tool server, kernel gateway, and connects to the IPython kernel.

stop `async`

stop()

Stop the executor.

Stops the tool server, kernel gateway, and disconnects from the IPython kernel.

stream `async`

stream(
    code: str,
    timeout: float | None = None,
    chunks: bool = False,
) -> AsyncIterator[
    ApprovalRequest
    | CodeExecutionChunk
    | CodeExecutionResult
]

Execute Python code in the IPython kernel with MCP tool call approval.

Code can call MCP tools using the API generated by generate_mcp_sources. Each tool call yields an ApprovalRequest. If accepted, the tool executes on the ToolServer and returns the result to the kernel. If rejected, the tool call fails with an error.

Use cancel to stop execution and return cleanly from a concurrent coroutine.

Parameters:

Name	Type	Description	Default
`code`	`str`	Python code to execute.	required
`timeout`	`float \| None`	Maximum time in seconds to wait for execution to complete. If `None`, no timeout is applied. Approval wait time is excluded from this timeout budget.	`None`
`chunks`	`bool`	Whether to yield `CodeExecutionChunk` objects during execution. When `False`, only `ApprovalRequest` and `CodeExecutionResult` are yielded.	`False`

Yields:

Type	Description
`AsyncIterator[ApprovalRequest \| CodeExecutionChunk \| CodeExecutionResult]`	`ApprovalRequest`: When executed code calls an MCP tool. Accept to execute the tool, reject to fail the tool call.
`AsyncIterator[ApprovalRequest \| CodeExecutionChunk \| CodeExecutionResult]`	`CodeExecutionChunk`: Output text chunks generated during execution (emitted only if `chunks=True`).
`AsyncIterator[ApprovalRequest \| CodeExecutionChunk \| CodeExecutionResult]`	`CodeExecutionResult`: The final result when execution completes successfully.

Raises:

Type	Description
`CodeExecutionError`	If code execution raises an error (syntax errors, runtime errors, rejected or timed-out approval requests, MCP tool errors). The error message contains the stack trace from the kernel.
`TimeoutError`	If code execution exceeds the timeout (excluding approval wait time).

ipybox.CodeExecutionChunk `dataclass`

CodeExecutionChunk(text: str)

A chunk of output text generated during streaming code execution.

Only yielded by CodeExecutor.stream when chunks=True.

Attributes:

Name	Type	Description
`text`	`str`	A chunk of output text.

ipybox.CodeExecutionResult `dataclass`

CodeExecutionResult(text: str | None, images: list[Path])

The result of a successful code execution.

Attributes:

Name	Type	Description
`text`	`str \| None`	Output text generated during execution, or `None` if no output.
`images`	`list[Path]`	Paths to images generated during execution.

ipybox.CodeExecutionError

Bases: Exception

Raised when code execution in an IPython kernel fails.

Code Executor

ipybox.CodeExecutor

cancel

execute async

reset async

start async

stop async

stream async

ipybox.CodeExecutionChunk dataclass

ipybox.CodeExecutionResult dataclass

ipybox.CodeExecutionError

execute `async`

reset `async`

start `async`

stop `async`

stream `async`

ipybox.CodeExecutionChunk `dataclass`

ipybox.CodeExecutionResult `dataclass`