CrewAI Tutorial (Python): adding audit logs for advanced developers

This tutorial shows you how to add structured audit logs to a CrewAI project so every agent action, tool call, and final output is traceable. You need this when you’re building agent workflows for regulated environments like banking or insurance, where you need an evidence trail for reviews, incident response, and compliance.

What You'll Need

• •Python 3.10+
• •crewai
• •python-dotenv
• •An OpenAI API key set as OPENAI_API_KEY
• •A writable local directory for log files
• •Basic CrewAI familiarity: agents, tasks, crews, and tools

Install the packages:

pip install crewai python-dotenv

Step-by-Step

1. •Start by creating a dedicated audit logger. Keep it separate from application logs so you can ship it to a different sink later, such as S3, Datadog, or a SIEM. Use JSON lines because they’re easy to parse and cheap to append.

import json
import logging
from datetime import datetime, timezone

logger = logging.getLogger("audit")
logger.setLevel(logging.INFO)

handler = logging.FileHandler("audit.log")
handler.setFormatter(logging.Formatter("%(message)s"))
logger.addHandler(handler)

def audit(event_type: str, **payload):
    record = {
        "ts": datetime.now(timezone.utc).isoformat(),
        "event_type": event_type,
        **payload,
    }
    logger.info(json.dumps(record))

2. •Load your API key and define a small CrewAI setup. The important part here is that we’ll call the audit function at every boundary we care about: crew start, task execution, tool usage, and completion.

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

load_dotenv()

researcher = Agent(
    role="Research Analyst",
    goal="Summarize customer-facing policy changes",
    backstory="You work in a regulated financial services environment.",
    verbose=True,
)

task = Task(
    description="Summarize the impact of a policy update in 3 bullets.",
    expected_output="Three concise bullets.",
    agent=researcher,
)

3. •Wrap execution with explicit audit events. In production you want deterministic records: who ran what, when it started, when it ended, and whether it failed. This example keeps the logging in the application layer so you don’t have to patch CrewAI internals.

import os

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

audit(
    "crew_start",
    crew_name="policy_summary_crew",
    process="sequential",
    task_count=1,
)

try:
    result = crew.kickoff()
    audit(
        "crew_success",
        crew_name="policy_summary_crew",
        result=str(result),
    )
except Exception as exc:
    audit(
        "crew_error",
        crew_name="policy_summary_crew",
        error_type=type(exc).__name__,
        error_message=str(exc),
    )
    raise

print(result)

4. •Add tool-level auditing if your agents call external systems. This is where most compliance teams care: what data left the system, what came back, and whether the call succeeded. The cleanest pattern is to wrap your tool function and log before and after the actual operation.

from crewai.tools import BaseTool

class AuditedEchoTool(BaseTool):
    name: str = "audited_echo"
    description: str = "Echoes input text and writes audit records."

    def _run(self, text: str) -> str:
        audit("tool_call_start", tool=self.name, input=text)
        output = f"echo:{text}"
        audit("tool_call_end", tool=self.name, output=output)
        return output

tool_agent = Agent(
    role="Operations Analyst",
    goal="Use tools carefully and record actions.",
    backstory="You must preserve an auditable trail of every external action.",
    tools=[AuditedEchoTool()],
)

5. •If you want stronger traceability across multiple tasks, attach correlation IDs yourself. That gives you one ID per run that can be threaded through logs, downstream services, and incident reports.

import uuid

run_id = str(uuid.uuid4())

audit("run_created", run_id=run_id)

task2 = Task(
    description=f"Use the audited tool once for run {run_id}.",
    expected_output="A single echoed string.",
    agent=tool_agent,
)

crew2 = Crew(agents=[tool_agent], tasks=[task2], process=Process.sequential)
audit("crew_start", run_id=run_id, crew_name="tool_crew")

result2 = crew2.kickoff()
audit("crew_success", run_id=run_id, result=str(result2))

print(result2)

Testing It

Run the script and confirm that audit.log is created in the working directory. You should see one JSON object per line with timestamps, event types, and payload fields like crew_start, tool_call_start, and crew_success.

Then trigger a failure on purpose by giving the agent an invalid configuration or breaking a tool call. Verify that crew_error is written before the exception bubbles up.

For production readiness, open the log file and check that every line parses as valid JSON. If your downstream system expects schema enforcement, add a validator for required keys like ts, event_type, and run_id.

Next Steps

• •Add request-scoped metadata like user ID, tenant ID, case ID, or policy number to every audit event.
• •Ship logs to a centralized store using logging.handlers.SysLogHandler, Fluent Bit, or OpenTelemetry.
• •Add tamper resistance by signing log batches or writing them to append-only storage.

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.