AutoGen Tutorial (Python): adding audit logs for intermediate developers

This tutorial shows how to add audit logging to an AutoGen Python agent setup so you can track every message, tool call, and final answer. You need this when agents are handling regulated workflows, internal approvals, or anything where you need a trace of what happened and why.

What You'll Need

• •Python 3.10+
• •autogen-agentchat
• •autogen-ext
• •An OpenAI API key in OPENAI_API_KEY
• •A writable log file path
• •Basic familiarity with AssistantAgent and UserProxyAgent

Install the packages:

pip install autogen-agentchat autogen-ext[openai]

Step-by-Step

1. •Start by creating a small audit logger that writes structured JSON lines. JSONL is easy to ship to SIEM tools, grep locally, or ingest into a database later.

import json
from datetime import datetime, timezone
from pathlib import Path

AUDIT_FILE = Path("audit.log")

def audit(event_type: str, **data):
    record = {
        "ts": datetime.now(timezone.utc).isoformat(),
        "event_type": event_type,
        **data,
    }
    with AUDIT_FILE.open("a", encoding="utf-8") as f:
        f.write(json.dumps(record, ensure_ascii=False) + "\n")

2. •Next, create a custom model client wrapper is not necessary here; instead, hook into AutoGen events by wrapping your agent interaction code. This keeps the logging layer explicit and avoids coupling it to internals that may change between versions.

import asyncio
import os

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.messages import TextMessage
from autogen_ext.models.openai import OpenAIChatCompletionClient

model_client = OpenAIChatCompletionClient(
    model="gpt-4o-mini",
    api_key=os.environ["OPENAI_API_KEY"],
)

agent = AssistantAgent(
    name="assistant",
    model_client=model_client,
)

3. •Add helper functions that log both the user input and the assistant output. For audit purposes, log the message role, content, conversation id, and any metadata you care about for compliance review.

import uuid

conversation_id = str(uuid.uuid4())

def log_user_message(content: str):
    audit(
        "user_message",
        conversation_id=conversation_id,
        role="user",
        content=content,
    )

def log_assistant_message(content: str):
    audit(
        "assistant_message",
        conversation_id=conversation_id,
        role="assistant",
        content=content,
    )

4. •Run the agent inside an async function and write an audit event before and after the call. If the run fails, capture the exception too; that matters just as much as successful completions in production systems.

async def main():
    user_input = "Summarize the risks of approving a loan without income verification."
    log_user_message(user_input)
    audit("run_started", conversation_id=conversation_id)

    try:
        result = await agent.run(task=user_input)
        final_text = result.messages[-1].content if result.messages else ""
        log_assistant_message(final_text)
        audit("run_completed", conversation_id=conversation_id)
        print(final_text)
    except Exception as e:
        audit(
            "run_failed",
            conversation_id=conversation_id,
            error_type=type(e).__name__,
            error_message=str(e),
        )
        raise

if __name__ == "__main__":
    asyncio.run(main())

5. •If you want better visibility, extend the logger to capture intermediate messages from multi-agent workflows. The pattern is the same: write an audit record whenever a message enters or leaves a boundary you care about.

def log_tool_call(tool_name: str, arguments: dict):
    audit(
        "tool_call",
        conversation_id=conversation_id,
        tool_name=tool_name,
        arguments=arguments,
    )

def log_tool_result(tool_name: str, result: str):
    audit(
        "tool_result",
        conversation_id=conversation_id,
        tool_name=tool_name,
        result=result,
    )

Testing It

Run the script once and confirm that audit.log is created in the working directory. You should see one JSON object per line with timestamps, event types, and your conversation ID.

Then intentionally break something, like removing OPENAI_API_KEY, and verify that run_failed gets written before the exception bubbles up. That tells you your failure path is covered.

If you use this in a multi-step workflow, make sure every user-visible action has a matching audit event. The goal is not just debugging; it’s reconstructing what happened later without guessing.

Next Steps

• •Add redaction for sensitive fields before writing logs to disk.
• •Send JSONL events to OpenTelemetry or your SIEM instead of local files.
• •Wrap this pattern into a reusable middleware layer for all agents in your codebase.

Keep learning

• •The complete AI Agents Roadmap — my full 8-step breakdown
• •Free: The AI Agent Starter Kit — PDF checklist + starter code
• •Work with me — I build AI for banks and insurance companies

By Cyprian Aarons, AI Consultant at Topiax.