Azure Container Code Executor

This guide will explain the Azure Container Apps dynamic sessions in Azure Container Apps and show you how to use the Azure Container Code Executor class.

Azure Container Apps dynamic sessions

The Azure Container Apps dynamic sessions is a component in the Azure Container Apps service. The environment is hosted on remote Azure instances and will not execute any code locally. The interpreter is capable of executing python code in a jupyter environment with a pre-installed base of commonly used packages. Custom environments can be created by users for their applications. Files can additionally be uploaded to, or downloaded from each session.

The code interpreter can run multiple sessions of code, each of which are delineated by a session identifier string.

Create a Container Apps Session Pool

In your Azure portal, create a new Container App Session Pool resource with the pool type set to Python code interpreter and note the Pool management endpoint. The format for the endpoint should be something like https://{region}.dynamicsessions.io/subscriptions/{subscription_id}/resourceGroups/{resource_group_name}/sessionPools/{session_pool_name}.

Alternatively, you can use the Azure CLI to create a session pool.

AzureContainerCodeExecutor

The AzureContainerCodeExecutor class is a python code executor that creates and executes arbitrary python code on a default Serverless code interpreter session. Its interface is as follows

Initialization

First, you will need to find or create a credentialing object that implements the TokenProvider interface. This is any object that implements the following function

def get_token(
    self, *scopes: str, claims: Optional[str] = None, tenant_id: Optional[str] = None, **kwargs: Any
) -> azure.core.credentials.AccessToken

An example of such an object is the azure.identity.DefaultAzureCredential class.

Lets start by installing that

# pip install azure.identity

Next, lets import all the necessary modules and classes for our code

import os
import tempfile

from anyio import open_file
from autogen_core.base import CancellationToken
from autogen_core.components.code_executor import CodeBlock
from autogen_ext.code_executor.aca_dynamic_sessions import AzureContainerCodeExecutor
from azure.identity import DefaultAzureCredential

Now, we create our Azure code executor and run some test code along with verification that it ran correctly. We’ll create the executor with a temporary working directory to ensure a clean environment as we show how to use each feature

cancellation_token = CancellationToken()
POOL_MANAGEMENT_ENDPOINT = "..."

with tempfile.TemporaryDirectory() as temp_dir:
    executor = AzureContainerCodeExecutor(
        pool_management_endpoint=POOL_MANAGEMENT_ENDPOINT, credential=DefaultAzureCredential(), work_dir=temp_dir
    )

    code_blocks = [CodeBlock(code="import sys; print('hello world!')", language="python")]
    code_result = await executor.execute_code_blocks(code_blocks, cancellation_token)
    assert code_result.exit_code == 0 and "hello world!" in code_result.output

Next, lets try uploading some files and verifying their integrity. All files uploaded to the Serverless code interpreter is uploaded into the /mnt/data directory. All downloadable files must also be placed in the directory. By default, the current working directory for the code executor is set to /mnt/data.

with tempfile.TemporaryDirectory() as temp_dir:
    test_file_1 = "test_upload_1.txt"
    test_file_1_contents = "test1 contents"
    test_file_2 = "test_upload_2.txt"
    test_file_2_contents = "test2 contents"

    async with await open_file(os.path.join(temp_dir, test_file_1), "w") as f:  # type: ignore[syntax]
        await f.write(test_file_1_contents)
    async with await open_file(os.path.join(temp_dir, test_file_2), "w") as f:  # type: ignore[syntax]
        await f.write(test_file_2_contents)

    assert os.path.isfile(os.path.join(temp_dir, test_file_1))
    assert os.path.isfile(os.path.join(temp_dir, test_file_2))

    executor = AzureContainerCodeExecutor(
        pool_management_endpoint=POOL_MANAGEMENT_ENDPOINT, credential=DefaultAzureCredential(), work_dir=temp_dir
    )
    await executor.upload_files([test_file_1, test_file_2], cancellation_token)

    file_list = await executor.get_file_list(cancellation_token)
    assert test_file_1 in file_list
    assert test_file_2 in file_list

    code_blocks = [
        CodeBlock(
            code=f"""
with open("{test_file_1}") as f:
  print(f.read())
with open("{test_file_2}") as f:
  print(f.read())
""",
            language="python",
        )
    ]
    code_result = await executor.execute_code_blocks(code_blocks, cancellation_token)
    assert code_result.exit_code == 0
    assert test_file_1_contents in code_result.output
    assert test_file_2_contents in code_result.output

Downloading files works in a similar way.

with tempfile.TemporaryDirectory() as temp_dir:
    test_file_1 = "test_upload_1.txt"
    test_file_1_contents = "test1 contents"
    test_file_2 = "test_upload_2.txt"
    test_file_2_contents = "test2 contents"

    assert not os.path.isfile(os.path.join(temp_dir, test_file_1))
    assert not os.path.isfile(os.path.join(temp_dir, test_file_2))

    executor = AzureContainerCodeExecutor(
        pool_management_endpoint=POOL_MANAGEMENT_ENDPOINT, credential=DefaultAzureCredential(), work_dir=temp_dir
    )

    code_blocks = [
        CodeBlock(
            code=f"""
with open("{test_file_1}", "w") as f:
  f.write("{test_file_1_contents}")
with open("{test_file_2}", "w") as f:
  f.write("{test_file_2_contents}")
""",
            language="python",
        ),
    ]
    code_result = await executor.execute_code_blocks(code_blocks, cancellation_token)
    assert code_result.exit_code == 0

    file_list = await executor.get_file_list(cancellation_token)
    assert test_file_1 in file_list
    assert test_file_2 in file_list

    await executor.download_files([test_file_1, test_file_2], cancellation_token)

    assert os.path.isfile(os.path.join(temp_dir, test_file_1))
    async with await open_file(os.path.join(temp_dir, test_file_1), "r") as f:  # type: ignore[syntax]
        content = await f.read()
        assert test_file_1_contents in content
    assert os.path.isfile(os.path.join(temp_dir, test_file_2))
    async with await open_file(os.path.join(temp_dir, test_file_2), "r") as f:  # type: ignore[syntax]
        content = await f.read()
        assert test_file_2_contents in content

New Sessions

Every instance of the AzureContainerCodeExecutor class will have a unique session ID. Every call to a particular code executor will be executed on the same session until the restart() function is called on it. Previous sessions cannot be reused.

Here we’ll run some code on the code session, restart it, then verify that a new session has been opened.

executor = AzureContainerCodeExecutor(
    pool_management_endpoint=POOL_MANAGEMENT_ENDPOINT, credential=DefaultAzureCredential()
)

code_blocks = [CodeBlock(code="x = 'abcdefg'", language="python")]
code_result = await executor.execute_code_blocks(code_blocks, cancellation_token)
assert code_result.exit_code == 0

code_blocks = [CodeBlock(code="print(x)", language="python")]
code_result = await executor.execute_code_blocks(code_blocks, cancellation_token)
assert code_result.exit_code == 0 and "abcdefg" in code_result.output

await executor.restart()
code_blocks = [CodeBlock(code="print(x)", language="python")]
code_result = await executor.execute_code_blocks(code_blocks, cancellation_token)
assert code_result.exit_code != 0 and "NameError" in code_result.output

Available Packages

Each code execution instance is pre-installed with most of the commonly used packages. However, the list of available packages and versions are not available outside of the execution environment. The packages list on the environment can be retrieved by calling the get_available_packages() function on the code executor.

print(executor.get_available_packages(cancellation_token))