Azure Container Apps Dynamic Sessions: Secure Code Execution for AI Agents

AI agents that can write and execute code introduce significant security risks—from data exfiltration to resource abuse. Azure Container Apps Dynamic Sessions provides a solution: ephemeral, sandboxed execution environments that isolate agent-generated code from your production infrastructure. This comprehensive guide explores how to implement secure code execution for AI code interpreters, automated testing agents, and data analysis workflows.

The Code Execution Security Problem

When AI agents generate and execute code, you face critical security challenges:

• Arbitrary code execution: LLMs can generate malicious code, intentionally or through prompt injection
• Data exfiltration: Generated code may attempt to access and transmit sensitive data
• Resource abuse: Infinite loops, memory bombs, or cryptocurrency mining
• Lateral movement: Code accessing other services through network connectivity
• Persistence: Malicious code attempting to survive beyond its intended session

Dynamic Sessions solve these by providing completely isolated, ephemeral environments that are destroyed after each use.

Dynamic Sessions Architecture

graph TB
    subgraph Agent ["AI Agent Application"]
        LLM["LLM / Agent"]
        CodeGen["Code Generator"]
        SessionMgr["Session Manager"]
    end
    
    subgraph ACA ["Azure Container Apps"]
        API["Agent API"]
        SessionPool["Session Pool"]
    end
    
    subgraph Sessions ["Dynamic Sessions (Isolated)"]
        S1["Session 1
Python Runtime"]
        S2["Session 2
Python Runtime"]
        S3["Session 3
Node.js Runtime"]
    end
    
    subgraph Security ["Security Boundary"]
        Firewall["Network Isolation"]
        ResourceLimits["CPU/Memory Limits"]
        TimeLimit["Execution Timeout"]
    end
    
    LLM --> CodeGen
    CodeGen --> SessionMgr
    SessionMgr --> API
    API --> SessionPool
    SessionPool --> S1
    SessionPool --> S2
    SessionPool --> S3
    
    S1 --> Firewall
    S2 --> Firewall
    S3 --> Firewall
    S1 --> ResourceLimits
    S2 --> ResourceLimits
    S3 --> ResourceLimits
    
    style Sessions fill:#E8F5E9,stroke:#2E7D32
    style Security fill:#FFCDD2,stroke:#C62828
    style SessionPool fill:#E3F2FD,stroke:#1565C0

Key Features

Setting Up Dynamic Sessions

# Create a Container Apps Environment with Dynamic Sessions enabled
az containerapp env create   --name aca-agents-env   --resource-group rg-ai-agents   --location eastus   --enable-dynamic-sessions

# Create a session pool for Python code execution
az containerapp sessionpool create   --name python-code-pool   --resource-group rg-ai-agents   --environment aca-agents-env   --container-type PythonLTS   --max-sessions 100   --ready-session-instances 10   --cooldown-period 300

# Create a session pool for custom runtime
az containerapp sessionpool create   --name custom-runtime-pool   --resource-group rg-ai-agents   --environment aca-agents-env   --container-type CustomContainer   --image myregistry.azurecr.io/code-sandbox:latest   --registry-server myregistry.azurecr.io   --max-sessions 50

Executing Code in Dynamic Sessions

Python SDK Integration

from azure.containerapp.sessions import SessionPoolClient
from azure.identity import DefaultAzureCredential
import asyncio

class SecureCodeExecutor:
    """Execute AI-generated code in isolated Azure Container Apps sessions."""
    
    def __init__(self, pool_endpoint: str):
        self.credential = DefaultAzureCredential()
        self.client = SessionPoolClient(
            endpoint=pool_endpoint,
            credential=self.credential
        )
    
    async def execute_python(
        self,
        code: str,
        timeout_seconds: int = 30,
        memory_mb: int = 512,
        allowed_packages: list[str] | None = None
    ) -> dict:
        """Execute Python code in an isolated session."""
        
        # Create a new session
        session = await self.client.create_session(
            identifier=f"agent-{uuid.uuid4()}",
            properties={
                "timeout": timeout_seconds,
                "memory_mb": memory_mb,
                "cpu_cores": 0.5,
                "network_access": "none"  # No internet access
            }
        )
        
        try:
            # Install allowed packages if specified
            if allowed_packages:
                install_code = f"import subprocess; subprocess.run(['pip', 'install', '-q', {', '.join(repr(p) for p in allowed_packages)}])"
                await session.execute(install_code)
            
            # Execute the agent-generated code
            result = await session.execute(
                code,
                timeout=timeout_seconds
            )
            
            return {
                "success": True,
                "output": result.stdout,
                "error": result.stderr,
                "return_value": result.return_value,
                "execution_time_ms": result.execution_time_ms
            }
            
        except TimeoutError:
            return {
                "success": False,
                "error": "Execution timed out",
                "output": None
            }
        except Exception as e:
            return {
                "success": False,
                "error": str(e),
                "output": None
            }
        finally:
            # Session is automatically destroyed
            await session.close()
    
    async def execute_with_files(
        self,
        code: str,
        input_files: dict[str, bytes],
        timeout_seconds: int = 60
    ) -> dict:
        """Execute code with file inputs and retrieve output files."""
        
        session = await self.client.create_session(
            properties={"timeout": timeout_seconds}
        )
        
        try:
            # Upload input files
            for filename, content in input_files.items():
                await session.upload_file(filename, content)
            
            # Execute code
            result = await session.execute(code)
            
            # Download output files
            output_files = {}
            for filename in await session.list_files():
                if filename not in input_files:
                    output_files[filename] = await session.download_file(filename)
            
            return {
                "success": True,
                "output": result.stdout,
                "files": output_files
            }
        finally:
            await session.close()

.NET SDK Integration

using Azure.ContainerApps.Sessions;
using Azure.Identity;

public class SecureCodeExecutor
{
    private readonly SessionPoolClient _client;
    
    public SecureCodeExecutor(string poolEndpoint)
    {
        _client = new SessionPoolClient(
            new Uri(poolEndpoint),
            new DefaultAzureCredential()
        );
    }
    
    public async Task<CodeExecutionResult> ExecutePythonAsync(
        string code,
        TimeSpan timeout,
        CancellationToken cancellationToken = default)
    {
        // Create an isolated session
        var session = await _client.CreateSessionAsync(
            new SessionProperties
            {
                ContainerType = ContainerType.PythonLTS,
                Timeout = timeout,
                MemoryMB = 512,
                NetworkAccess = NetworkAccess.None
            },
            cancellationToken);
        
        try
        {
            var result = await session.ExecuteCodeAsync(
                code,
                cancellationToken);
            
            return new CodeExecutionResult
            {
                Success = result.ExitCode == 0,
                StandardOutput = result.StandardOutput,
                StandardError = result.StandardError,
                ExecutionTimeMs = result.ExecutionTime.TotalMilliseconds
            };
        }
        catch (SessionTimeoutException)
        {
            return new CodeExecutionResult
            {
                Success = false,
                StandardError = "Execution timed out"
            };
        }
        finally
        {
            await session.DisposeAsync();
        }
    }
}

public record CodeExecutionResult
{
    public bool Success { get; init; }
    public string? StandardOutput { get; init; }
    public string? StandardError { get; init; }
    public double ExecutionTimeMs { get; init; }
}

Integrating with AI Agents

from openai import AzureOpenAI
import json

class CodeInterpreterAgent:
    """AI agent with secure code execution capabilities."""
    
    def __init__(self, openai_client: AzureOpenAI, code_executor: SecureCodeExecutor):
        self.llm = openai_client
        self.executor = code_executor
        
        self.system_prompt = """You are a helpful data analysis assistant. 
When asked to analyze data or perform calculations, write Python code to accomplish the task.
Your code will be executed in a secure sandbox environment.

Rules:
- Use pandas for data manipulation
- Use matplotlib for visualizations (save to 'output.png')
- Print results to stdout
- The sandbox has no internet access
- Available packages: pandas, numpy, matplotlib, scikit-learn"""
    
    async def analyze(self, user_request: str, data_files: dict[str, bytes]) -> dict:
        """Process a user request with code execution."""
        
        # Step 1: Generate code with LLM
        response = self.llm.chat.completions.create(
            model="gpt-5",
            messages=[
                {"role": "system", "content": self.system_prompt},
                {"role": "user", "content": f"Data files available: {list(data_files.keys())}

Request: {user_request}"}
            ],
            tools=[{
                "type": "function",
                "function": {
                    "name": "execute_python",
                    "description": "Execute Python code in a secure sandbox",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "code": {"type": "string", "description": "Python code to execute"}
                        },
                        "required": ["code"]
                    }
                }
            }]
        )
        
        # Step 2: Extract and validate code
        tool_call = response.choices[0].message.tool_calls[0]
        code = json.loads(tool_call.function.arguments)["code"]
        
        # Step 3: Execute in secure sandbox
        result = await self.executor.execute_with_files(
            code=code,
            input_files=data_files,
            timeout_seconds=60
        )
        
        # Step 4: Generate natural language response
        if result["success"]:
            final_response = self.llm.chat.completions.create(
                model="gpt-5",
                messages=[
                    {"role": "system", "content": "Summarize the code execution results for the user."},
                    {"role": "user", "content": f"Code output:
{result['output']}

Original request: {user_request}"}
                ]
            )
            
            return {
                "answer": final_response.choices[0].message.content,
                "code": code,
                "output": result["output"],
                "files": result.get("files", {})
            }
        else:
            return {
                "answer": f"Code execution failed: {result['error']}",
                "code": code,
                "error": result["error"]
            }

Network Access Control

# Configure network allowlist for sessions that need limited external access
session = await client.create_session(
    properties={
        "network_access": "restricted",
        "allowed_hosts": [
            "api.github.com",
            "pypi.org",
            "*.blob.core.windows.net"  # Azure Storage only
        ],
        "blocked_ports": [22, 23, 3389],  # Block SSH, Telnet, RDP
        "max_outbound_connections": 10
    }
)

# For data analysis with Azure resources, use managed identity
session = await client.create_session(
    properties={
        "network_access": "azure_only",  # Only Azure services
        "managed_identity": {
            "enabled": True,
            "client_id": "your-managed-identity-client-id"
        }
    }
)

# Code in session can now access Azure resources securely
code = """
from azure.identity import DefaultAzureCredential
from azure.storage.blob import BlobServiceClient

credential = DefaultAzureCredential()  # Uses session's managed identity
client = BlobServiceClient(account_url="https://mystorage.blob.core.windows.net", credential=credential)

# Read data from blob storage
blob = client.get_blob_client("data", "sales.csv")
data = blob.download_blob().readall()
print(f"Downloaded {len(data)} bytes")
"""

Monitoring and Observability

// Enable telemetry for session execution
services.AddOpenTelemetry()
    .WithTracing(builder =>
    {
        builder
            .AddSource("Azure.ContainerApps.Sessions")
            .AddAzureMonitorTraceExporter();
    })
    .WithMetrics(builder =>
    {
        builder
            .AddMeter("Azure.ContainerApps.Sessions")
            .AddAzureMonitorMetricExporter();
    });

// Custom logging for audit trail
public class AuditingCodeExecutor : SecureCodeExecutor
{
    private readonly ILogger<AuditingCodeExecutor> _logger;
    
    public override async Task<CodeExecutionResult> ExecutePythonAsync(
        string code,
        TimeSpan timeout,
        CancellationToken cancellationToken = default)
    {
        var executionId = Guid.NewGuid().ToString();
        
        _logger.LogInformation(
            "Code execution started. ExecutionId: {ExecutionId}, CodeHash: {CodeHash}, Timeout: {Timeout}",
            executionId,
            ComputeHash(code),
            timeout);
        
        var stopwatch = Stopwatch.StartNew();
        var result = await base.ExecutePythonAsync(code, timeout, cancellationToken);
        stopwatch.Stop();
        
        _logger.LogInformation(
            "Code execution completed. ExecutionId: {ExecutionId}, Success: {Success}, Duration: {Duration}ms",
            executionId,
            result.Success,
            stopwatch.ElapsedMilliseconds);
        
        // Store full audit record
        await _auditStore.RecordExecutionAsync(new ExecutionAuditRecord
        {
            ExecutionId = executionId,
            Code = code,  // Consider encrypting for sensitive environments
            Result = result,
            Timestamp = DateTimeOffset.UtcNow,
            UserId = _currentUser.Id
        });
        
        return result;
    }
}

Pricing and Optimization

Use Cases

• AI Code Interpreters: ChatGPT-style code execution for data analysis and visualization
• Automated Testing: Execute user-submitted test code against APIs or libraries
• Data Processing Agents: Run ETL scripts generated by AI on uploaded datasets
• Educational Platforms: Execute student code submissions safely
• Low-Code Platforms: Run custom scripts defined by end users
• Security Research: Safely analyze potentially malicious code samples

Key Takeaways

• Azure Container Apps Dynamic Sessions provide ephemeral, isolated environments for executing untrusted code.
• Pre-warmed session pools enable <100ms cold start for responsive AI agent experiences.
• Network isolation by default prevents data exfiltration; allowlists enable controlled access.
• Managed identity integration allows secure access to Azure resources without exposing credentials.
• Comprehensive auditing is essential for compliance and debugging AI agent behavior.

Conclusion

As AI agents become more capable of generating and executing code, secure execution environments become non-negotiable. Azure Container Apps Dynamic Sessions provides the isolation, performance, and integration needed for production AI systems. By combining ephemeral containers, network isolation, resource limits, and comprehensive monitoring, you can enable powerful code execution capabilities while maintaining security boundaries. For any application where AI-generated code touches production data, Dynamic Sessions should be your default execution environment.

References

• Azure Container Apps Dynamic Sessions Documentation
• Code Interpreter Session Tutorial
• Microsoft Tech Community: Secure Code Execution
• Dynamic Sessions Code Samples