Logging & Observability

Anvil includes a comprehensive logging system for observability into tool generation, execution, and self-healing.

Enabling Logging

Enable file-based logging:

from anvil import Anvil

anvil = Anvil(log_file="anvil.log")

Events are logged to the specified file in JSON Lines format.

Event Types

Anvil logs these events:

Event Type	Description
`tool_generated`	New tool code was generated
`tool_loaded`	Tool was loaded from cache
`tool_executed`	Tool ran successfully
`tool_failed`	Tool execution failed
`tool_healed`	Tool was regenerated after failure
`chain_started`	Tool chain began execution
`chain_completed`	Tool chain finished successfully
`chain_failed`	Tool chain failed

Accessing the Logger

logger = anvil.logger

# Get all events
events = logger.get_history()

# Filter by tool
events = logger.get_history(tool_name="search_tool")

# Filter by event type
events = logger.get_history(event_type="tool_failed")

# Filter by time
from datetime import datetime, timedelta
since = datetime.now() - timedelta(hours=1)
events = logger.get_history(since=since)

# Limit results
events = logger.get_history(limit=10)

Event Structure

Each event contains:

event = events[0]

print(event.timestamp)      # datetime
print(event.event_type)     # "tool_executed"
print(event.tool_name)      # "search_tool"
print(event.duration_ms)    # 150.5
print(event.metadata)       # {"version": "1.0", ...}

Statistics

Get aggregate statistics:

# Stats for a specific tool
stats = anvil.logger.get_stats("search_tool")
print(stats)
# {
#     "execution_count": 42,
#     "success_count": 40,
#     "failure_count": 2,
#     "heal_count": 1,
#     "success_rate": 0.952,
#     "avg_duration_ms": 125.3,
#     "min_duration_ms": 50.1,
#     "max_duration_ms": 450.2,
# }

# Stats for all tools
stats = anvil.logger.get_stats()

Recent Errors

Get recent failures:

errors = anvil.logger.get_recent_errors(limit=5)

for error in errors:
    print(f"{error.tool_name}: {error.metadata.get('error')}")

Log File Format

Logs are stored as JSON Lines (one JSON object per line):

{"timestamp": "2025-01-18T10:30:00Z", "event_type": "tool_generated", "tool_name": "search", "duration_ms": 1250.5, "metadata": {"version": "1.0", "intent": "Search the web"}}
{"timestamp": "2025-01-18T10:30:01Z", "event_type": "tool_executed", "tool_name": "search", "duration_ms": 150.2, "metadata": {"args": {"query": "test"}}}

This format is:

Easy to parse
Streamable
Compatible with log aggregation tools

Loading Historical Logs

Load events from a log file:

anvil.logger.load_from_file()

# Now get_history() includes historical events
all_events = anvil.logger.get_history()

Clearing Events

Clear in-memory events:

anvil.logger.clear()

This doesn’t affect the log file.

Custom Logging

Access events programmatically for custom handling:

from anvil.logger import AnvilEvent, EventType

# Create custom event
event = AnvilEvent(
    event_type=EventType.TOOL_EXECUTED,
    tool_name="custom_tool",
    duration_ms=100.0,
    metadata={"custom": "data"}
)

# Log it
anvil.logger.log(event)

Integration with External Systems

Export to JSON

import json

events = anvil.logger.get_history()
data = [event.to_dict() for event in events]

with open("export.json", "w") as f:
    json.dump(data, f, indent=2)

Stream to Monitoring

import requests

def send_to_monitoring(event):
    requests.post(
        "https://monitoring.example.com/events",
        json=event.to_dict()
    )

# Hook into logger
original_log = anvil.logger.log
def wrapped_log(event):
    original_log(event)
    send_to_monitoring(event)
anvil.logger.log = wrapped_log

Structured Logging

Combine with Python’s logging:

import logging
import json

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("anvil")

events = anvil.logger.get_history()
for event in events:
    logger.info(json.dumps(event.to_dict()))

Dashboard Ideas

With logged data, you can build dashboards showing:

Tool usage frequency
Success/failure rates
Average execution times
Self-healing effectiveness
Error patterns

Best Practices

1. Enable in Production

anvil = Anvil(log_file="/var/log/anvil/anvil.log")

2. Rotate Log Files

Use logrotate or similar:

/var/log/anvil/*.log {
    daily
    rotate 7
    compress
    missingok
}

3. Monitor Failure Rate

stats = anvil.logger.get_stats()
if stats["success_rate"] < 0.95:
    alert("Anvil success rate below 95%")

4. Track Self-Healing

heals = anvil.logger.get_history(event_type="tool_healed")
if len(heals) > 10:
    # Many heals might indicate unstable tools
    review_tools()