lyzr-audit-logger 0.1.5

Audit logging library for microservices

Audit Logger

A Python package for audit logging across microservices. Provides async, non-blocking audit log capture with MongoDB storage.

Installation

pip install audit-logger

Or install from source:

# Using HTTPS
pip install git+https://github.com/NeuralgoLyzr/audit-logger.git

# Using SSH
pip install git+ssh://[e-mail]/NeuralgoLyzr/audit-logger.git

For local development:

cd audit_logger
pip install -e .

Quick Start

1. Initialize at Application Startup

from audit_logger import init_audit_logger

# Option 1: Use environment variable AUDIT_MONGO_URL (defaults to mongodb://localhost:27017)
audit_manager, audit_logger = init_audit_logger()

# Option 2: Explicit configuration
audit_manager, audit_logger = init_audit_logger(
    mongo_url="mongodb://audit-db:27017",
    db_name="my_service_audit",
    collection_name="audit_logs"
)

# Create indexes (call once on startup)
await audit_manager.ensure_indexes()

2. Set Context in Middleware

from audit_logger import set_audit_context, clear_audit_context

@app.middleware("http")
async def audit_middleware(request: Request, call_next):
    # Set context from auth info
    set_audit_context(
        org_id=request.state.auth_user.org_id,
        user_id=request.state.auth_user.user_id,
        api_key=request.headers.get("x-api-key"),
        ip_address=request.client.host,
    )
    try:
        return await call_next(request)
    finally:
        clear_audit_context()

3. Log Audit Events

from audit_logger import audit_logger, AuditResource

# Log a create action (non-blocking, returns immediately)
audit_logger.log_create(
    resource_type=AuditResource.AGENT,
    resource_id="agent_123",
    resource_name="My Agent",
    metadata={"tools_count": 5}
)

# Log an update action
from audit_logger import AuditChange

audit_logger.log_update(
    resource_type=AuditResource.AGENT,
    resource_id="agent_123",
    changes=[
        AuditChange(field="name", old_value="Old Name", new_value="New Name"),
        AuditChange(field="description", old_value=None, new_value="New description"),
    ]
)

# Log a delete action
audit_logger.log_delete(
    resource_type=AuditResource.AGENT,
    resource_id="agent_123",
    resource_name="My Agent"
)

4. Shutdown Gracefully

from audit_logger import shutdown_audit_logger

@app.on_event("shutdown")
async def shutdown():
    await shutdown_audit_logger()

Advanced Usage

Direct Manager Access

For more control, use the AuditLogManager directly:

from audit_logger import get_audit_log_manager, AuditResource, AuditResult

manager = get_audit_log_manager()

# Log with full control
manager.log_create(
    org_id="org_123",
    resource_type=AuditResource.TOOL,
    resource_id="tool_456",
    user_id="user_789",
    api_key="sk-xxx",
    ip_address="192.168.1.1",
    resource_name="My Tool",
    metadata={"type": "openapi"},
    result=AuditResult.SUCCESS,
)

# Log inference events
manager.log_inference(
    org_id="org_123",
    agent_id="agent_456",
    session_id="session_789",
    user_id="user_123",
    result=AuditResult.SUCCESS,
    metadata={"model": "gpt-4", "tokens": 1500}
)

# Log guardrail violations
from audit_logger import GuardrailViolation, AuditSeverity

manager.log_guardrail_violation(
    org_id="org_123",
    session_id="session_789",
    violations=[
        GuardrailViolation(
            violation_type="pii",
            severity=AuditSeverity.MEDIUM,
            details={"entities": ["email", "phone"]}
        )
    ]
)

Query Audit Logs

from audit_logger import get_audit_log_manager, AuditLogQuery, AuditResource

manager = get_audit_log_manager()

# Query with filters
logs = await manager.query(AuditLogQuery(
    org_id="org_123",
    resource_type=AuditResource.AGENT,
    limit=50,
))

# Get history for a specific resource
history = await manager.get_by_resource(
    org_id="org_123",
    resource_type=AuditResource.AGENT,
    resource_id="agent_456",
)

# Get session activity
session_logs = await manager.get_by_session(session_id="session_789")

# Count matching logs
count = await manager.count(AuditLogQuery(org_id="org_123"))

Using the Decorator

from audit_logger import audit_action, AuditAction, AuditResource

@audit_action(
    action=AuditAction.DELETE,
    resource_type=AuditResource.AGENT,
    resource_id_param="agent_id",
)
async def delete_agent(agent_id: str, _audit_manager=None):
    # Delete logic here
    pass

Context Manager

from audit_logger import audit_context

async with audit_context(org_id="org_123", user_id="user_456") as request_id:
    # All audit logs in this block will have the context set
    audit_logger.log_create(AuditResource.AGENT, "agent_789")

Configuration

Environment Variables

Variable	Default	Description
AUDIT_MONGO_URL	mongodb://localhost:27017	MongoDB connection URL

Programmatic Configuration

from audit_logger import init_audit_logger

# Full configuration
manager, logger = init_audit_logger(
    mongo_url="mongodb://audit-db:27017",
    db_name="lyzr_audit_logs",
    collection_name="audit_logs",
)

Using Existing Collection

If you already have a MongoDB collection configured:

from motor.motor_asyncio import AsyncIOMotorClient
from audit_logger import init_audit_logger

client = AsyncIOMotorClient("mongodb://localhost:27017")
collection = client["my_db"]["audit_logs"]

manager, logger = init_audit_logger(collection=collection)

Available Enums

AuditAction

• CREATE, READ, UPDATE, DELETE
• EXECUTE, LOGIN, LOGOUT
• ACCESS_DENIED, EXPORT, IMPORT

AuditResource

• AGENT, TOOL, PROVIDER, WORKFLOW
• SESSION, MESSAGE, CONTEXT, MEMORY
• CREDENTIAL, USER, ORGANIZATION, API_KEY
• INFERENCE, GUARDRAIL, ARTIFACT

AuditResult

• SUCCESS, FAILURE, BLOCKED, PARTIAL

AuditSeverity

• LOW, MEDIUM, HIGH, CRITICAL

MongoDB Schema

{
    "_id": ObjectId,
    "timestamp": ISODate,
    "actor": {
        "org_id": "org_123",
        "user_id": "user_456",
        "api_key_hash": "a1b2c3d4...xyz",
        "ip_address": "192.168.1.1"
    },
    "action": "create",
    "target": {
        "resource_type": "agent",
        "resource_id": "agent_789",
        "resource_name": "Support Bot"
    },
    "result": "success",
    "error_message": null,
    "session_id": "session_abc",
    "request_id": "req_def",
    "changes": [...],
    "guardrail_violations": [...],
    "metadata": {...},
    "severity": "low"
}

Non-Blocking Writes

All logging methods are non-blocking by default. They use asyncio.create_task() to write to MongoDB in the background without blocking your request handlers.

For synchronous writes (when you need confirmation):

log_id = await manager.log_sync(entry)

License

MIT