Agent Governance Foundation

Authorizing Tool Calls in AutoGen Multi-Agent Systems

Agent Governance Foundation
autogenauthorizationpythonintegrationmulti-agent

AutoGen's tool system is plain Python functions registered in a function_map. There's no BaseTool class to subclass and no framework-level hook for policy — a ConversableAgent calls whatever function is mapped to the name the model chose, full stop.

That's also what makes it straightforward to gate: since tools are just functions, wrapping them with an authorization check is idiomatic Python, and it works across every AutoGen version without needing a dedicated adapter.

Install

pip install agf-sdk

export AGF_API_KEY=agfk_your_key_here

No extras needed — the core SDK works with plain Python.

Authorization helper

AgentGovernance.authorize() checks policy synchronously and never raises on DENY — it returns an AuthResult you check explicitly:

import os
from agf import AgentGovernance

agf = AgentGovernance(
    api_key=os.environ["AGF_API_KEY"],
    org_id="org_acme",
)

def guarded_tool(action: str, resource: str, fn, *args, **kwargs):
    result = agf.authorize(
        agent_id="did:agf:agt_01abc",
        action=action,
        resource=resource,
    )
    if not result.allowed:
        return f"Action blocked by policy: {result.reason}"
    return fn(*args, **kwargs)

Wrapping a tool

Register the guarded version in function_map. AutoGen calls the wrapper; the wrapper checks the PDP before calling the real function:

import autogen

def read_file(path: str) -> str:
    return open(path).read()

def guarded_read_file(path: str) -> str:
    return guarded_tool("read:file", path, read_file, path)

# Register guarded version as an AutoGen function tool
assistant = autogen.AssistantAgent(
    name="assistant",
    llm_config={
        "functions": [
            {
                "name": "read_file",
                "description": "Read a file from disk",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "path": {"type": "string", "description": "Absolute path to read"}
                    },
                    "required": ["path"],
                },
            }
        ],
        "config_list": autogen.config_list_from_json("OAI_CONFIG_LIST"),
    },
)

user_proxy = autogen.UserProxyAgent(
    name="user_proxy",
    human_input_mode="NEVER",
    function_map={"read_file": guarded_read_file},
)

Group chats with per-agent policy

Register every agent you deploy in AGF with a distinct agent_id first — the ID has to match a registered agent for the policy check to resolve. Different agents in the same group chat can then carry different policies for the same action:

import autogen

# Each agent uses a different agf agent_id for per-agent policy
def make_guarded_tool(agent_id: str, action: str, resource: str, fn):
    def wrapper(*args, **kwargs):
        result = agf.authorize(agent_id=agent_id, action=action, resource=resource)
        if not result.allowed:
            return f"Blocked: {result.reason}"
        return fn(*args, **kwargs)
    return wrapper

researcher = autogen.AssistantAgent(name="researcher", llm_config=llm_config)
writer     = autogen.AssistantAgent(name="writer",     llm_config=llm_config)
manager    = autogen.GroupChatManager(
    groupchat=autogen.GroupChat(agents=[researcher, writer], messages=[], max_round=10),
    llm_config=llm_config,
)

user_proxy = autogen.UserProxyAgent(
    name="user_proxy",
    human_input_mode="NEVER",
    function_map={
        "search_web":  make_guarded_tool("did:agf:agt_researcher", "read:web",  "internet",  search_web),
        "write_report": make_guarded_tool("did:agf:agt_writer",    "write:file", "reports/",  write_report),
    },
)

user_proxy.initiate_chat(manager, message="Research and write a report on Q1 churn.")

For systems where one agent's authority derives from another (a manager delegating to a specialist), pass the full delegation chain via AGFClient.decide() directly so the PDP evaluates cross-agent authority instead of treating each agent_id as independently authorized.

Async agents

For async AutoGen workflows, use AGFClient directly — it's an async client built on httpx:

import asyncio
from agf import AGFClient

client = AGFClient(api_key=os.environ["AGF_API_KEY"])

async def guarded_tool_async(action: str, resource: str, fn, *args, **kwargs):
    response = await client.decide({
        "chain": ["did:agf:agt_01abc"],
        "action": {"type": action, "resource": resource},
        "audience": "internal",
    })
    if response["data"]["decision"] != "ALLOW":
        raise PermissionError(f"Denied: {response['data'].get('reasoning', '')}")
    return await fn(*args, **kwargs)

What you get

Every authorization call produces a signed audit artifact, exportable from the dashboard or GET /v1/audit — the same artifact structure covered in the LangChain integration post.

Next steps

agf-sdk is open source. PyPI →

All posts