Environment

Workspace

Workspace(path: Path | str | None = None, key: str | None = None)

A workspace for private and shared agent skills i.e. Python modules that implement special agent skills. These are skills that are not pre-installed in the code execution container.

A workspace defines paths for private and shared skills, both in the container and on the host machine. Workspace paths on the host machine can be bind-mounted into the container, if desired. This is especially useful when skills are being (inter)actively developed, so that they can be inspected and edited on the host machine while being executed in the container.

Parameters:

Name	Type	Description	Default
path	Path | str | None	Root path of the workspace directory on the host.	None
key	str | None	A key to designate: a private skill sub-directory on the host a private image sub-directory on the host	None

Source code in freeact/environment.py

def __init__(self, path: Path | str | None = None, key: str | None = None):
    self._path = Path(path) if path else Path("workspace")
    self._key = key or "default"

skills_host_path property

skills_host_path: Path

Path to skills root directory on host.

private_skills_host_path property

private_skills_host_path: Path

Path to private skills directory on host.

shared_skills_host_path property

shared_skills_host_path: Path

Path to shared skills directory on host.

private_images_host_path property

private_images_host_path: Path

Path to private images directory on host.

private_skills_container_path property

private_skills_container_path: str

Path to private skills directory in container.

shared_skills_container_path property

shared_skills_container_path: str

Path to shared skills directory in container.

private_mcp_container_path property

private_mcp_container_path: str

Path to private MCP directory in container.

CodeExecutionContainer

CodeExecutionContainer(tag: str, env: dict[str, str] | None = None, executor_port: int | None = None, resource_port: int | None = None, show_pull_progress: bool = True, workspace_path: Path | str | None = None, workspace_key: str | None = None)

Bases: ExecutionContainer

Context manager for the lifecycle of a code execution Docker container.

Extends ipybox's ExecutionContainer with workspace-specific bind mounts of skill directories.

Parameters:

Name	Type	Description	Default
tag	str	Name and optionally tag of the ipybox Docker image to use (format: name:tag)	required
env	dict[str, str] | None	Environment variables to set in the container	None
executor_port	int | None	Host port for the container's executor port. A random port is allocated if not specified.	None
resource_port	int | None	Host port for the container's resource port. A random port is allocated if not specified.	None
show_pull_progress	bool	Whether to show progress when pulling the Docker image.	True
workspace_path	Path | str | None	Path to workspace directory on host. Defaults to "workspace".	None
workspace_key	str | None	Key to designate private sub-directories on host. Defaults to "default".	None

Source code in freeact/environment.py

def __init__(
    self,
    tag: str,
    env: dict[str, str] | None = None,
    executor_port: int | None = None,
    resource_port: int | None = None,
    show_pull_progress: bool = True,
    workspace_path: Path | str | None = None,
    workspace_key: str | None = None,
):
    self._workspace = Workspace(workspace_path, workspace_key)

    binds = {
        self._workspace.private_skills_host_path: self._workspace.private_skills_container_path,
        self._workspace.shared_skills_host_path: self._workspace.shared_skills_container_path,
    }

    env = (env or {}) | {
        "PYTHONPATH": f".:/app/{self._workspace.shared_skills_container_path}:/app/{self._workspace.private_skills_container_path}",
    }

    super().__init__(
        binds=binds,
        tag=tag,
        env=env,
        executor_port=executor_port,
        resource_port=resource_port,
        show_pull_progress=show_pull_progress,
    )

workspace property

workspace: Workspace

The container's workspace.

CodeExecutionResult dataclass

CodeExecutionResult(text: str, images: Dict[Path, Image], is_error: bool)

Result of a code execution in a code executor.

text instance-attribute

text: str

Execution output text or error trace.

images instance-attribute

images: Dict[Path, Image]

Images generated during code execution. Keys are image file paths in the container.workspace, values are pre-loaded images from these files.

is_error instance-attribute

is_error: bool

Whether the execution resulted in an error. If True, text contains the corresponding error trace.

CodeExecution

CodeExecution(execution: Execution, images_dir: Path)

