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 = 60,
    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`	Timeout in seconds for approval requests. If an approval request is not accepted or rejected within this time, the tool call fails.	`60`
`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'`

execute `async`

execute(
    code: str, timeout: float = 120
) -> 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`	Maximum time in seconds to wait for execution to complete.	`120`

Returns:

Type	Description
`CodeExecutionResult`	The execution result containing output text and generated images.

Raises:

Type	Description
`CodeExecutionError`	If code execution raises an error.
`TimeoutError`	If code execution exceeds the timeout.

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 = 120, 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.

Parameters:

Name	Type	Description	Default
`code`	`str`	Python code to execute.	required
`timeout`	`float`	Maximum time in seconds to wait for execution to complete.	`120`
`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.

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.

ipybox.generate_mcp_sources `async`

generate_mcp_sources(
    server_name: str,
    server_params: dict[str, Any],
    root_dir: Path,
) -> list[str]

Generate a typed Python tool API for an MCP server.

Connects to an MCP server, discovers available tools, and generates a Python package with typed functions backed by Pydantic models. Each tool becomes a module with a Params class for input validation and a run() function to invoke the tool.

When calling the generated API, the corresponding tools are executed on a ToolServer.

If a directory for the server already exists under root_dir, it is removed and recreated.

Parameters:

Name	Type	Description	Default
`server_name`	`str`	Name for the generated package directory. Also used to identify the server in the generated client code.	required
`server_params`	`dict[str, Any]`	MCP server connection parameters. For stdio servers, provide `command`, `args`, and optionally `env`. For HTTP servers, provide `url` and optionally `headers`.	required
`root_dir`	`Path`	Parent directory where the package will be created. The generated package is written to `root_dir/server_name/`.	required

Returns:

Type	Description
`list[str]`	List of sanitized tool names corresponding to the generated module files.

Example

Generate a Python tool API for the fetch MCP server:

server_params = {
    "command": "uvx",
    "args": ["mcp-server-fetch"],
}
await generate_mcp_sources("fetch_mcp", server_params, Path("mcptools"))

Execute code that uses the generated API:

from ipybox.code_exec import CodeExecutor

code = """
from mcptools.fetch_mcp import fetch

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

async with CodeExecutor() as executor:
    async for item in executor.execute(code):
        ...

Code Executor

ipybox.CodeExecutor

execute async

reset async

start async

stop async

stream async

ipybox.CodeExecutionChunk dataclass

ipybox.CodeExecutionResult dataclass

ipybox.CodeExecutionError

ipybox.generate_mcp_sources async

execute `async`

reset `async`

start `async`

stop `async`

stream `async`

ipybox.CodeExecutionChunk `dataclass`

ipybox.CodeExecutionResult `dataclass`

ipybox.generate_mcp_sources `async`