For any call to ARM (management.azure.com), use the access token retrieved from the below call:

az account get-access-token --resource <https://management.azure.com/>

a. Call the following API to generate a Session Pool:

PUT <https://management.azure.com/subscriptions/{{SubscriptionId}}/resourceGroups/{{ResourceGroup}}/providers/Microsoft.App/sessionPools/{{SessionPoolName}}?api-version=2023-08-01-preview>

Body properties:

• location: Azure Region
• properties:
  • poolManagementType:
    • Today there are two Pool Management Types supported: - "Manual" - In this model, the user will call generateSessions API which supports batch mode (to generate 100s of sessions in one API call, and then user is free to update/specialize the session as needed or execute code in the session) - "Dynamic" - In this mode, the pool management is handled by the platform. Currently, the dynamic mode is only implemented for Python code execution scenario, which has its own APIs to execute code.
  • maxConcurrentSessions:
    • Maximum number of active sessions allowed
  • name:
    • Name of the sessions pool
  • dynamicPoolConfiguration: Specifies the type of sessions generated by the platform
    • poolType: Type of images used for the pool
      • Valid values ["JupyterPython", "FunctionsPython"]
    • executionType:
      • Valid values ["Timed"]
    • coolDownPeriodSeconds:
      • Integer representing the maximum time allowed before the platform scales down the container
  • sessionPoolSecrets: Secrets associated with the Session Pool
    • name: Name of the secret
    • value: Secret Value

Example Generation of Session Pool:

{
    "location": "koreacentral",
    "properties": {
        "poolManagementType": "Dynamic",
        "maxConcurrentSessions": 10,
        "name": "{{SessionPoolName}}",
        "dynamicPoolConfiguration": {
            "poolType": "JupyterPython",
            "executionType": "Timed",
            "coolDownPeriodInSecond": 310
        }
    }
}

Curl Example:

curl -X PUT "https://management.azure.com/subscriptions/{{SubscriptionId}}/resourceGroups/{{ResourceGroup}}/providers/Microsoft.App/sessionPools/{{SessionPoolName}}?api-version=2023-08-01-preview" \
    -H "Content-Type: application/json" \
    -H "Authorization: Bearer $token" \
    -d '{"location": "koreacentral","properties": { "poolManagementType": "Dynamic", "maxConcurrentSessions": 10, "name": "{{SessionPoolName}}", "dynamicPoolConfiguration": { "poolType": "JupyterPython", "executionType": "Timed", "coolDownPeriodInSecond": 310 } } }'

If all goes well, you should receive a 200 Status Code. The response will contain a poolManagementEndpoint which is required to configure the Python Plugin below.

To successfully use the Python Plugin in Semantic Kernel, you must install the Poetry azure extras by running poetry install -E azure.

Next, in the .env file, add the poolManagementEndpoint value from above to the variable ACA_POOL_MANAGEMENT_ENDPOINT. The poolManagementEndpoint should look something like:

https://eastus.acasessions.io/subscriptions/{{subscriptionId}}/resourceGroups/{{resourceGroup}}/sessionPools/{{sessionPool}}/python/execute

It is possible to add the code interpreter plugin as follows:

kernel = Kernel()

service_id = "azure_oai"
chat_service = AzureChatCompletion(
    service_id=service_id, **azure_openai_settings_from_dot_env_as_dict(include_api_version=True)
)
kernel.add_service(chat_service)

python_code_interpreter = SessionsPythonTool(
    auth_callback=auth_callback
)

sessions_tool = kernel.add_plugin(python_code_interpreter, "PythonCodeInterpreter")

code = "import json\n\ndef add_numbers(a, b):\n    return a + b\n\nargs = '{\"a\": 1, \"b\": 1}'\nargs_dict = json.loads(args)\nprint(add_numbers(args_dict['a'], args_dict['b']))"
result = await kernel.invoke(sessions_tool["execute_code"], code=code)

print(result)

Instead of hard-coding a well-formatted Python code string, you may use automatic function calling inside of SK and allow the model to form the Python and call the plugin.

The authentication callback must return a valid token for the session pool. One possible way of doing this with a DefaultAzureCredential is as follows:

async def auth_callback() -> str:
    """Auth callback for the SessionsPythonTool.
    This is a sample auth callback that shows how to use Azure's DefaultAzureCredential
    to get an access token.
    """
    global auth_token
    current_utc_timestamp = int(datetime.datetime.now(datetime.timezone.utc).timestamp())

    if not auth_token or auth_token.expires_on < current_utc_timestamp:
        credential = DefaultAzureCredential()

        try:
            auth_token = credential.get_token(ACA_TOKEN_ENDPOINT)
        except ClientAuthenticationError as cae:
            err_messages = getattr(cae, "messages", [])
            raise FunctionExecutionException(
                f"Failed to retrieve the client auth token with messages: {' '.join(err_messages)}"
            ) from cae

    return auth_token.token

Currently, there are two concept examples that show this plugin in more detail:

• Plugin example: shows the basic usage of calling the code execute function on the plugin.
• Function Calling example: shows a simple chat application that leverages the Python code interpreter plugin for function calling.