A code execution running in a code executor.

Source code in freeact/environment.py

def __init__(self, execution: Execution, images_dir: Path):
    self.execution = execution
    self.images_dir = images_dir
    self._result: CodeExecutionResult | None = None

result async

result(timeout: float = 120) -> CodeExecutionResult

Retrieves the complete result of this code execution. Waits until the result is available.

Parameters:

Name	Type	Description	Default
timeout	float	Maximum time in seconds to wait for the execution result	120

Raises:

Type	Description
TimeoutError	If code execution duration exceeds the specified timeout

Source code in freeact/environment.py

async def result(self, timeout: float = 120) -> CodeExecutionResult:
    """Retrieves the complete result of this code execution. Waits until the
    result is available.

    Args:
        timeout: Maximum time in seconds to wait for the execution result

    Raises:
        asyncio.TimeoutError: If code execution duration exceeds the specified timeout
    """
    if self._result is None:
        async for _ in self.stream(timeout=timeout):
            pass
    return self._result  # type: ignore

stream async

stream(timeout: float = 120) -> AsyncIterator[str]

Streams the code execution result as it is generated. Once the stream is consumed, a result is immediately available without waiting.

Generated images are not streamed. They can be obtained from the return value of result.

Parameters:

Name	Type	Description	Default
timeout	float	Maximum time in seconds to wait for the complete execution result	120

Raises:

Type	Description
TimeoutError	If code execution duration exceeds the specified timeout

Source code in freeact/environment.py

async def stream(self, timeout: float = 120) -> AsyncIterator[str]:
    """Streams the code execution result as it is generated. Once the stream
    is consumed, a [`result`][freeact.environment.CodeExecution.result] is
    immediately available without waiting.

    Generated images are not streamed. They can be obtained from the
    return value of [`result`][freeact.environment.CodeExecution.result].

    Args:
        timeout: Maximum time in seconds to wait for the complete execution result

    Raises:
        asyncio.TimeoutError: If code execution duration exceeds the specified timeout
    """
    images = {}

    try:
        async for chunk in self.execution.stream(timeout=timeout):
            yield chunk
    except ExecutionError as e:
        is_error = True
        text = e.trace
        yield text
    except asyncio.TimeoutError:
        is_error = True
        text = "Execution timed out"
        yield text
    else:
        result = await self.execution.result()
        text = result.text
        is_error = False

        if result.images:
            chunk = "\n\nProduced images:"
            yield chunk
            text += chunk

        for i, image in enumerate(result.images):
            path = await self._save_image(image)
            chunk = f"\n![image_{i}]({path})"
            yield chunk
            text += chunk
            images[path] = image

    self._result = CodeExecutionResult(text=text, images=images, is_error=is_error)

CodeExecutor

CodeExecutor(workspace: Workspace, port: int, host: str = 'localhost')

Context manager for executing code in an IPython kernel running in a CodeExecutionContainer. The kernel is created on entering the context and destroyed on exit.

Code execution is stateful for a given CodeExecutor instance. Definitions and variables of previous executions are available to subsequent executions.

Parameters:

Name	Type	Description	Default
workspace	Workspace	The workspace of the code execution container	required
port	int	Host port for the container's executor port	required
host	str	Hostname or IP address of the container's host	'localhost'

Source code in freeact/environment.py

def __init__(self, workspace: Workspace, port: int, host: str = "localhost"):
    self.workspace = workspace
    self.workspace.private_images_host_path.mkdir(parents=True, exist_ok=True)

    self._client = ExecutionClient(port=port, host=host)

execute async

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

Executes code in this executor's IPython kernel and returns the result.

Parameters:

Name	Type	Description	Default
code	str	Code to execute	required
timeout	float	Maximum time in seconds to wait for the execution result	120

Raises:

Type	Description
TimeoutError	If code execution duration exceeds the specified timeout

Source code in freeact/environment.py

