Skip to main content

OpenAI Agents SDK Integration

Enforce Control Zero policies on OpenAI Agents SDK tool calls, handoffs, and guardrails.

Overview​

The OpenAI Agents SDK (formerly Swarm) provides a lightweight framework for building multi-agent systems with tool calling, handoffs between agents, and built-in guardrails. Control Zero integrates at the tool execution layer: every tool call is checked against your policies before the function runs. Denied calls raise a PolicyDeniedError that the agent framework surfaces as a tool error.

Installation​

pip install controlzero openai-agents

Setup​

Tool Wrapper Approach​

Wrap your agent tools with Control Zero enforcement:

from controlzero import Client
from agents import Agent, Runner, function_tool

cz = Client(api_key="cz_live_your_api_key_here")


@function_tool
def search_knowledge_base(query: str) -> str:
    """Search the internal knowledge base."""
    cz.guard("search_knowledge_base", args={"agent_id": "support-agent"})
    return f"KB results for: {query}"


@function_tool
def send_email(to: str, subject: str, body: str) -> str:
    """Send an email to a customer."""
    cz.guard("send_email", args={"agent_id": "support-agent", "recipient": to})
    return f"Email sent to {to}"


agent = Agent(
    name="Support Agent",
    instructions="You help customers with their questions.",
    tools=[search_knowledge_base, send_email],
)

Guardrail Approach​

Use Control Zero as an input or output guardrail within the OpenAI Agents SDK guardrail system:

from controlzero import Client
from agents import Agent, Runner, InputGuardrail, GuardrailFunctionOutput

cz = Client(api_key="cz_live_your_api_key_here")


async def policy_guardrail(ctx, agent, input_data) -> GuardrailFunctionOutput:
    """Check input against Control Zero DLP policies."""
    try:
        cz.guard(f"agent/{agent.name}", method="input", args={"input_text": str(input_data)})
        return GuardrailFunctionOutput(
            output_info="Policy check passed",
            tripwire_triggered=False,
        )
    except Exception:
        return GuardrailFunctionOutput(
            output_info="Input blocked by Control Zero policy",
            tripwire_triggered=True,
        )


agent = Agent(
    name="Analyst",
    instructions="Analyze data and provide insights.",
    input_guardrails=[
        InputGuardrail(guardrail_function=policy_guardrail),
    ],
)

What Gets Enforced​

OpenAI Agents EventPolicy ActionPolicy Resource
Tool calltool.calltool/{tool_name}
Agent handoffagent.handoffagent/{target_agent_name}
Input guardrailagent.inputagent/{agent_name}
Output guardrailagent.outputagent/{agent_name}

Full Example: Multi-Agent Support System​

from controlzero import Client
import controlzero
from agents import Agent, Runner, function_tool

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


def guarded(tool_name: str, agent_id: str = ""):
    """Decorator for policy-enforced tools."""
    def decorator(func):
        original = func
        def wrapper(*args, **kwargs):
            cz.guard(tool_name, args={"agent_id": agent_id})
            return original(*args, **kwargs)
        wrapper.__name__ = original.__name__
        wrapper.__doc__ = original.__doc__
        return wrapper
    return decorator


# --- Triage agent tools ---
@function_tool
def lookup_customer(customer_id: str) -> str:
    """Look up customer information by ID."""
    cz.guard("lookup_customer", args={"agent_id": "triage-agent"})
    return f"Customer {customer_id}: Premium tier, active since 2024"


@function_tool
def check_order_status(order_id: str) -> str:
    """Check the status of an order."""
    cz.guard("check_order_status", args={"agent_id": "triage-agent"})
    return f"Order {order_id}: Shipped, arriving tomorrow"


# --- Escalation agent tools ---
@function_tool
def issue_refund(order_id: str, amount: float) -> str:
    """Issue a refund for an order."""
    cz.guard("issue_refund", args={"agent_id": "escalation-agent", "amount": str(amount)})
    return f"Refund of ${amount} issued for order {order_id}"


@function_tool
def escalate_to_human(reason: str) -> str:
    """Escalate the case to a human support agent."""
    cz.guard("escalate_to_human", args={"agent_id": "escalation-agent"})
    return f"Escalated to human agent. Reason: {reason}"


# --- Define agents ---
escalation_agent = Agent(
    name="Escalation Agent",
    instructions="You handle complex cases that require refunds or human escalation.",
    tools=[issue_refund, escalate_to_human],
)

triage_agent = Agent(
    name="Triage Agent",
    instructions=(
        "You are the first point of contact. Look up customer info and order status. "
        "Hand off to the Escalation Agent for refunds or complex issues."
    ),
    tools=[lookup_customer, check_order_status],
    handoffs=[escalation_agent],
)

# --- Run ---
try:
    result = Runner.run_sync(
        triage_agent,
        "I need a refund for order ORD-12345",
    )
    print(result.final_output)
except controlzero.PolicyDeniedError as e:
    print(f"Blocked by policy: {e.decision.reason}")

Example Policy​

Allow lookup tools, block refunds without approval, and control agent handoffs:

{
  "name": "support-agents-policy",
  "rules": [
    {
      "effect": "allow",
      "action": "tool:call",
      "resource": "tool/lookup_customer"
    },
    {
      "effect": "allow",
      "action": "tool:call",
      "resource": "tool/check_order_status"
    },
    {
      "effect": "deny",
      "action": "tool:call",
      "resource": "tool/issue_refund"
    },
    {
      "effect": "allow",
      "action": "tool:call",
      "resource": "tool/escalate_to_human"
    },
    {
      "effect": "allow",
      "action": "agent:handoff",
      "resource": "agent/Escalation Agent"
    }
  ]
}

What happens at runtime:

  • Customer lookup and order status: ALLOWED.
  • Refund issuance: DENIED. The agent receives a policy error and can escalate to a human instead.
  • Escalation to human: ALLOWED.
  • Handoff from Triage to Escalation: ALLOWED.

Handoff Enforcement​

Control handoffs between agents by enforcing the agent.handoff action:

def enforced_handoff(source_agent: str, target_agent: str):
    """Check policy before allowing agent handoff."""
    cz.guard(f"agent/{target_agent}", method="handoff", args={"source_agent": source_agent})

This prevents unauthorized agent-to-agent delegation in multi-agent workflows.

Gateway Alternative​

If you do not want to modify your agent code, point the OpenAI API base URL to the Control Zero gateway instead:

from agents import Agent, Runner
import openai

# Route all OpenAI calls through the Control Zero gateway
# Pass base_url to the client constructor (module-level assignment does not work in openai SDK v1.x)
client = openai.OpenAI(
    base_url="https://gateway.controlzero.ai/v1",
    default_headers={
        "X-ControlZero-Agent-ID": "support-agent",
        "X-ControlZero-API-Key": "cz_live_xxx",
    },
)

Alternatively, set the base URL via environment variable before starting your process:

OPENAI_BASE_URL=https://gateway.controlzero.ai/v1

The gateway enforces policies on all LLM calls and tool call responses at the network layer, with no SDK installation required. See the Gateway Guide for details.

Next Steps​