Implementing Approval Workflows with CodeExecutorAgent Approval Callbacks in AutoGen
The CodeExecutorAgent in Microsoft AutoGen supports human-in-the-loop safety gates through an optional approval_func callback that receives an ApprovalRequest and must return an ApprovalResponse to either permit or block code execution.
The Microsoft AutoGen framework enables autonomous AI agents to execute code, but production systems require safeguards before running arbitrary commands. By implementing approval workflows with CodeExecutorAgent approval callbacks, you can enforce security policies, request explicit human confirmation, or integrate external governance systems without modifying the underlying executor logic. This mechanism supports both synchronous console prompts and asynchronous remote approval services.
Core Architecture of the Approval System
The approval workflow relies on three tightly integrated components defined in python/packages/autogen-agentchat/src/autogen_agentchat/agents/_code_executor_agent.py (lines 70-84).
ApprovalRequest and ApprovalResponse Data Classes
The ApprovalRequest dataclass (lines 70-77) encapsulates the execution context with three fields: code (the string to execute), language (the programming language), and an optional context dictionary for additional metadata. Your callback must return an ApprovalResponse containing an approved boolean and a reason string explaining the decision.
ApprovalFuncType Signature
The ApprovalFuncType type alias (lines 83-84) defines the callable signature as either:
- Synchronous:
Callable[[ApprovalRequest], ApprovalResponse] - Asynchronous:
Callable[[ApprovalRequest], Awaitable[ApprovalResponse]]
When the agent's _run_code method processes an execution request (around lines 182-184), it checks for the presence of self._approval_func. If provided, the agent invokes the callback and awaits the response if necessary; if approved is False, the agent aborts and returns the denial reason instead of executing the code.
Implementing a Synchronous Approval Callback
For command-line interfaces or deterministic policy checks, implement a synchronous function that inspects the ApprovalRequest and returns immediately.
from autogen_agentchat.agents import CodeExecutorAgent
from autogen_agentchat.agents._code_executor_agent import ApprovalRequest, ApprovalResponse
from autogen_agentchat.code_executors import LocalCommandLineCodeExecutor
def console_approval(request: ApprovalRequest) -> ApprovalResponse:
"""Prompt the user via stdin to approve or deny code execution."""
print(f"\n🔒 APPROVAL REQUIRED ({request.language})")
print(request.code)
decision = input("Execute? (y/n): ").strip().lower()
if decision == "y":
return ApprovalResponse(approved=True, reason="User confirmed via console")
return ApprovalResponse(approved=False, reason="User denied execution")
agent = CodeExecutorAgent(
name="secure_cli_agent",
code_executor=LocalCommandLineCodeExecutor(),
approval_func=console_approval,
)
This pattern is validated in python/packages/autogen-agentchat/tests/test_code_executor_agent.py (lines 506-509), which tests scenarios ranging from "allow-all" to "deny-dangerous" policies that block specific keywords like rm.
Implementing Asynchronous Approval Workflows
For web-based dashboards, Slack notifications, or microservice-based governance, use an async callback to avoid blocking the event loop.
import asyncio
from autogen_agentchat.agents import CodeExecutorAgent
from autogen_agentchat.agents._code_executor_agent import ApprovalRequest, ApprovalResponse
from autogen_agentchat.code_executors import LocalCommandLineCodeExecutor
async def remote_policy_check(request: ApprovalRequest) -> ApprovalResponse:
"""Simulate calling an external approval API."""
await asyncio.sleep(0.5) # Simulate network latency
# Example policy: Auto-approve Python, deny shell scripts
if request.language == "python":
return ApprovalResponse(approved=True, reason="Python auto-approved by policy")
return ApprovalResponse(approved=False, reason="Only Python execution permitted")
agent = CodeExecutorAgent(
name="async_policy_agent",
code_executor=LocalCommandLineCodeExecutor(),
approval_func=remote_policy_check,
)
# The agent automatically awaits async approval functions internally
result = await agent.run("print('Hello')", language="python")
The test suite verifies async handling in test_code_executor_agent.py (lines 462-470 and 483-490), ensuring that the agent correctly awaits coroutine-based callbacks.
Integrating with MagenticOne Teams
Higher-level orchestrations like MagenticOne propagate approval callbacks to their internal CodeExecutorAgent instances. The constructor signature in python/packages/autogen-ext/src/autogen_ext/teams/magentic_one.py (lines 40-41, 95-110, and 128-150) accepts an approval_func parameter that gets forwarded to the underlying agent.
from autogen_ext.teams import MagenticOne
from autogen_agentchat.agents._code_executor_agent import ApprovalRequest, ApprovalResponse
def strict_security_policy(request: ApprovalRequest) -> ApprovalResponse:
"""Block any code containing dangerous shell commands."""
dangerous_keywords = ["rm -rf", "del /f", "format"]
if any(keyword in request.code for keyword in dangerous_keywords):
return ApprovalResponse(approved=False, reason="Dangerous delete command detected")
return ApprovalResponse(approved=True, reason="Passed security scan")
team = MagenticOne(
client=my_chat_client,
approval_func=strict_security_policy,
)
All code execution performed by this team instance now routes through your custom policy function before reaching the executor.
Serialization Constraints and Limitations
Agents configured with approval callbacks cannot be serialized to JSON. The approval_func is a Python callable that is not JSON-serializable, and attempting to serialize the agent raises a ValueError. This behavior is explicitly tested in python/packages/autogen-agentchat/tests/test_code_executor_agent.py (lines 564-572). If you need to persist agent state, you must remove or reset the approval_func before serialization.
Summary
- Approval callbacks in
CodeExecutorAgentprovide a human-in-the-loop gate that intercepts code execution requests before they reach the underlying executor. - The callback receives an
ApprovalRequestcontaining the code, language, and context, and must return anApprovalResponsewith anapprovedboolean andreasonstring. - Both synchronous and asynchronous callbacks are supported via the
ApprovalFuncTypetype alias (lines 83-84). - Higher-level teams like
MagenticOneacceptapproval_funcparameters and forward them to internal agents (lines 40-41, 95-110 inmagentic_one.py). - Agents with active approval callbacks are not serializable and will raise
ValueErrorduring JSON serialization attempts (lines 564-572 in test file).
Frequently Asked Questions
What happens if the approval callback rejects the code?
If the callback returns ApprovalResponse(approved=False, ...), the CodeExecutorAgent aborts execution and returns a message containing the denial reason instead of the actual execution output. The underlying executor never receives the code, ensuring dangerous commands are blocked at the agent layer.
Can I use both synchronous and asynchronous approval functions?
Yes. The ApprovalFuncType type alias (lines 83-84 of _code_executor_agent.py) accepts both sync callables and async coroutines. The agent automatically detects whether the callback is a coroutine and awaits it if necessary before proceeding with execution.
How do I implement approval workflows with MagenticOne?
Pass your approval function directly to the MagenticOne constructor via the approval_func parameter. According to the source in magentic_one.py (lines 128-150), this function is forwarded to the internal CodeExecutorAgent, ensuring all team-orchestrated code execution respects your approval policy.
Why can't I serialize an agent that has an approval callback?
Python functions and lambdas are not JSON-serializable. The CodeExecutorAgent validates this constraint during serialization attempts and raises a ValueError if approval_func is set (see test validation in test_code_executor_agent.py, lines 564-572). To serialize the agent state, you must set approval_func to None before calling json.dumps() or similar methods.
Have a question about this repo?
These articles cover the highlights, but your codebase questions are specific. Give your agent direct access to the source. Share this with your agent to get started:
curl -s "https://instagit.com/install.md" Maintain an open-source project? Get it listed too →