async def execute(self, code: str, timeout: float = 120) -> CodeExecutionResult:
    """Executes code in this executor's IPython kernel and returns the result.

    Args:
        code: Code to execute
        timeout: Maximum time in seconds to wait for the execution result

    Raises:
        asyncio.TimeoutError: If code execution duration exceeds the specified timeout
    """
    code_exec = await self.submit(code)
    return await code_exec.result(timeout=timeout)

submit async

submit(code: str) -> CodeExecution

Submits code for execution in this executor's IPython kernel and returns a CodeExecution object for consuming the execution result.

Parameters:

Name	Type	Description	Default
code	str	Python code to execute	required

Returns:

Type	Description
CodeExecution	A CodeExecution object to track the code execution

Source code in freeact/environment.py

async def submit(self, code: str) -> CodeExecution:
    """Submits code for execution in this executor's IPython kernel and returns
    a [`CodeExecution`][freeact.environment.CodeExecution] object for consuming the
    execution result.

    Args:
        code: Python code to execute

    Returns:
        A [`CodeExecution`][freeact.environment.CodeExecution] object to track the code execution
    """
    code_exec = await self._client.submit(code)
    return CodeExecution(code_exec, self.workspace.private_images_host_path)

CodeProvider

CodeProvider(workspace: Workspace, port: int, host: str = 'localhost')

Context manager for

• loading the source code of Python modules and generated MCP client functions from a CodeExecutionContainer.
• registering MCP servers and generating client functions for their tools in a CodeExecutionContainer.

Source code loaded with this context manager is provided as skill sources to code action models so that they can include them into code actions.

Parameters:

Name	Type	Description	Default
workspace	Workspace	The workspace of the code execution container	required
port	int	Host port for the container's resource port	required
host	str	Hostname or IP address of the container's host	'localhost'

Source code in freeact/environment.py

def __init__(self, workspace: Workspace, port: int, host: str = "localhost"):
    self.workspace = workspace
    self._client = ResourceClient(port, host)

register_mcp_servers async

register_mcp_servers(server_params_dict: dict[str, dict[str, Any]]) -> dict[str, list[str]]

Registers MCP servers and generates Python client functions for their tools. These functions can be included into code actions, and calling them runs the corresponding MCP server tools. This works for both stdio and sse based MCP servers.

The source code of generated client functions can be loaded with the get_sources method.

Parameters:

Name	Type	Description	Default
server_params_dict	dict[str, dict[str, Any]]	Dictionary of application-defined server names and their MCP server parameters. stdio server parameters must specify at least a command key, sse server parameters must specify a url key. Application-defined server names must be valid Python module names.	required

Returns:

Type	Description
dict[str, list[str]]	Dictionary of server names and provided tool names. Tool names are sanitized
dict[str, list[str]]	to be valid Python module names.

Source code in freeact/environment.py

async def register_mcp_servers(self, server_params_dict: dict[str, dict[str, Any]]) -> dict[str, list[str]]:
    """Registers MCP servers and generates Python client functions for their tools. These
    functions can be included into code actions, and calling them runs the corresponding
    MCP server tools. This works for both `stdio` and `sse` based MCP servers.

    The source code of generated client functions can be loaded with the
    [`get_sources`][freeact.environment.CodeProvider.get_sources] method.

    Args:
        server_params_dict: Dictionary of application-defined server names and their MCP
            server parameters. `stdio` server parameters must specify at least a `command`
            key, `sse` server parameters must specify a `url` key. Application-defined
            server names must be valid Python module names.

    Returns:
        Dictionary of server names and provided tool names. Tool names are sanitized
        to be valid Python module names.
    """
    result = {}
    for server_name, server_params in server_params_dict.items():
        tool_names = await self._client.generate_mcp_sources(
            self.workspace.private_mcp_container_path, server_name, server_params
        )
        result[server_name] = tool_names
    return result

get_sources async

get_sources(module_names: list[str] | None = None, mcp_tool_names: Mapping[str, list[str] | None] | None = None) -> str

Loads the source code of given Python modules and generated MCP client functions and returns them in the following format:

```python
# Module: {module_name_1}
{module_source_1}
```

```python
# Module: {module_name_2}
{module_source_2}
```

...

Module names of generated MCP client functions follow the pattern mcpgen.{server_name}.{tool_name}. Hence, calling

