Skip to main content

​
What You’ll Learn

  • Implementing the CodeExecutor ABC from autogen_core.code_executor backed by Declaw sandboxes
  • Supporting both Python and shell code blocks
  • Integrating with CodeExecutorAgent and AssistantAgent
  • Running a RoundRobinGroupChat team with termination conditions
  • Demo mode that exercises the executor without needing an OpenAI key

​
Prerequisites

  • Declaw instance running and DECLAW_API_KEY / DECLAW_DOMAIN set
  • OPENAI_API_KEY (optional — the example runs in demo mode without it)
pip install declaw python-dotenv autogen-agentchat "autogen-ext[openai]"
This example is available in Python. TypeScript support coming soon.

​
Code Walkthrough

​
1. Implement DecawCodeExecutor

Subclass CodeExecutor and implement execute_code_blocks. The method receives a list of CodeBlock objects — each has a language and code field: 
from autogen_core.code_executor import CodeBlock, CodeExecutor, CodeResult
from declaw import Sandbox

class DecawCodeExecutor(CodeExecutor):
    """AutoGen CodeExecutor that runs code in Declaw sandboxes."""

    async def execute_code_blocks(
        self,
        code_blocks: list[CodeBlock],
        cancellation_token=None,
    ) -> CodeResult:
        sbx = Sandbox.create(template="python", timeout=300)
        try:
            outputs = []
            last_exit_code = 0
            for block in code_blocks:
                if block.language in ("python", "py", "python3"):
                    sbx.files.write("/tmp/code.py", block.code)
                    result = sbx.commands.run("python3 /tmp/code.py", timeout=30)
                elif block.language in ("bash", "sh", "shell"):
                    sbx.files.write("/tmp/script.sh", block.code)
                    result = sbx.commands.run("sh /tmp/script.sh", timeout=30)
                else:
                    outputs.append(f"Unsupported language: {block.language}")
                    continue
                outputs.append(result.stdout)
                if result.stderr:
                    outputs.append(result.stderr)
                last_exit_code = result.exit_code
            return CodeResult(
                exit_code=last_exit_code,
                output="\n".join(outputs),
            )
        finally:
            sbx.kill()

    async def restart(self) -> None:
        pass  # Stateless — each call gets a fresh sandbox

    async def stop(self) -> None:
        pass
A fresh sandbox is created per execute_code_blocks call, ensuring full isolation between turns.

​
2. Wire into AutoGen agents

from autogen_agentchat.agents import CodeExecutorAgent, AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import TextMentionTermination
from autogen_ext.models.openai import OpenAIChatCompletionClient

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

coder = AssistantAgent(
    name="coder",
    model_client=model_client,
    system_message=(
        "You write Python code to solve tasks. "
        "Put code in ```python code blocks. "
        "Say TERMINATE when done."
    ),
)

executor = CodeExecutorAgent(
    name="executor",
    code_executor=DecawCodeExecutor(),
)

​
3. Run the team

import asyncio

termination = TextMentionTermination("TERMINATE")
team = RoundRobinGroupChat(
    [coder, executor],
    termination_condition=termination,
    max_turns=6,
)

result = await team.run(
    task="Write Python code to find the 20th triangular number and print it."
)

for msg in result.messages:
    print(f"\n[{msg.source}] {msg.content}")

​
4. Demo mode (no API key needed)

Run the executor directly without an AutoGen agent: 
import asyncio
from declaw import Sandbox

code = """\
n = 20
triangular = n * (n + 1) // 2
print(f"The {n}th triangular number is {triangular}")
"""

sbx = Sandbox.create(template="python", timeout=300)
try:
    sbx.files.write("/tmp/code.py", code)
    result = sbx.commands.run("python3 /tmp/code.py", timeout=30)
    print(result.stdout)
finally:
    sbx.kill()

​
Expected Output

In demo mode: 
=======================================================
AutoGen + Declaw Sandbox Example
=======================================================
No OPENAI_API_KEY found -- running demo mode.

--- CodeExecutor Definition ---
class DecawCodeExecutor(CodeExecutor):
    async def execute_code_blocks(self, code_blocks, cancellation_token=None):
        sbx = Sandbox.create(template="python", timeout=300)
        ...

--- Running Code Directly in Declaw Sandbox ---
Code:
n = 20
triangular = n * (n + 1) // 2
print(f"The {n}th triangular number is {triangular}")

stdout: The 20th triangular number is 210
stderr:
exit_code: 0

Sandbox cleaned up.

=======================================================
Done!
=======================================================