Implementation:Microsoft Autogen CodeExecutorAgent

Knowledge Sources	Microsoft_Autogen
Domains	Agent, Code_Execution, LLM, Multi_Agent
Last Updated	2026-02-11 17:00 GMT

Overview

An experimental agent that generates and executes code snippets based on user instructions, with optional LLM-based code generation, approval mechanisms, and automatic error retry capabilities.

Description

The `CodeExecutorAgent` is a specialized agent in the AutoGen framework that handles code execution workflows. It operates in two modes:

Without model_client (Executor-only mode):

• Extracts code blocks from incoming TextMessage messages
• Executes code using a provided CodeExecutor (typically Docker-based)
• Returns execution results as TextMessage responses

With model_client (Generator-Executor mode):

• Generates code based on user queries using an LLM
• Executes the generated code
• Reflects on execution results and iterates if errors occur
• Supports automatic retry with error analysis up to max_retries_on_error attempts
• Requires models with structured output support for retry mechanism

Key features include:

• Safety: Optional approval functions (sync/async) for code review before execution
• Streaming: Support for streaming LLM responses via model_client_stream
• Language Support: Configurable language filtering (default: python, sh)
• Context Management: Maintains conversation history for multi-turn interactions
• Thought Events: Yields hidden reasoning from models that support thought generation

Usage

Use CodeExecutorAgent when you need to execute code in a controlled environment, either from other agents in a group chat or by generating code autonomously based on user instructions.

Code Reference

Source Location

• Repository: Microsoft_Autogen
• File: python/packages/autogen-agentchat/src/autogen_agentchat/agents/_code_executor_agent.py
• Lines: 1-893

Signature

class CodeExecutorAgent(BaseChatAgent, Component[CodeExecutorAgentConfig]):
    def __init__(
        self,
        name: str,
        code_executor: CodeExecutor,
        *,
        model_client: ChatCompletionClient | None = None,
        model_context: ChatCompletionContext | None = None,
        model_client_stream: bool = False,
        max_retries_on_error: int = 0,
        description: str | None = None,
        system_message: str | None = DEFAULT_SYSTEM_MESSAGE,
        sources: Sequence[str] | None = None,
        supported_languages: List[str] | None = None,
        approval_func: Optional[ApprovalFuncType] = None,
    ) -> None

    async def on_messages(
        self,
        messages: Sequence[BaseChatMessage],
        cancellation_token: CancellationToken
    ) -> Response

    async def on_messages_stream(
        self,
        messages: Sequence[BaseChatMessage],
        cancellation_token: CancellationToken
    ) -> AsyncGenerator[BaseAgentEvent | BaseChatMessage | Response, None]

    async def execute_code_block(
        self,
        code_blocks: List[CodeBlock],
        cancellation_token: CancellationToken
    ) -> CodeResult

Import

from autogen_agentchat.agents import CodeExecutorAgent
from autogen_agentchat.agents import ApprovalRequest, ApprovalResponse

I/O Contract

Inputs

Name	Type	Required	Description
name	str	Yes	Unique identifier for the agent
code_executor	CodeExecutor	Yes	Code execution backend (e.g., DockerCommandLineCodeExecutor)
model_client	ChatCompletionClient	No	LLM client for code generation (if None, only executes code from messages)
model_context	ChatCompletionContext	No	Context manager for conversation history (default: UnboundedChatCompletionContext)
model_client_stream	bool	No	Enable streaming responses (default: False)
max_retries_on_error	int	No	Number of retry attempts on code execution errors (default: 0)
description	str	No	Agent description for team coordination
system_message	str	No	System prompt for the model (default: DEFAULT_SYSTEM_MESSAGE)
sources	Sequence[str]	No	Filter messages by source agent names (executor-only mode)
supported_languages	List[str]	No	Languages to parse and execute (default: ["python", "sh"])
approval_func	ApprovalFuncType	No	Sync/async function for code execution approval

Outputs

Name	Type	Description
Response	Response	Final response containing chat_message (TextMessage) and optional inner_messages
CodeGenerationEvent	BaseAgentEvent	Streamed event when code is generated (includes code_blocks and retry_attempt)
CodeExecutionEvent	BaseAgentEvent	Streamed event after code execution (includes result and retry_attempt)
ThoughtEvent	BaseAgentEvent	Streamed event for model's hidden reasoning (if supported by model)
ModelClientStreamingChunkEvent	BaseAgentEvent	Streamed text chunks when model_client_stream=True

Core Components

Approval Mechanism

The approval system allows human-in-the-loop or model-based code review:

class ApprovalRequest(BaseModel):
    code: str  # Combined code blocks in markdown format
    context: List[LLMMessage]  # Full conversation history

class ApprovalResponse(BaseModel):
    approved: bool
    reason: str

Approval functions can be:

• Synchronous: `def approval_func(request: ApprovalRequest) -> ApprovalResponse`
• Asynchronous: `async def approval_func(request: ApprovalRequest) -> ApprovalResponse`

Code Extraction

