Skip to content

AutoGen

AutoGen

Wrap AutoGen's function_map so every function call is authorised before execution. DENY decisions return a safe error string rather than raising, keeping the conversation intact.

# autogen_xybern.py
import requests
import autogen

XYBERN_API_KEY = "xb_your_api_key"

def authorise_action(agent_name: str, tool: str, params: dict) -> bool:
    resp = requests.post(
        "https://www.xybern.com/api/v1/enforce/intercept",
        headers={"X-API-Key": XYBERN_API_KEY},
        json={
            "agent_id": agent_name,
            "action": {"type": "tool_call", "tool": tool, "parameters": params},
            "context": {"framework": "autogen"}
        }
    ).json()
    return resp["decision"] == "ALLOW"

config_list = [{"model": "gpt-4", "api_key": "your_openai_key"}]

def guarded_function_map(agent_name: str, function_map: dict) -> dict:
    """Wrap an AutoGen function_map with Xybern checks."""
    guarded = {}
    for fn_name, fn in function_map.items():
        def make_guarded(name, original_fn):
            def guarded_fn(**kwargs):
                if not authorise_action(agent_name, name, kwargs):
                    return f"Action '{name}' was denied by Xybern authorisation."
                return original_fn(**kwargs)
            return guarded_fn
        guarded[fn_name] = make_guarded(fn_name, fn)
    return guarded

def query_database(table: str, query: str):
    return f"Results from {table}: [...]"

analyst = autogen.AssistantAgent(
    name="analyst_agent",
    llm_config={"config_list": config_list}
)

user_proxy = autogen.UserProxyAgent(
    name="user_proxy",
    human_input_mode="NEVER",
    function_map=guarded_function_map("analyst_agent", {"query_database": query_database})
)

user_proxy.initiate_chat(analyst, message="Query the customers table and summarise the data.")