AutoGen Agent Guardrails with Veto

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_ext.models.openai import OpenAIChatCompletionClient
from veto import Veto

veto = await Veto.init()

# Define tools as normal functions
async def send_email(to: str, subject: str, body: str) -> str:
    """Send an email to a recipient."""
    return f"Email sent to {to}"

async def query_database(sql: str) -> str:
    """Run a read-only SQL query."""
    return f"Query returned 42 rows"

async def delete_records(table: str, where: str) -> str:
    """Delete records from a database table."""
    return f"Deleted matching records from {table}"

# Wrap tools with Veto before passing to agents
safe_tools = veto.wrap([send_email, query_database, delete_records])

model = OpenAIChatCompletionClient(model="gpt-5.4")

agent = AssistantAgent(
    name="data_assistant",
    model_client=model,
    tools=safe_tools,  # Guarded tools
    system_message="You help users query data and send reports.",
)

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import TextMentionTermination
from autogen_ext.models.openai import OpenAIChatCompletionClient
from veto import Veto

veto = await Veto.init()

model = OpenAIChatCompletionClient(model="gpt-5.4")

# Each agent gets its own scoped tool set
researcher_tools = veto.wrap(
    [search_web, read_document],
    scope="researcher"  # Policy scope limits what this agent can access
)

writer_tools = veto.wrap(
    [write_file, send_email],
    scope="writer"
)

reviewer_tools = veto.wrap(
    [approve_document, reject_document],
    scope="reviewer"
)

researcher = AssistantAgent(
    name="researcher",
    model_client=model,
    tools=researcher_tools,
    system_message="Research topics and gather information.",
)

writer = AssistantAgent(
    name="writer",
    model_client=model,
    tools=writer_tools,
    system_message="Write reports based on research findings.",
)

reviewer = AssistantAgent(
    name="reviewer",
    model_client=model,
    tools=reviewer_tools,
    system_message="Review and approve or reject documents.",
)

termination = TextMentionTermination("APPROVED")
team = RoundRobinGroupChat(
    [researcher, writer, reviewer],
    termination_condition=termination,
)

result = await team.run(task="Research Q3 sales and write a report")

from autogen_agentchat.agents import AssistantAgent
from autogen_ext.models.openai import OpenAIChatCompletionClient
from veto import Veto

veto = await Veto.init()

# The delegation problem: Agent A asks Agent B to call a tool
# that Agent A itself is not authorized to use.
# Veto enforces policies at the tool boundary, not the agent boundary.
# Even if Agent A convinces Agent B to call delete_production_db,
# the tool itself checks authorization.

async def delete_production_db(table: str) -> str:
    """Delete a table from the production database."""
    return f"Deleted {table}"

# Wrap with strict policy: only "admin" scope can use this tool
admin_tools = veto.wrap(
    [delete_production_db],
    scope="admin"
)

# Even if a non-admin agent delegates to an admin agent,
# the tool call is validated against the calling context
worker = AssistantAgent(
    name="worker",
    model_client=OpenAIChatCompletionClient(model="gpt-5.4"),
    tools=veto.wrap([search_web, read_file], scope="worker"),
    system_message="You handle research tasks. Never modify production data.",
)

admin = AssistantAgent(
    name="admin",
    model_client=OpenAIChatCompletionClient(model="gpt-5.4"),
    tools=admin_tools,
    system_message="You handle admin tasks when explicitly approved.",
)

from autogen_core.tools import FunctionTool
from veto import Veto

veto = await Veto.init()

# AutoGen 0.4+ FunctionTool pattern
async def calculate_risk(portfolio_id: str, scenario: str) -> str:
    """Calculate risk metrics for a portfolio under a scenario."""
    return f"VaR: $1.2M, CVaR: $1.8M for {portfolio_id}"

# Wrap the raw function, then create FunctionTool
safe_calculate = veto.wrap([calculate_risk])[0]

risk_tool = FunctionTool(
    safe_calculate.handler,
    description="Calculate portfolio risk metrics",
    name="calculate_risk",
)

from veto import Veto

veto = await Veto.init()

# Pre-flight check without wrapping
async def custom_tool_handler(agent_name: str, tool_name: str, args: dict):
    decision = await veto.guard(
        tool_name,
        args,
        context={
            "agent": agent_name,
            "session_id": current_session,
        }
    )

    if decision.decision == "deny":
        return f"Action blocked: {decision.reason}"

    if decision.decision == "require_approval":
        return f"Pending human approval (id: {decision.approval_id})"

    return await execute_tool(tool_name, args)

version: "1.0"
name: AutoGen multi-agent policies

rules:
  - id: researcher-read-only
    action: block
    scope: researcher
    tools: [write_file, send_email, delete_records]
    message: "Researcher agent is read-only"

  - id: writer-no-delete
    action: block
    scope: writer
    tools: [delete_records, delete_production_db]
    message: "Writer cannot delete data"

  - id: admin-approval-required
    action: require_approval
    scope: admin
    tools: [delete_production_db]
    message: "Production deletions require human approval"

  - id: block-external-emails
    action: block
    tools: [send_email]
    conditions:
      - field: arguments.to
        operator: not_matches
        value: "@company\.com$"
    message: "External emails blocked in multi-agent workflows"

  - id: rate-limit-all-agents
    action: block
    tools: ["*"]
    conditions:
      - field: context.session_tool_count
        operator: greater_than
        value: 50
    message: "Session tool call limit exceeded"