Step 1: Install dependencies

pip install pyautogen requests docker

Set your keys:

export A2A_API_KEY="a2a_your_key_here"
export OPENAI_API_KEY="sk-..."

Step 2: Add the safety layer

Wrap DockerCommandLineCodeExecutor with a safety check that screens every code block before execution:

import os, requests
from autogen import AssistantAgent, UserProxyAgent
from autogen.coding import DockerCommandLineCodeExecutor

A2A_URL = "https://a2ainfrastructure.com/v1/evaluate"

def safety_screen(code: str) -> bool:
    resp = requests.post(A2A_URL, json={"command": code}, headers={
        "Authorization": f"Bearer {os.getenv('A2A_API_KEY')}"
    })
    return resp.json()["allowed"]

class SafeDockerExecutor(DockerCommandLineCodeExecutor):
    def execute_code_blocks(self, code_blocks):
        for block in code_blocks:
            if not safety_screen(block.code):
                raise RuntimeError(
                    f"Blocked by safety shield: {block.code[:80]}..."
                )
        return super().execute_code_blocks(code_blocks)

# Use the safe executor in your agent setup
executor = SafeDockerExecutor(image="python:3.11-slim")

user_proxy = UserProxyAgent(
    name="user_proxy",
    code_execution_config={"executor": executor}
)

assistant = AssistantAgent(
    name="assistant",
    llm_config={"model": "gpt-4o"}
)

user_proxy.initiate_chat(assistant, message="Analyze server disk usage")

Step 3: Verify

# Safe code blocks execute normally inside Docker
# assistant generates: import os; print(os.statvfs('/').f_bavail)
# -> ALLOWED, runs in container

# Dangerous code is blocked before reaching the container
# assistant generates: import shutil; shutil.rmtree('/')
# -> RuntimeError: Blocked by safety shield