CrewAI Tutorial (Python): adding audit logs for beginners

This tutorial shows you how to add audit logging to a CrewAI Python project so every agent action, task result, and failure gets recorded. You need this when you want traceability for compliance, debugging, incident review, or just to answer “what happened?” after an agent makes a bad call.

What You'll Need

• •Python 3.10 or newer
• •crewai
• •python-dotenv
• •An LLM API key, such as:
  • •OPENAI_API_KEY
• •A basic CrewAI project with at least one agent and one task
• •Permission to write local log files

Install the packages first:

pip install crewai python-dotenv

Step-by-Step

1. •Create a small audit logger first.
   Keep it simple: write JSON lines to a file so every event is easy to parse later with scripts, SIEM tools, or log pipelines.

import json
from datetime import datetime
from pathlib import Path

AUDIT_FILE = Path("audit.log")

def audit(event_type: str, payload: dict) -> None:
    record = {
        "timestamp": datetime.utcnow().isoformat() + "Z",
        "event_type": event_type,
        "payload": payload,
    }
    with AUDIT_FILE.open("a", encoding="utf-8") as f:
        f.write(json.dumps(record) + "\n")

2. •Load your API key and create the CrewAI objects.
   This example uses OpenAI through CrewAI’s standard LLM integration and keeps the setup minimal enough for beginners.

import os
from dotenv import load_dotenv
from crewai import Agent, Task, Crew, Process, LLM

load_dotenv()

llm = LLM(
    model="gpt-4o-mini",
    api_key=os.getenv("OPENAI_API_KEY"),
)

researcher = Agent(
    role="Research Analyst",
    goal="Summarize a topic clearly",
    backstory="You produce concise business-friendly summaries.",
    llm=llm,
    verbose=False,
)

3. •Wrap task execution with audit events.
   CrewAI does not give you “audit logs” out of the box in the format most teams need, so the practical pattern is to log before and after execution, plus any exception path.

task = Task(
    description="Summarize why audit logs matter in AI systems.",
    expected_output="A short explanation of audit logging benefits.",
    agent=researcher,
)

crew = Crew(
    agents=[researcher],
    tasks=[task],
    process=Process.sequential,
    verbose=False,
)

audit("crew_run_started", {"task_count": len(crew.tasks), "process": "sequential"})
try:
    result = crew.kickoff()
    audit("crew_run_succeeded", {"result": str(result)})
except Exception as exc:
    audit("crew_run_failed", {"error": type(exc).__name__, "message": str(exc)})
    raise

4. •Add per-task context so the log is useful later.
   A plain “success” entry is not enough in production. Log identifiers, task names, and any metadata you care about before the run starts.

audit(
    "task_registered",
    {
        "agent_role": researcher.role,
        "task_description": task.description,
        "expected_output": task.expected_output,
    },
)

print(result)

5. •Make the logger reusable across multiple runs.
   If you run this from an API or worker process, keep the same helper and call it around each crew execution. That gives you a consistent audit trail without changing CrewAI internals.

def run_crew_with_audit() -> str:
    audit("run_started", {"workflow": "basic_summary"})
    try:
        output = crew.kickoff()
        audit("run_completed", {"workflow": "basic_summary", "output": str(output)})
        return str(output)
    except Exception as exc:
        audit("run_failed", {"workflow": "basic_summary", "error": str(exc)})
        raise

if __name__ == "__main__":
    print(run_crew_with_audit())

Testing It

Run the script once with a valid OPENAI_API_KEY set in your environment. You should see normal CrewAI output on stdout and a new audit.log file in the same directory.

Open audit.log and confirm that each line is valid JSON with timestamps and event types like task_registered, crew_run_started, and crew_run_succeeded. If you intentionally break the API key or network access, verify that crew_run_failed gets written before the exception bubbles up.

For a quick sanity check, parse the file with Python and make sure every line loads cleanly:

import json

with open("audit.log", "r", encoding="utf-8") as f:
    for line in f:
        record = json.loads(line)
        print(record["event_type"], record["timestamp"])

Next Steps

• •Add request IDs or correlation IDs so one user action can be traced across multiple crews.
• •Send the same JSON events to CloudWatch, ELK, Datadog, or Splunk instead of only writing local files.
• •Extend the pattern to log tool calls, token usage, and human approval steps for regulated workflows.

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.