await get_sources(mcp_tool_names={"my_server": ["my_tool"]})

is equivalent to

await get_sources(module_names=["mcpgen.my_server.my_tool"])

For loading the source code of all generated client functions for an MCP server, use None as value in the mcp_tool_names dictionary:

await get_sources(mcp_tool_names={"my_server": None})

Parameters:

Name	Type	Description	Default
module_names	list[str] | None	Names of modules available on the container's Python path	None
mcp_tool_names	Mapping[str, list[str] | None] | None	Dictionary of MCP server names and their tool names (as returned by register_mcp_servers). Values can be None which means all tool names for a given server name.	None

Returns:

Type	Description
str	The formatted source code of all requested Python modules and generated MCP client functions.

Source code in freeact/environment.py

async def get_sources(
    self,
    module_names: list[str] | None = None,
    mcp_tool_names: Mapping[str, list[str] | None] | None = None,
) -> str:
    """
    Loads the source code of given Python modules and generated MCP client functions
    and returns them in the following format:

        ```python
        # Module: {module_name_1}
        {module_source_1}
        ```

        ```python
        # Module: {module_name_2}
        {module_source_2}
        ```

        ...

    Module names of generated MCP client functions follow the pattern
    `mcpgen.{server_name}.{tool_name}`. Hence, calling

    ```python
    await get_sources(mcp_tool_names={"my_server": ["my_tool"]})
    ```

    is equivalent to

    ```python
    await get_sources(module_names=["mcpgen.my_server.my_tool"])
    ```

    For loading the source code of all generated client functions for an MCP server,
    use `None` as value in the `mcp_tool_names` dictionary:

    ```python
    await get_sources(mcp_tool_names={"my_server": None})
    ```

    Args:
        module_names: Names of modules available on the container's Python path
        mcp_tool_names: Dictionary of MCP server names and their tool names (as returned by
            [`register_mcp_servers`][freeact.environment.CodeProvider.register_mcp_servers]).
            Values can be `None` which means all tool names for a given server name.

    Returns:
        The formatted source code of all requested Python modules and generated MCP client functions.
    """
    mod_sources = await self._get_module_sources(module_names or [])
    mcp_sources = await self._get_mcp_sources(mcp_tool_names or {})
    return self._render(mod_sources | mcp_sources)

CodeExecutionEnvironment

CodeExecutionEnvironment(container: CodeExecutionContainer, host: str = 'localhost')

An environment for

• executing code actions in,
• loading source code from,
• and registering MCP servers at

a running CodeExecutionContainer.

Parameters:

Name	Type	Description	Default
container	CodeExecutionContainer	A running code execution container.	required

Source code in freeact/environment.py

def __init__(self, container: CodeExecutionContainer, host: str = "localhost"):
    self.container = container
    self.host = host

code_executor async

code_executor() -> AsyncIterator[CodeExecutor]

Context manager for CodeExecutors in this environment.

Source code in freeact/environment.py

@asynccontextmanager
async def code_executor(self) -> AsyncIterator[CodeExecutor]:
    """Context manager for [`CodeExecutor`][freeact.environment.CodeExecutor]s in this environment."""
    async with CodeExecutor(
        workspace=self.container.workspace,
        port=self.container.executor_port,
        host=self.host,
    ) as executor:
        yield executor

code_provider async

code_provider() -> AsyncIterator[CodeProvider]

Context manager for CodeProviders in this environment.

Source code in freeact/environment.py

@asynccontextmanager
async def code_provider(self) -> AsyncIterator[CodeProvider]:
    """Context manager for [`CodeProvider`][freeact.environment.CodeProvider]s in this environment."""
    async with CodeProvider(
        workspace=self.container.workspace,
        port=self.container.resource_port,
        host=self.host,
    ) as provider:
        yield provider

dotenv_variables

dotenv_variables(dotenv_path: Path | None = Path('.env'), export: bool = True, **kwargs) -> Dict[str, str]

Load environment variables from a .env file.

