AutoGen Integration

Enforce Control Zero policies on Microsoft AutoGen agents, tool calls, and multi-agent teams.

No first-party SDK integration

Control Zero does not have a dedicated AutoGen integration module. The examples on this page show how to add cz.guard() calls manually to your AutoGen tools and agents.

Overview

AutoGen v0.4+ is a complete rewrite of the framework with a new agent architecture built around the on_messages method. Agents are composed into teams (e.g. RoundRobinGroupChat) and tools are plain async functions registered directly on agents. Control Zero integrates by wrapping tool functions with cz.guard() calls before the tool logic executes.

Installation

pip install autogen-agentchat autogen-ext controlzero

Setup

Governing Tool Functions

Wrap tool functions with cz.guard() before registering them on an agent:

from autogen_agentchat.agents import AssistantAgent
from autogen_ext.models.openai import OpenAIChatCompletionClient
from controlzero import Client, PolicyDeniedError

cz = Client(api_key="cz_live_your_api_key_here")


# Govern a tool function by calling cz.guard() inside it
async def search_web(query: str) -> str:
    cz.guard("search", method="web", args={"query": query}, raise_on_deny=True)
    # ... actual search implementation
    return f"Results for: {query}"


async def calculator(expression: str) -> str:
    cz.guard("calculator", args={"expression": expression}, raise_on_deny=True)
    # use a safe math parser in production
    return str(expression)


# Register governed tools on the agent
agent = AssistantAgent(
    name="research_agent",
    model_client=OpenAIChatCompletionClient(model="gpt-5.4"),
    tools=[search_web, calculator],
)

Using a Guard Decorator

For cleaner code across many tools, use a decorator pattern:

import functools
from controlzero import Client

cz = Client(api_key="cz_live_your_api_key_here")


def guarded(action: str):
    """Decorator that enforces a Control Zero policy before a tool runs."""
    def decorator(func):
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            cz.guard(action, args={"tool": func.__name__}, raise_on_deny=True)
            return await func(*args, **kwargs)
        return wrapper
    return decorator


@guarded("search")
async def search_web(query: str) -> str:
    return f"Results for: {query}"


@guarded("retrieve_docs")
async def retrieve_docs(query: str) -> str:
    return f"Documents for: {query}"

What Gets Enforced

AutoGen Event	Policy Action	Policy Resource
Tool call	tool.call	tool/{function_name}
Code execution	tool.call	tool/execute_code
Python REPL	tool.call	tool/python_repl
Shell execution	tool.call	tool/shell_exec
Inter-agent message	agent.message	agent/{recipient_name}

Full Example: Research Pipeline with Guards

import asyncio
from autogen_agentchat.agents import AssistantAgent, UserProxyAgent
from autogen_agentchat.conditions import TextMentionTermination
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient
from controlzero import Client, PolicyDeniedError

# --- Control Zero setup ---
cz = Client(api_key="cz_live_your_api_key_here")


# --- Define guarded tools ---
async def search(query: str) -> str:
    """Search the web for information."""
    cz.guard("search", args={"agent_id": "research-pipeline"}, raise_on_deny=True)
    return f"Search results for: {query}"


async def retrieve_docs(query: str) -> str:
    """Retrieve documents from the knowledge base."""
    cz.guard("retrieve_docs", args={"agent_id": "research-pipeline"}, raise_on_deny=True)
    return f"Documents for: {query}"


async def safe_calculate(expression: str) -> str:
    """Evaluate a mathematical expression safely."""
    cz.guard("calculator", args={"agent_id": "research-pipeline"}, raise_on_deny=True)
    # use a safe math parser (e.g. simpleeval) in production
    return f"Result: {expression}"


# --- Configure agents ---
model_client = OpenAIChatCompletionClient(model="gpt-5.4")

assistant = AssistantAgent(
    name="research_assistant",
    model_client=model_client,
    tools=[search, retrieve_docs, safe_calculate],
)

user_proxy = UserProxyAgent(name="user_proxy")


# --- Run the conversation ---
async def main():
    termination = TextMentionTermination("TERMINATE")
    team = RoundRobinGroupChat(
        [assistant, user_proxy],
        termination_condition=termination,
    )
    try:
        await Console(
            team.run_stream(
                task="Find the latest quarterly revenue and calculate year-over-year growth."
            )
        )
    except PolicyDeniedError as e:
        print(f"Blocked by policy: {e.decision.reason}")


asyncio.run(main())

Example Policy

Allow research tools but block code execution and outbound actions:

{
  "name": "autogen-research-policy",
  "rules": [
    {
      "effect": "allow",
      "action": "tool:call",
      "resource": "tool/search"
    },
    {
      "effect": "allow",
      "action": "tool:call",
      "resource": "tool/retrieve_docs"
    },
    {
      "effect": "allow",
      "action": "tool:call",
      "resource": "tool/calculator"
    },
    {
      "effect": "deny",
      "action": "tool:call",
      "resource": "tool/execute_code"
    },
    {
      "effect": "deny",
      "action": "tool:call",
      "resource": "tool/python_repl*"
    },
    {
      "effect": "deny",
      "action": "tool:call",
      "resource": "tool/shell_exec*"
    },
    {
      "effect": "deny",
      "action": "tool:call",
      "resource": "tool/send_*"
    },
    {
      "effect": "deny",
      "action": "tool:call",
      "resource": "tool/post_*"
    }
  ]
}

What happens at runtime:

• search and retrieve_docs calls: ALLOWED.
• calculator calls: ALLOWED.
• Any attempt to run execute_code, python_repl, or shell_exec: BLOCKED. The agent receives a PolicyDeniedError.
• Any outbound send_* or post_* actions: BLOCKED.

Pairing with AutoGen's Termination Conditions

AutoGen v0.4+ uses termination conditions (e.g. TextMentionTermination, MaxMessageTermination) to stop team runs. Control Zero complements this:

• Autonomous teams (MaxMessageTermination): Fully autonomous execution with Control Zero enforcing guardrails on every tool call automatically.
• Human-in-the-loop (UserProxyAgent): The user reviews messages, and Control Zero blocks policy violations before the user sees the tool result.
• Token-limited runs (TokenUsageTermination): Useful for cost-capped pipelines; pair with Control Zero to block high-risk tools after budget thresholds.

Local Policy File

For local development, use the controlzero init command to generate an AutoGen-specific policy template:

controlzero init --template autogen

This creates a controlzero.yaml that denies bare code execution and shell access while allowing named research and retrieval tools.

Next Steps

• LangChain Integration: Single-agent LLM governance with callback handlers.
• CrewAI Integration: Multi-agent crew governance.
• Python SDK: Full API reference for enforcement, wrapping, and audit.
• Policies: Learn how to construct dashboard policies.