The agent extracts code blocks using regex pattern matching for markdown code blocks:

```python
print("Hello World")
```

Only languages in `supported_languages` are parsed and executed.

Retry Logic

When `max_retries_on_error > 0` and code execution fails:

1. Agent asks the model: "Should we retry?" using structured output (RetryDecision)
2. If model says yes, it regenerates code with error context
3. Process repeats until success or max retries reached
4. Final reflection is always performed regardless of success

Message Flow (Generator Mode)

1. Add user/handoff messages to model context
2. Call LLM to generate code (optionally streaming)
3. Yield ThoughtEvent if model produces hidden reasoning
4. Extract code blocks from model response
5. Yield CodeGenerationEvent
6. Request approval if approval_func is set
7. Execute code blocks
8. Yield CodeExecutionEvent
9. If error and retries available, ask model for retry decision
10. Always perform final reflection and yield Response

Usage Examples

Basic Executor Mode (No Model)

import asyncio
from autogen_agentchat.agents import CodeExecutorAgent
from autogen_agentchat.messages import TextMessage
from autogen_ext.code_executors.docker import DockerCommandLineCodeExecutor
from autogen_core import CancellationToken

async def main():
    code_executor = DockerCommandLineCodeExecutor(work_dir="coding")
    await code_executor.start()

    agent = CodeExecutorAgent("executor", code_executor=code_executor)

    task = TextMessage(
        content='```python\nprint("Hello World")\n```',
        source="user"
    )

    response = await agent.on_messages([task], CancellationToken())
    print(response.chat_message.content)

    await code_executor.stop()

asyncio.run(main())

Generator Mode with Approval

import asyncio
from autogen_agentchat.agents import CodeExecutorAgent, ApprovalRequest, ApprovalResponse
from autogen_ext.code_executors.docker import DockerCommandLineCodeExecutor
from autogen_ext.models.openai import OpenAIChatCompletionClient

def human_approval(request: ApprovalRequest) -> ApprovalResponse:
    print(f"Code to approve:\n{request.code}")
    choice = input("Approve? (y/n): ")
    if choice.lower() == 'y':
        return ApprovalResponse(approved=True, reason="User approved")
    return ApprovalResponse(approved=False, reason="User denied")

async def main():
    code_executor = DockerCommandLineCodeExecutor(work_dir="coding")
    await code_executor.start()

    model_client = OpenAIChatCompletionClient(model="gpt-4o")

    agent = CodeExecutorAgent(
        "code_agent",
        code_executor=code_executor,
        model_client=model_client,
        approval_func=human_approval,
        max_retries_on_error=2
    )

    from autogen_agentchat.ui import Console
    await Console(agent.run_stream(task="Calculate fibonacci(10)"))

    await code_executor.stop()

asyncio.run(main())

Model-Based Approval

async def model_approval(request: ApprovalRequest) -> ApprovalResponse:
    review_client = OpenAIChatCompletionClient(model="gpt-4o")
    instruction = "Review this code for safety. Respond with JSON: {approved: bool, reason: str}"

    response = await review_client.create(
        messages=[
            SystemMessage(content=instruction),
            *request.context,
            UserMessage(content=request.code, source="user")
        ],
        json_output=ApprovalResponse
    )

    return ApprovalResponse.model_validate_json(response.content)

agent = CodeExecutorAgent(
    "secure_agent",
    code_executor=executor,
    model_client=model,
    approval_func=model_approval
)

Group Chat Integration

from autogen_agentchat.agents import AssistantAgent, CodeExecutorAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import MaxMessageTermination

async def main():
    model_client = OpenAIChatCompletionClient(model="gpt-4o")
    code_executor = DockerCommandLineCodeExecutor(work_dir="coding")
    await code_executor.start()

    coder = AssistantAgent("coder", model_client=model_client)
    executor = CodeExecutorAgent("executor", code_executor=code_executor)

    team = RoundRobinGroupChat(
        participants=[coder, executor],
        termination_condition=MaxMessageTermination(10)
    )

    await Console(team.run_stream(task="Write and run a prime number checker"))

    await code_executor.stop()

Configuration

Default Constants

• DEFAULT_TERMINAL_DESCRIPTION: "A computer terminal that performs no other action than running Python scripts..."
• DEFAULT_AGENT_DESCRIPTION: "A Code Execution Agent that generates and executes Python and shell scripts..."
• DEFAULT_SYSTEM_MESSAGE: "You are a Code Execution Agent. Your role is to generate and execute Python code..."
• NO_CODE_BLOCKS_FOUND_MESSAGE: "No code blocks found in the thread. Please provide at least one markdown-encoded code block..."
• DEFAULT_SUPPORTED_LANGUAGES: ["python", "sh"]

Serialization

The agent supports serialization via `dump_component()` and `load_component()`, but approval_func cannot be serialized and will be set to None when loading from config.

Related Pages

• BaseChatAgent
• CodeExecutor
• ChatCompletionClient
• DockerCommandLineCodeExecutor
• AssistantAgent
• RoundRobinGroupChat