Reads environment variables from a .env file and optionally exports them to os.environ. If no path is provided, searches for a .env file in parent directories.

Parameters:

Name	Type	Description	Default
dotenv_path	Path | None	Path to the .env file. Defaults to .env in current directory.	Path('.env')
export	bool	Whether to export variables to current environment. Defaults to True.	True
**kwargs		Additional keyword arguments passed to DotEnv constructor.	{}

Returns:

Type	Description
Dict[str, str]	Dictionary mapping environment variable names to their values.

Source code in freeact/environment.py

def dotenv_variables(dotenv_path: Path | None = Path(".env"), export: bool = True, **kwargs) -> Dict[str, str]:
    """Load environment variables from a `.env` file.

    Reads environment variables from a `.env` file and optionally exports them to `os.environ`.
    If no path is provided, searches for a `.env` file in parent directories.

    Args:
        dotenv_path: Path to the `.env` file. Defaults to `.env` in current directory.
        export: Whether to export variables to current environment. Defaults to `True`.
        **kwargs: Additional keyword arguments passed to `DotEnv` constructor.

    Returns:
        Dictionary mapping environment variable names to their values.
    """

    if dotenv_path is None:
        dotenv_path = find_dotenv()

    dotenv = DotEnv(dotenv_path=dotenv_path, **kwargs)

    if export:
        dotenv.set_as_environment_variables()

    return {k: v for k, v in dotenv.dict().items() if v is not None}

execution_environment async

execution_environment(host: str = 'localhost', ipybox_tag: str = 'ghcr.io/gradion-ai/ipybox:minimal', ipybox_env: dict[str, str] = dotenv_variables(), executor_port: int | None = None, resource_port: int | None = None, workspace_path: Path | str | None = None, workspace_key: str | None = None) -> AsyncIterator[CodeExecutionEnvironment]

Context manager providing a CodeExecutionEnvironment. It manages the lifecycle of the environment's CodeExecutionContainer.

Parameters:

Name	Type	Description	Default
ipybox_tag	str	Name and optionally tag of the ipybox Docker image to use (format: name:tag)	'ghcr.io/gradion-ai/ipybox:minimal'
ipybox_env	dict[str, str]	Environment variables to set in the ipybox Docker container	dotenv_variables()
executor_port	int | None	Host port for the container's executor port. A random port is allocated if not specified	None
resource_port	int | None	Host port for the container's resource port. A random port is allocated if not specified	None
workspace_path	Path | str | None	Path to workspace directory on host. Defaults to "workspace".	None
workspace_key	str | None	Key to designate private sub-directories on host. Defaults to "default".	None

Source code in freeact/environment.py

@asynccontextmanager
async def execution_environment(
    host: str = "localhost",
    ipybox_tag: str = "ghcr.io/gradion-ai/ipybox:minimal",
    ipybox_env: dict[str, str] = dotenv_variables(),
    executor_port: int | None = None,
    resource_port: int | None = None,
    workspace_path: Path | str | None = None,
    workspace_key: str | None = None,
) -> AsyncIterator[CodeExecutionEnvironment]:
    """Context manager providing a [`CodeExecutionEnvironment`][freeact.environment.CodeExecutionEnvironment]. It
    manages the lifecycle of the environment's [`CodeExecutionContainer`][freeact.environment.CodeExecutionContainer].

    Args:
        ipybox_tag: Name and optionally tag of the `ipybox` Docker image to use (format: `name:tag`)
        ipybox_env: Environment variables to set in the `ipybox` Docker container
        executor_port: Host port for the container's executor port. A random port is allocated if not specified
        resource_port: Host port for the container's resource port. A random port is allocated if not specified
        workspace_path: Path to workspace directory on host. Defaults to "workspace".
        workspace_key: Key to designate private sub-directories on host. Defaults to "default".
    """
    async with CodeExecutionContainer(
        tag=ipybox_tag,
        env=ipybox_env,
        executor_port=executor_port,
        resource_port=resource_port,
        workspace_path=workspace_path,
        workspace_key=workspace_key,
    ) as container:
        yield CodeExecutionEnvironment(container=container, host=host)