What Are AutoGen Code Executors?
AutoGen Code Executors are specialized components within Microsoft's AutoGen framework that handle the execution of code generated by AI agents. When an LLM-powered agent produces Python, shell scripts, or other executable code, the Code Executor takes that code and runs it in a controlled, isolated environment. This separation between code generation and code execution is a fundamental architectural decision that makes agent-based systems safer and more predictable.
At its core, a Code Executor accepts code as input, executes it, and returns the output — including stdout, stderr, and any generated files — back to the agent or workflow. AutoGen provides multiple executor implementations out of the box, each offering different trade-offs between convenience, security, and performance.
Key Code Executor Implementations
- LocalCommandLineCodeExecutor — Executes code directly on the host machine in a designated workspace directory. Fast and simple, but offers limited isolation.
- DockerCodeExecutor — Runs code inside a Docker container, providing strong filesystem and process isolation from the host.
- AzureContainerCodeExecutor — Similar to Docker executor but leverages Azure Container Instances for cloud-scale, fully managed sandboxing.
- PythonCodeExecutionCapability — A lighter-weight option for executing Python specifically, often used within a single agent's tool set.
Why Safe Sandboxing Matters
AI agents that can write and execute code autonomously pose unique security risks. An agent tasked with data analysis might generate a script that accidentally deletes files, consumes excessive resources, or leaks sensitive data. A malicious prompt injection could cause an agent to execute arbitrary system commands. Without proper sandboxing, generated code runs with the same privileges as the agent process — which could be disastrous in production environments.
Core Risks Without Sandboxing
- Filesystem damage — Generated code can read, modify, or delete arbitrary files on the host.
- Network exposure — Code can make outbound connections to exfiltrate data or download payloads.
- Resource exhaustion — Infinite loops or memory-intensive operations can crash the entire agent system.
- Privilege escalation — Code executed with host-level permissions can access system configurations and secrets.
Sandboxing creates a controlled boundary where code can run with restricted capabilities. A well-configured sandbox limits filesystem access to a specific workspace, blocks or restricts network access, enforces resource limits, and prevents interference with the host system. This allows agents to be productive while keeping the blast radius of any mishap contained.
Setting Up AutoGen Code Executors
Let's walk through a complete setup, starting from installation and progressing through each executor type. All examples assume Python 3.10+ and a working AutoGen installation.
Installation
pip install pyautogen
# For Docker-based executors, ensure Docker is installed and running
# docker --version # verify Docker availability1. LocalCommandLineCodeExecutor — The Quick Start Option
The local executor runs code in a subprocess on your machine, confined to a specified workspace directory. It's ideal for prototyping and trusted environments where the agent operates on pre-vetted datasets.
import os
import tempfile
from autogen.coding import LocalCommandLineCodeExecutor
# Create a temporary workspace directory
workspace = tempfile.mkdtemp()
print(f"Workspace directory: {workspace}")
# Initialize the executor
executor = LocalCommandLineCodeExecutor(
timeout=60, # Max execution time in seconds
work_dir=workspace, # Confines file operations to this directory
code_functions=[], # Optional: whitelist of allowed functions
code_imports=["numpy", "pandas", "matplotlib"] # Allowed imports
)
# Example: Execute a simple Python code block
code = """
import pandas as pd
import os
# Create a sample CSV
data = {'name': ['Alice', 'Bob'], 'age': [25, 30]}
df = pd.DataFrame(data)
df.to_csv('output.csv', index=False)
print(f"File created at: {os.path.abspath('output.csv')}")
print(df.describe())
"""
result = executor.execute_code_blocks(
code_blocks=[("code", code)]
)
print("Output:", result.output)
print("Generated files:", result.code_file) if hasattr(result, 'code_file') else NoneThe local executor creates a Python file in the workspace, runs it as a subprocess, captures stdout/stderr, and returns the results. The workspace directory acts as a soft sandbox — files written outside it are possible but the executor encourages containment through the working directory context.
2. DockerCodeExecutor — Strong Isolation
The Docker executor provides true container-level isolation. Every code execution gets its own container instance (or reuses a pooled one), ensuring complete filesystem separation, network control, and resource limiting.
from autogen.coding import DockerCodeExecutor
from autogen.coding.docker_commandline_code_executor import DockerCommandLineCodeExecutor
# Docker executor with custom image
executor = DockerCommandLineCodeExecutor(
image="python:3.11-slim", # Base Docker image
timeout=120, # Execution timeout
work_dir="/workspace", # Container-side workspace path
bind_dir="/tmp/autogen_workspace", # Host directory to bind-mount
# Environment variables inside the container
env_vars={"PYTHONUNBUFFERED": "1"},
# Network policy: "none" blocks all networking
network="none",
# Resource constraints
mem_limit="512m",
cpu_limit=1.0,
# Auto-remove container after execution
auto_remove=True
)
# Execute code inside the container
code = """
import sys
import json
import hashlib
# Verify Python version
print(f"Python version: {sys.version}")
# Compute something that proves we're isolated
data = {"status": "ok", "container": True}
hash_val = hashlib.sha256(json.dumps(data).encode()).hexdigest()
print(f"Hash: {hash_val}")
# Write output file
with open('/workspace/result.json', 'w') as f:
json.dump(data, f)
"""
result = executor.execute_code_blocks([("code", code)])
print("Container output:", result.output)
# Check for generated files in the bind-mounted host directory
import glob
print("Files in bind mount:", glob.glob("/tmp/autogen_workspace/*"))The Docker executor builds a temporary container, copies (or bind-mounts) the workspace, executes the code, and captures everything before cleaning up. Setting network="none" is a critical security measure — it prevents any outbound connections. For agents that need API access, you can use "bridge" networking with firewall rules instead.
3. Integrating Code Executors with AutoGen Agents
The real power comes when you wire the executor into an AutoGen agent workflow. Here's a complete two-agent setup where an assistant writes code and a code-executing agent runs it safely.
import os
from autogen import AssistantAgent, UserProxyAgent, ConversableAgent
from autogen.coding import LocalCommandLineCodeExecutor
# Create workspace
workspace = os.path.join(os.getcwd(), "agent_workspace")
os.makedirs(workspace, exist_ok=True)
# Configure the code executor
executor = LocalCommandLineCodeExecutor(
timeout=90,
work_dir=workspace,
code_imports=["pandas", "numpy", "matplotlib", "requests"]
)
# Create a code-executing agent that uses the executor
code_executor_agent = UserProxyAgent(
name="CodeExecutor",
human_input_mode="NEVER",
code_execution_config={
"executor": executor,
"last_n_messages": 3,
"work_dir": workspace
}
)
# Create an assistant agent that generates code
assistant = AssistantAgent(
name="Assistant",
system_message="""You are a data analyst. When asked to perform analysis,
write complete, executable Python code. Your code should be self-contained,
handle errors gracefully, and produce clear output. Save any plots or
results files to the current working directory.""",
llm_config={
"model": "gpt-4",
"temperature": 0.2
}
)
# Initiate the conversation
response = code_executor_agent.initiate_chat(
assistant,
message="""
Analyze the relationship between variables in this dataset.
Create a CSV file 'synthetic_data.csv' with 100 rows containing:
- id: sequential integer
- category: random choice from ['A', 'B', 'C']
- value: normally distributed number (mean=50, std=10)
- timestamp: random datetime in 2024
Then compute summary statistics by category and save the results
as 'summary_by_category.csv'.
"""
)
print("Conversation completed.")
print("Check workspace for output files:", workspace)This pattern creates a powerful loop: the assistant writes code, the executor runs it, the output feeds back to the assistant for refinement. The executor acts as the safety gatekeeper for every code execution cycle.
4. Advanced: Custom Code Execution Policies
AutoGen allows you to define execution policies that control which code gets executed. This is a critical defense-in-depth layer. You can filter code blocks before they reach the executor.
from autogen.coding import CodeBlock, CodeExtractor, MarkdownCodeExtractor
from typing import List
class RestrictedCodePolicy:
"""Custom policy that blocks dangerous patterns before execution."""
FORBIDDEN_PATTERNS = [
"os.system", "subprocess.call", "eval(", "exec(",
"__import__", "open('/etc/", "shutil.rmtree",
"socket.", "http.client", "requests.delete",
"!rm -rf", "!sudo", "!chmod 777"
]
ALLOWED_IMPORTS = {
"pandas", "numpy", "matplotlib", "scipy",
"json", "csv", "datetime", "math", "statistics",
"collections", "itertools", "typing", "pathlib"
}
def filter_code(self, code_blocks: List[CodeBlock]) -> List[CodeBlock]:
"""Filter out dangerous code blocks."""
safe_blocks = []
for block in code_blocks:
code_str = block.code.lower()
is_safe = True
for pattern in self.FORBIDDEN_PATTERNS:
if pattern.lower() in code_str:
print(f"BLOCKED: Found forbidden pattern '{pattern}'")
is_safe = False
break
if is_safe:
safe_blocks.append(block)
else:
print(f"Skipping dangerous code block")
return safe_blocks
# Use the policy with an executor
policy = RestrictedCodePolicy()
# In a real workflow, filter before execution
raw_code = """
# This code contains a blocked pattern
import pandas as pd
df = pd.DataFrame({'x': [1,2,3]})
os.system('rm -rf /') # This would be caught
print(df)
"""
extractor = MarkdownCodeExtractor()
blocks = extractor.extract_code_blocks(raw_code)
filtered_blocks = policy.filter_code(blocks)
if filtered_blocks:
result = executor.execute_code_blocks(filtered_blocks)
print("Safe execution completed:", result.output)
else:
print("All code blocks were filtered out")Best Practices for Production-Grade Sandboxing
1. Always Use Container-Based Isolation in Production
Local executors are fine for development, but production workloads should always use Docker or Azure Container-based executors. Containers provide genuine isolation boundaries that cannot be easily bypassed by cleverly generated code. The overhead is minimal — container startup typically takes under a second with cached images.
# Production-ready Docker executor configuration
from autogen.coding import DockerCommandLineCodeExecutor
production_executor = DockerCommandLineCodeExecutor(
image="python:3.11-slim",
timeout=300,
network="none", # No network access
mem_limit="256m", # Strict memory cap
cpu_limit=0.5, # Half a CPU core max
read_only=True, # Read-only root filesystem
user="nobody", # Non-root user inside container
security_opt=["no-new-privileges"], # Prevent privilege escalation
cap_drop=["ALL"], # Drop all Linux capabilities
auto_remove=True,
# Mount workspace as read-write, everything else read-only
mounts={
"/tmp/workspace": {"bind": "/workspace", "mode": "rw"}
}
)2. Enforce Timeouts Religiously
Never execute code without a timeout. LLMs can generate infinite loops, deadlocks, or computationally expensive operations that would otherwise hang your agent system indefinitely.
# Always set a reasonable timeout
executor = LocalCommandLineCodeExecutor(
timeout=30, # Kill after 30 seconds
work_dir="/safe/workspace"
)
# For longer-running tasks, implement heartbeat checks
def execute_with_heartbeat(executor, code, max_wait=300):
"""Execute code with progressive timeout logic."""
import signal, time
start = time.time()
result = executor.execute_code_blocks([("code", code)])
elapsed = time.time() - start
if elapsed > max_wait:
print(f"Warning: Execution took {elapsed:.1f}s, exceeded soft limit")
return result3. Restrict Python Capabilities Inside the Sandbox
Even within a container, restrict what Python can do. Use a combination of import whitelisting and runtime checks.
# Prepend a security preamble to every executed code block
SECURITY_PREAMBLE = """
# Auto-generated security wrapper - DO NOT REMOVE
import builtins, sys, os
# Override dangerous builtins
_original_import = builtins.__import__
def _safe_import(name, *args, **kwargs):
allowed = {'pandas', 'numpy', 'matplotlib', 'json', 'csv',
'datetime', 'math', 'statistics', 'pathlib',
'collections', 'typing', 'copy', 'functools'}
if name in allowed or name.startswith(('matplotlib.', 'numpy.', 'pandas.')):
return _original_import(name, *args, **kwargs)
raise ImportError(f"Module '{name}' is not allowed in sandbox")
builtins.__import__ = _safe_import
# Restrict file operations to workspace
_original_open = builtins.open
def _safe_open(file, mode='r', *args, **kwargs):
allowed_prefixes = ['/workspace/', './', '../workspace/']
path = os.path.abspath(file) if not file.startswith('/') else file
if any(path.startswith(p) for p in allowed_prefixes) or mode == 'r':
return _original_open(file, mode, *args, **kwargs)
raise PermissionError(f"Write access denied for path: {file}")
builtins.open = _safe_open
print("[SANDBOX] Security wrappers active")
"""
# Apply preamble to all code executions
def wrap_code_with_security(code: str) -> str:
return SECURITY_PREAMBLE + "\n" + code
safe_code = wrap_code_with_security(user_code)
result = executor.execute_code_blocks([("code", safe_code)])4. Implement Comprehensive Logging and Auditing
Every code execution should be logged with full context: what was executed, when, by which agent, and what the result was. This creates an audit trail essential for debugging and security forensics.
import logging
import json
import hashlib
from datetime import datetime
class AuditedCodeExecutor:
"""Wraps an executor with comprehensive audit logging."""
def __init__(self, base_executor, log_path="execution_audit.log"):
self.executor = base_executor
self.logger = logging.getLogger("code_executor_audit")
self.logger.setLevel(logging.INFO)
handler = logging.FileHandler(log_path)
handler.setFormatter(logging.Formatter(
'%(asctime)s | %(levelname)s | %(message)s'
))
self.logger.addHandler(handler)
def execute_code_blocks(self, code_blocks, context=None):
for i, (lang, code) in enumerate(code_blocks):
code_hash = hashlib.sha256(code.encode()).hexdigest()
audit_record = {
"timestamp": datetime.utcnow().isoformat(),
"code_hash": code_hash,
"language": lang,
"code_length": len(code),
"first_100_chars": code[:100],
"context": context or {}
}
self.logger.info(f"EXECUTING: {json.dumps(audit_record)}")
try:
result = self.executor.execute_code_blocks(code_blocks)
self.logger.info(f"RESULT: success, output_length={len(str(result))}")
return result
except Exception as e:
self.logger.error(f"RESULT: failed, error={str(e)}")
raise
# Usage
base = LocalCommandLineCodeExecutor(timeout=60, work_dir="/workspace")
audited_executor = AuditedCodeExecutor(base, log_path="/var/log/agent_executions.log")5. Use Layered Defense: Combine Multiple Safety Measures
No single security measure is foolproof. Combine container isolation, code filtering, Python-level restrictions, timeouts, and auditing for defense in depth. Here's a complete production executor that layers all these protections.
from autogen.coding import DockerCommandLineCodeExecutor
import hashlib, json, logging, os
class ProductionSandboxExecutor:
"""
Layered-defense code executor combining:
- Docker container isolation
- Code pattern filtering
- Python capability restrictions
- Strict resource limits
- Comprehensive audit logging
"""
FORBIDDEN = [
"os.system", "subprocess.", "eval(", "exec(",
"rm -rf", "sudo ", "chmod", "chown",
"/etc/passwd", "/etc/shadow", "~/.ssh",
"socket.", "http.client", "urllib.request"
]
def __init__(self, workspace="/sandbox"):
self.workspace = workspace
os.makedirs(workspace, exist_ok=True)
self.docker_executor = DockerCommandLineCodeExecutor(
image="python:3.11-slim",
timeout=120,
work_dir="/workspace",
bind_dir=workspace,
network="none",
mem_limit="512m",
cpu_limit=1.0,
read_only=True,
user="nobody",
security_opt=["no-new-privileges"],
cap_drop=["ALL"],
auto_remove=True
)
self.logger = self._setup_logging()
def _setup_logging(self):
logger = logging.getLogger("sandbox_executor")
handler = logging.FileHandler("sandbox_audit.log")
handler.setFormatter(logging.Formatter(
'%(asctime)s | %(levelname)s | %(message)s'
))
logger.addHandler(handler)
return logger
def _filter_code(self, code: str) -> tuple[bool, str]:
"""Returns (is_safe, reason)."""
code_lower = code.lower()
for pattern in self.FORBIDDEN:
if pattern.lower() in code_lower:
return False, f"Forbidden pattern: {pattern}"
return True, "OK"
def execute(self, code: str, context: dict = None) -> dict:
"""Execute code with full layered security."""
# Layer 1: Pattern-based filtering
is_safe, reason = self._filter_code(code)
if not is_safe:
self.logger.warning(f"BLOCKED: {reason}")
return {"success": False, "error": reason, "output": ""}
# Layer 2: Prepend security wrappers
wrapped_code = self._security_wrapper() + "\n" + code
# Layer 3: Audit log before execution
code_hash = hashlib.sha256(code.encode()).hexdigest()
self.logger.info(json.dumps({
"event": "EXECUTION_START",
"hash": code_hash,
"length": len(code),
"context": context or {}
}))
# Layer 4: Docker container execution
try:
result = self.docker_executor.execute_code_blocks(
[("code", wrapped_code)]
)
self.logger.info(json.dumps({
"event": "EXECUTION_SUCCESS",
"hash": code_hash,
"output_length": len(str(result))
}))
return {
"success": True,
"output": str(result),
"files": result.code_file or []
}
except Exception as e:
self.logger.error(json.dumps({
"event": "EXECUTION_FAILURE",
"hash": code_hash,
"error": str(e)
}))
return {"success": False, "error": str(e), "output": ""}
def _security_wrapper(self) -> str:
return '''
# === SANDBOX SECURITY WRAPPER ===
import builtins, os
_original_import = builtins.__import__
def _restricted_import(name, *a, **kw):
allowed = {"pandas","numpy","matplotlib","json","csv",
"datetime","math","statistics","pathlib","collections"}
if name in allowed or any(name.startswith(p) for p in ["matplotlib.","numpy.","pandas."]):
return _original_import(name, *a, **kw)
raise ImportError(f"'{name}' not permitted")
builtins.__import__ = _restricted_import
_original_open = builtins.open
def _restricted_open(file, mode="r", *a, **kw):
if "w" in mode or "a" in mode:
allowed = ["/workspace/", "./"]
if not any(os.path.abspath(file).startswith(p) for p in allowed):
raise PermissionError(f"Write access denied: {file}")
return _original_open(file, mode, *a, **kw)
builtins.open = _restricted_open
print("[SANDBOX ACTIVE]", flush=True)
'''
# Instantiate the production executor
executor = ProductionSandboxExecutor(workspace="/tmp/safe_sandbox")
# Test with safe code
result = executor.execute(
'import pandas as pd; df = pd.DataFrame({"a":[1,2]}); print(df)',
context={"agent": "assistant", "task": "test"}
)
print("Safe execution result:", result["output"][:200])
# Test with blocked code
result_blocked = executor.execute(
'import os; os.system("rm -rf /")',
context={"agent": "test"}
)
print("Blocked execution:", result_blocked["error"])6. Handle Execution Failures Gracefully
Code execution can fail for many reasons: syntax errors, runtime exceptions, resource exhaustion, or timeout. Your agent workflow must handle these gracefully and provide useful feedback for the assistant to iterate.
class GracefulExecutionHandler:
"""Wraps executor with intelligent error handling."""
def __init__(self, executor):
self.executor = executor
self.error_count = 0
self.max_retries = 3
def execute_with_feedback(self, code: str) -> dict:
"""Execute code and return structured feedback."""
try:
result = self.executor.execute_code_blocks([("code", code)])
return {
"success": True,
"output": str(result),
"error_type": None,
"suggestion": None
}
except TimeoutError:
self.error_count += 1
return {
"success": False,
"output": "",
"error_type": "TIMEOUT",
"suggestion": "Code exceeded time limit. Optimize loops, "
"reduce data size, or split into smaller chunks."
}
except Exception as e:
self.error_count += 1
error_msg = str(e)
# Classify error for better feedback
if "SyntaxError" in error_msg:
suggestion = "Fix Python syntax. Check indentation and brackets."
elif "ImportError" in error_msg or "ModuleNotFoundError" in error_msg:
suggestion = "Use only allowed modules. Check import whitelist."
elif "MemoryError" in error_msg:
suggestion = "Reduce memory usage. Use generators, smaller batches."
elif "PermissionError" in error_msg:
suggestion = "Code attempted restricted file access. Stay in workspace."
else:
suggestion = f"Error: {error_msg[:200]}. Revise code accordingly."
return {
"success": False,
"output": "",
"error_type": type(e).__name__,
"suggestion": suggestion
}
def should_retry(self) -> bool:
return self.error_count < self.max_retriesConclusion
AutoGen Code Executors transform AI agents from text generators into systems that can safely interact with the computational world. By decoupling code generation from execution and wrapping execution in layered sandboxing — Docker containers, network isolation, resource limits, code filtering, and audit logging — you create an environment where agents can be productive without being dangerous. The key takeaway is that sandboxing is not a single checkbox; it's a stack of complementary defenses. Start with local executors for development, move to Docker-based executors for production, and always combine them with timeouts, import restrictions, and comprehensive logging. With these patterns in place, your AutoGen agents can write and execute code reliably while keeping your infrastructure secure.