Introduction to Building a Log Analysis Agent with Claude Code
Modern applications generate enormous volumes of logs across distributed services, containers, and cloud infrastructure. Manually sifting through these logs to find anomalies, trace errors, and understand incidents is time-consuming and error-prone. By building a log analysis agent with Claude Code, you can automate the detection, triage, and explanation of issues hidden in your log data, turning raw text into actionable engineering insights.
This guide walks you through everything you need to know: what a log analysis agent is, why it matters, how to build one step by step, and the best practices that separate a toy demo from a production-grade tool.
What Is a Log Analysis Agent?
A log analysis agent is an AI-powered system that reads log files or log streams, identifies patterns and anomalies, correlates related events, and produces human-readable explanations of what happened and why. Unlike traditional rule-based log monitors that fire on keyword matches, an agent built on Claude Code can reason about context, follow multi-step investigation workflows, and even propose remediation steps.
Claude Code is Anthropic's CLI-based coding agent that can read files, run shell commands, and execute scripts. This makes it an ideal foundation for a log analysis agent because it can combine file I/O, command-line tools like grep and jq, and Claude's language understanding in a single workflow.
Core Capabilities of a Log Analysis Agent
- Error detection: Identify stack traces, exceptions, and failed requests across services.
- Pattern recognition: Spot recurring error signatures, spikes in error rates, and unusual latency.
- Correlation: Link related log lines across services using request IDs, trace IDs, or timestamps.
- Root cause analysis: Trace an error back to its originating cause through a chain of log events.
- Summarization: Produce concise incident summaries for on-call engineers.
- Remediation suggestions: Recommend fixes based on the error type and historical context.
Why Log Analysis Agents Matter
Traditional log management tools like ELK stacks, Splunk, or Datadog are excellent at indexing and querying logs, but they still require a human to write the queries, interpret the results, and connect the dots. This creates several pain points that an agent-based approach directly addresses.
Speed of Incident Response
During an outage, every minute counts. An agent can scan thousands of log lines in seconds, identify the first error in a cascade, and present a summary before a human engineer has even opened their terminal. This compresses mean time to resolution (MTTR) significantly.
Handling Unstructured and Semi-Structured Logs
Many services emit logs in inconsistent formats — mixed JSON and plain text, varying timestamp formats, multi-line stack traces. Traditional parsers struggle with this. Claude's language understanding handles format variation gracefully, extracting meaning even from messy logs.
Reducing Alert Fatigue
Rule-based monitors generate noisy alerts. An agent can distinguish between a transient network blip and a genuine service failure by reasoning about the surrounding context, reducing false positives and letting engineers focus on real issues.
Knowledge Retention
When a senior engineer investigates an incident, much of their reasoning lives in their head. An agent can document its investigation path, creating a reusable record that helps the team respond faster to similar issues in the future.
Prerequisites and Setup
Before building the agent, ensure you have the following in place.
Prerequisites
- Node.js 18+ and npm installed
- Claude Code CLI installed and authenticated (
npm install -g @anthropic-ai/claude-code) - Python 3.10+ for log processing scripts
- Sample log files for testing (we will generate these)
Project Structure
Create a project directory with the following layout:
log-analysis-agent/
├── logs/
│ ├── app.log
│ ├── auth-service.log
│ └── payment-service.log
├── scripts/
│ ├── parse_logs.py
│ ├── correlate.py
│ └── generate_sample_logs.py
├── prompts/
│ ├── system_prompt.md
│ └── analysis_prompt.md
├── CLAUDE.md
└── analyze.sh
Step 1: Generating Sample Log Data
To build and test the agent, you need realistic log data. Create a Python script that generates sample logs simulating a microservices architecture with an authentication service, a payment service, and a main application.
# scripts/generate_sample_logs.py
import random
import json
from datetime import datetime, timedelta
SERVICES = ["auth-service", "payment-service", "app"]
LOG_LEVELS = ["INFO", "WARN", "ERROR", "DEBUG"]
ERROR_MESSAGES = [
"NullPointerException at UserService.java:142",
"Connection refused to database at 10.0.1.5:5432",
"Timeout waiting for auth-service response",
"Payment gateway returned 502 Bad Gateway",
"JWT token expired for user_id={user_id}",
"Failed to deserialize response from payment-service",
"Circuit breaker opened for payment-service",
"Redis cache miss rate exceeded 80%",
]
def generate_log_entry(timestamp, service, level, message, trace_id=None):
entry = {
"timestamp": timestamp.isoformat(),
"level": level,
"service": service,
"message": message,
}
if trace_id:
entry["trace_id"] = trace_id
if level == "ERROR":
entry["stack_trace"] = f"at com.app.{service}.handleRequest({service}.java:{random.randint(50, 300)})"
return json.dumps(entry)
def generate_logs():
start_time = datetime(2024, 1, 15, 9, 0, 0)
logs = []
for i in range(500):
ts = start_time + timedelta(seconds=i * random.randint(1, 10))
service = random.choice(SERVICES)
# Inject an incident around log 350-380
if 350 <= i <= 380:
level = random.choice(["ERROR", "WARN", "ERROR"])
msg = random.choice(ERROR_MESSAGES).format(user_id=random.randint(1000, 9999))
trace_id = f"trace-{350 + (i - 350) // 5}"
logs.append(generate_log_entry(ts, service, level, msg, trace_id))
else:
level = random.choices(LOG_LEVELS, weights=[50, 20, 5, 25])[0]
if level == "ERROR":
msg = random.choice(ERROR_MESSAGES).format(user_id=random.randint(1000, 9999))
elif level == "WARN":
msg = random.choice([
"High memory usage detected: 85%",
"Slow query detected: 2.3s execution time",
"Rate limit approaching threshold",
])
else:
msg = random.choice([
f"Request processed successfully for user_id={random.randint(1000, 9999)}",
"Health check passed",
f"Cache hit for key=user_session_{random.randint(1000, 9999)}",
])
logs.append(generate_log_entry(ts, service, level, msg))
return logs
if __name__ == "__main__":
logs = generate_logs()
with open("logs/app.log", "w") as f:
for line in logs:
f.write(line + "\n")
print(f"Generated {len(logs)} log entries in logs/app.log")
Run the script to generate test data:
python scripts/generate_sample_logs.pyStep 2: Writing the CLAUDE.md Configuration
The CLAUDE.md file is the heart of your Claude Code agent. It defines the agent's identity, behavior, available tools, and workflow. This file is automatically read by Claude Code when it starts in the project directory.
# CLAUDE.md
# Log Analysis Agent
You are a senior site reliability engineer (SRE) specialized in log analysis.
Your job is to investigate log files, identify anomalies and errors, correlate
related events, and produce clear incident reports.
## Your Capabilities
You can:
- Read log files using `cat`, `head`, `tail`, and `grep`
- Parse JSON log entries using `jq`
- Run Python scripts in the `scripts/` directory
- Write analysis reports to the `reports/` directory
## Investigation Workflow
When asked to analyze logs, follow this workflow:
1. **Survey**: Read the first and last 20 lines of each log file to understand
the time range and format.
2. **Error scan**: Use `grep` to find all ERROR and WARN level entries.
3. **Correlation**: Group errors by trace_id and timestamp proximity.
4. **Root cause**: Identify the first error in each error cascade.
5. **Report**: Write a structured report to `reports/incident_report.md`.
## Output Format
Your final report must include:
- **Incident Summary**: One paragraph describing what happened.
- **Timeline**: Chronological list of key events with timestamps.
- **Root Cause**: The originating error and its cause.
- **Affected Services**: List of services impacted.
- **Recommended Actions**: Specific steps to fix or prevent the issue.
## Rules
- Always cite specific log lines with timestamps as evidence.
- Do not speculate without supporting log evidence.
- If logs are insufficient to determine root cause, state that clearly.
- Prioritize errors with trace_ids as they indicate user-facing requests.
Step 3: Building the Log Parsing Script
While Claude Code can use grep and jq directly, a dedicated parsing script gives the agent a powerful tool for structured analysis. This script extracts errors, groups them by trace ID, and identifies error cascades.
# scripts/parse_logs.py
import json
import sys
from collections import defaultdict
from datetime import datetime
def parse_log_file(filepath):
"""Parse a JSON-lines log file and return structured entries."""
entries = []
with open(filepath, 'r') as f:
for line_num, line in enumerate(f, 1):
line = line.strip()
if not line:
continue
try:
entry = json.loads(line)
entry['_line_number'] = line_num
entry['_source_file'] = filepath
entries.append(entry)
except json.JSONDecodeError:
# Handle non-JSON log lines
entries.append({
'timestamp': None,
'level': 'UNKNOWN',
'message': line,
'_line_number': line_num,
'_source_file': filepath,
'_parse_error': True,
})
return entries
def filter_by_level(entries, levels):
"""Filter entries by log level."""
return [e for e in entries if e.get('level') in levels]
def group_by_trace_id(entries):
"""Group log entries by trace_id."""
groups = defaultdict(list)
for entry in entries:
trace_id = entry.get('trace_id')
if trace_id:
groups[trace_id].append(entry)
return dict(groups)
def find_error_cascades(entries, window_seconds=30):
"""Find clusters of errors that occur within a time window."""
errors = filter_by_level(entries, ['ERROR', 'WARN'])
errors.sort(key=lambda e: e.get('timestamp', ''))
cascades = []
current_cascade = []
for error in errors:
if not current_cascade:
current_cascade.append(error)
continue
last_ts = datetime.fromisoformat(current_cascade[-1]['timestamp'])
curr_ts = datetime.fromisoformat(error['timestamp'])
if (curr_ts - last_ts).total_seconds() <= window_seconds:
current_cascade.append(error)
else:
if len(current_cascade) >= 3:
cascades.append(current_cascade)
current_cascade = [error]
if len(current_cascade) >= 3:
cascades.append(current_cascade)
return cascades
def print_summary(entries):
"""Print a summary of the log entries."""
total = len(entries)
by_level = defaultdict(int)
by_service = defaultdict(int)
for entry in entries:
by_level[entry.get('level', 'UNKNOWN')] += 1
by_service[entry.get('service', 'unknown')] += 1
print(f"=== Log Summary ===")
print(f"Total entries: {total}")
print(f"\nBy level:")
for level, count in sorted(by_level.items()):
print(f" {level}: {count}")
print(f"\nBy service:")
for service, count in sorted(by_service.items()):
print(f" {service}: {count}")
errors = filter_by_level(entries, ['ERROR'])
print(f"\nTotal errors: {len(errors)}")
cascades = find_error_cascades(entries)
print(f"Error cascades found: {len(cascades)}")
for i, cascade in enumerate(cascades):
print(f"\n Cascade {i+1}:")
print(f" Start: {cascade[0].get('timestamp')}")
print(f" End: {cascade[-1].get('timestamp')}")
print(f" Events: {len(cascade)}")
print(f" First error: {cascade[0].get('message')}")
if __name__ == "__main__":
if len(sys.argv) < 2:
print("Usage: python parse_logs.py <logfile> [--summary|--errors|--cascades|--traces]")
sys.exit(1)
filepath = sys.argv[1]
mode = sys.argv[2] if len(sys.argv) > 2 else '--summary'
entries = parse_log_file(filepath)
if mode == '--summary':
print_summary(entries)
elif mode == '--errors':
errors = filter_by_level(entries, ['ERROR'])
for e in errors:
print(json.dumps(e, indent=2))
elif mode == '--cascades':
cascades = find_error_cascades(entries)
for i, cascade in enumerate(cascades):
print(f"\n--- Cascade {i+1} ---")
for entry in cascade:
print(f" [{entry.get('timestamp')}] {entry.get('service')}: {entry.get('message')}")
elif mode == '--traces':
traces = group_by_trace_id(entries)
for trace_id, trace_entries in traces.items():
print(f"\n--- Trace: {trace_id} ---")
for entry in trace_entries:
print(f" [{entry.get('timestamp')}] {entry.get('level')} {entry.get('service')}: {entry.get('message')}")
Test the parsing script:
python scripts/parse_logs.py logs/app.log --summary
python scripts/parse_logs.py logs/app.log --cascades
python scripts/parse_logs.py logs/app.log --tracesStep 4: Building the Correlation Script
For multi-service architectures, you need to correlate logs across different files. This script merges logs from multiple sources, sorts them by timestamp, and reconstructs request flows using trace IDs.
# scripts/correlate.py
import json
import sys
import glob
from datetime import datetime
from collections import defaultdict
def load_all_logs(log_dir="logs"):
"""Load and merge all log files from a directory."""
all_entries = []
for filepath in sorted(glob.glob(f"{log_dir}/*.log")):
with open(filepath, 'r') as f:
for line in f:
line = line.strip()
if not line:
continue
try:
entry = json.loads(line)
entry['_source_file'] = filepath
all_entries.append(entry)
except json.JSONDecodeError:
continue
return all_entries
def sort_by_timestamp(entries):
"""Sort entries chronologically."""
return sorted(entries, key=lambda e: e.get('timestamp', ''))
def reconstruct_request_flows(entries):
"""Reconstruct request flows using trace IDs."""
flows = defaultdict(list)
for entry in entries:
trace_id = entry.get('trace_id')
if trace_id:
flows[trace_id].append(entry)
# Sort each flow by timestamp
for trace_id in flows:
flows[trace_id].sort(key=lambda e: e.get('timestamp', ''))
return dict(flows)
def find_cross_service_errors(entries):
"""Find errors that span multiple services."""
flows = reconstruct_request_flows(entries)
cross_service = []
for trace_id, flow in flows.items():
services_involved = set(e.get('service') for e in flow)
has_error = any(e.get('level') == 'ERROR' for e in flow)
if has_error and len(services_involved) > 1:
cross_service.append({
'trace_id': trace_id,
'services': list(services_involved),
'flow': flow,
})
return cross_service
def print_correlation_report(entries):
"""Print a cross-service correlation report."""
flows = reconstruct_request_flows(entries)
cross_service = find_cross_service_errors(entries)
print("=== Cross-Service Correlation Report ===")
print(f"Total trace IDs found: {len(flows)}")
print(f"Cross-service error flows: {len(cross_service)}")
for item in cross_service:
print(f"\n--- Trace: {item['trace_id']} ---")
print(f"Services involved: {', '.join(item['services'])}")
print("Request flow:")
for entry in item['flow']:
level = entry.get('level', '?')
ts = entry.get('timestamp', '?')
service = entry.get('service', '?')
msg = entry.get('message', '?')
marker = " *** ERROR ***" if level == 'ERROR' else ""
print(f" [{ts}] {level} {service}: {msg}{marker}")
if __name__ == "__main__":
log_dir = sys.argv[1] if len(sys.argv) > 1 else "logs"
entries = load_all_logs(log_dir)
sorted_entries = sort_by_timestamp(entries)
print_correlation_report(sorted_entries)
Step 5: Creating the Analysis Shell Script
The shell script serves as the entry point for running the agent. It sets up the environment, ensures the reports directory exists, and invokes Claude Code with the appropriate prompt.
#!/bin/bash
# analyze.sh - Entry point for the log analysis agent
set -e
LOG_DIR="${1:-logs}"
REPORT_DIR="reports"
# Create reports directory if it doesn't exist
mkdir -p "$REPORT_DIR"
# Generate a timestamped report filename
TIMESTAMP=$(date +"%Y%m%d_%H%M%S")
REPORT_FILE="$REPORT_DIR/incident_report_${TIMESTAMP}.md"
echo "Starting log analysis agent..."
echo "Log directory: $LOG_DIR"
echo "Report will be saved to: $REPORT_FILE"
echo ""
# Run Claude Code with the analysis prompt
claude --print "$(cat <<'PROMPT'
Analyze the log files in the logs/ directory. Follow the investigation workflow
defined in CLAUDE.md. Use the scripts in scripts/ to help with parsing and
correlation. Write your final incident report to the file specified below.
Steps:
1. Run: python scripts/parse_logs.py logs/app.log --summary
2. Run: python scripts/parse_logs.py logs/app.log --cascades
3. Run: python scripts/parse_logs.py logs/app.log --traces
4. Run: python scripts/correlate.py logs/
5. Read the key error log lines directly with grep for additional context.
6. Write a complete incident report following the format in CLAUDE.md.
Save the report to: REPORT_FILE_PLACEHOLDER
PROMPT
)" | sed "s|REPORT_FILE_PLACEHOLDER|$REPORT_FILE|"
echo ""
echo "Analysis complete. Report saved to: $REPORT_FILE"
echo ""
echo "=== Report Preview ==="
head -50 "$REPORT_FILE"
Make the script executable:
chmod +x analyze.sh
./analyze.shStep 6: Crafting Effective Prompts
The quality of your agent's output depends heavily on the prompts you use. Create dedicated prompt files that can be reused and refined.
System Prompt
<!-- prompts/system_prompt.md -->
You are a log analysis agent operating as a senior SRE. You have deep expertise
in distributed systems, microservices architecture, and incident response.
## Your Knowledge Base
- You understand common error patterns: NullPointerException, connection
timeouts, circuit breaker trips, cache failures, and authentication errors.
- You can reason about request flows across microservices using trace IDs.
- You know how to distinguish between root causes and downstream symptoms.
- You understand that the first error in a time-ordered cascade is often the
root cause, while subsequent errors are cascading failures.
## Your Tone
- Technical and precise.
- Evidence-based: every claim must reference a specific log line.
- Concise but complete: no filler, no speculation.
- Action-oriented: always end with concrete recommendations.
Analysis Prompt Template
<!-- prompts/analysis_prompt.md -->
Analyze the following log data and produce an incident report.
## Context
- Log source: {LOG_SOURCE}
- Time range: {TIME_RANGE}
- Known issues: {KNOWN_ISSUES}
## Instructions
1. Begin by running the summary script to get an overview.
2. Identify all ERROR level entries and note their timestamps.
3. Group errors by trace_id to reconstruct request flows.
4. Identify error cascades — clusters of errors within 30 seconds.
5. For each cascade, determine the root cause (the first error).
6. Check if the root cause error appears in multiple services.
7. Write the incident report to {OUTPUT_FILE}.
## Report Structure
### Incident Summary
A single paragraph (3-5 sentences) describing the incident.
### Timeline
A chronological list of the 10-15 most significant events, each with:
- Timestamp
- Service
- Event description
- Severity (INFO/WARN/ERROR)
### Root Cause Analysis
- The originating error (with exact log line reference)
- Why it occurred (based on available evidence)
- How it propagated to other services
### Impact Assessment
- Services affected
- Estimated number of impacted requests
- Duration of the incident
### Recommended Actions
- Immediate fix (what to do right now)
- Short-term improvement (within the next sprint)
- Long-term prevention (architectural or process changes)
Step 7: Running the Agent in Interactive Mode
While the shell script is great for automated analysis, you can also run Claude Code interactively for exploratory investigations. This is useful when you need to ask follow-up questions or drill down into specific errors.
# Start Claude Code in the project directory
cd log-analysis-agent
claude
# Then interact with the agent:
> Read the last 100 lines of logs/app.log and tell me what errors occurred.
> Find all log entries with trace_id "trace-350" across all log files.
> Run the parse_logs.py script with --cascades mode and explain the largest cascade.
> Compare the error patterns in auth-service vs payment-service logs.
> Write a postmortem for the incident around 9:45 AM based on the logs.
Step 8: Handling Real-World Log Formats
Real applications rarely produce clean JSON logs. You will encounter plain text logs, syslog format, Apache/Nginx access logs, and mixed formats. Extend your parsing script to handle these.
# scripts/parse_mixed_logs.py
import re
import json
import sys
from datetime import datetime
# Common log format patterns
PATTERNS = {
'json': lambda line: json.loads(line),
'syslog': re.compile(
r'(?P<timestamp>\w{3}\s+\d+\s\d{2}:\d{2}:\d{2})\s+'
r'(?P<host>\S+)\s+(?P<service>\S+):\s+(?P<message>.*)'
),
'apache': re.compile(
r'(?P<ip>\S+)\s+\S+\s+\S+\s+\[(?P<timestamp>[^\]]+)\]\s+'
r'"(?P<method>\S+)\s+(?P<path>\S+)\s+(?P<protocol>[^"]+)"\s+'
r'(?P<status>\d+)\s+(?P<size>\S+)'
),
'plain': re.compile(
r'(?P<timestamp>\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2})\s+'
r'\[(?P<level>\w+)\]\s+(?P<service>\S+)\s+-\s+(?P<message>.*)'
),
}
def detect_and_parse(line):
"""Auto-detect log format and parse accordingly."""
line = line.strip()
if not line:
return None
# Try JSON first
try:
entry = json.loads(line)
entry['_format'] = 'json'
return entry
except json.JSONDecodeError:
pass
# Try each regex pattern
for name, pattern in PATTERNS.items():
if name == 'json':
continue
match = pattern.match(line)
if match:
entry = match.groupdict()
entry['_format'] = name
# Normalize level for plain format
if 'level' not in entry and name == 'apache':
status = int(entry.get('status', 200))
entry['level'] = 'ERROR' if status >= 500 else ('WARN' if status >= 400 else 'INFO')
return entry
# Fallback: treat as unstructured
return {
'message': line,
'level': 'UNKNOWN',
'_format': 'unstructured',
}
def parse_mixed_file(filepath):
"""Parse a file with potentially mixed log formats."""
entries = []
with open(filepath, 'r') as f:
for line_num, line in enumerate(f, 1):
entry = detect_and_parse(line)
if entry:
entry['_line_number'] = line_num
entry['_source_file'] = filepath
entries.append(entry)
return entries
if __name__ == "__main__":
if len(sys.argv) < 2:
print("Usage: python parse_mixed_logs.py <logfile>")
sys.exit(1)
entries = parse_mixed_file(sys.argv[1])
# Print format distribution
from collections import Counter
formats = Counter(e.get('_format') for e in entries)
print("Format distribution:")
for fmt, count in formats.items():
print(f" {fmt}: {count}")
# Print errors
errors = [e for e in entries if e.get('level') in ('ERROR', 'WARN')]
print(f"\nErrors and warnings: {len(errors)}")
for e in errors[:20]:
print(f" [{e.get('timestamp', '?')}] {e.get('level')}: {e.get('message', '?')}")
Best Practices for Log Analysis Agents
1. Keep Logs Structured
The agent performs best with structured JSON logs. If you control the logging code, emit JSON with consistent fields: timestamp, level, service, message, trace_id, and user_id. This eliminates parsing ambiguity and lets the agent focus on analysis rather than format detection.
2. Limit Context Window Usage
Large log files can exceed Claude's context window. Instead of feeding entire log files to the agent, use scripts to pre-filter and summarize. Feed the agent the filtered results, error cascades, and specific log lines rather than raw data. A good rule of thumb is to keep the input under 50,000 tokens.
# Example: pre-filter before feeding to agent
grep '"level": "ERROR"' logs/app.log | head -100 > logs/errors_only.log
python scripts/parse_logs.py logs/app.log --cascades > logs/cascades.txt
3. Use Trace IDs Everywhere
Trace IDs are the backbone of cross-service correlation. Ensure every service in your architecture propagates trace IDs through request headers and includes them in log output. Without trace IDs, the agent has to rely on timestamp proximity, which is far less reliable.
4. Provide Context in the Prompt
The agent analyzes better when it knows the context. Include information about your architecture, recent deployments, known issues, and the time window of interest. This helps the agent distinguish between expected behavior and genuine anomalies.
5. Validate Agent Findings
Always have a human review the agent's conclusions, especially for production incidents. The agent can miss context that an experienced engineer would catch. Use the agent as a first-pass investigator that produces a draft report, not as a final authority.
6. Build a Knowledge Base
Over time, accumulate incident reports and known error signatures in a knowledge base file. Reference this file in your CLAUDE.md so the agent can recognize recurring patterns and reference past resolutions.
# CLAUDE.md addition
## Known Error Patterns
- **Connection refused to database**: Usually caused by DB failover. Check
if a deployment happened in the last 30 minutes. See incident 2024-01-10.
- **JWT token expired**: Expected for long-running sessions. Not an incident
unless rate exceeds 100/minute.
- **Circuit breaker opened**: Downstream service is unhealthy. Check the
specific service's logs for the root cause.
7. Handle Sensitive Data
Logs often contain sensitive information: API keys, passwords, PII. Before feeding logs to the agent, run a redaction script that masks sensitive fields. Never send unredacted production logs to any external service.
# scripts/redact_logs.py
import re
import json
import sys
SENSITIVE_PATTERNS = {
'api_key': re.compile(r'(api[_-]?key["\s:=]+)([^\s"]+)', re.IGNORECASE),
'password': re.compile(r'(password["\s:=]+)([^\s"]+)', re.IGNORECASE),
'email': re.compile(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b'),
'credit_card': re.compile(r'\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b'),
'ssn': re.compile(r'\b\d{3}-\d{2}-\d{4}\b'),
}
def redact_entry(entry):
"""Redact sensitive data from a log entry."""
if isinstance(entry, dict):
return {k: redact_entry(v) for k, v in entry.items()}
elif isinstance(entry, str):
result = entry
for field, pattern in SENSITIVE_PATTERNS.items():
if field in ('email',):
result = pattern.sub('[REDACTED_EMAIL]', result)
else:
result = pattern.sub(r'\1[REDACTED]', result)
return result
return entry
def redact_file(input_path, output_path):
with open(input_path, 'r') as f:
for line in f:
line = line.strip()
if not line:
continue
try:
entry = json.loads(line)
redacted = redact_entry(entry)
with open(output_path, 'a') as out:
out.write(json.dumps(redacted) + '\n')
except json.JSONDecodeError:
# Redact plain text
redacted = redact_entry(line)
with open(output_path, 'a') as out:
out.write(redacted + '\n')
if __name__ == "__main__":
if len(sys.argv) < 3:
print("Usage: python redact_logs.py <input> <output>")
sys.exit(1)
redact_file(sys.argv[1], sys.argv[2])
print(f"Redacted logs written to {sys.argv[2]}")
8. Schedule Regular Analysis
Do not wait for incidents to run the agent. Schedule periodic analysis (hourly or daily) to catch degrading trends before they become incidents. Use a cron job or CI/CD pipeline to run the agent automatically.
# Cron job: run log analysis every hour
0 * * * * cd /path/to/log-analysis-agent && ./analyze.sh >> /var/log/agent-runs.log 2>&1
Advanced: Integrating with Observability Platforms
For production use, you will want to integrate the agent with your existing observability stack. Here is a sketch of how to pull logs from common platforms.
Pulling from Elasticsearch
#!/bin/bash
# scripts/fetch_from_elasticsearch.sh
# Fetch recent error logs from Elasticsearch
ES_HOST="${ELASTICSEARCH_HOST:-http://localhost:9200}"
INDEX="app-logs-*"
SINCE="now-1h"
curl -s "$ES_HOST/$INDEX/_search" -H 'Content-Type: application/json' -d "{
\"query\": {
\"bool\": {
\"must\": [
{\"range\": {\"@timestamp\": {\"gte\": \"$SINCE\"}}},
{\"terms\": {\"level\": [\"ERROR\", \"WARN\"]}}
]
}
},
\"sort\": [{\"@timestamp\": \"asc\"}],
\"size\": 1000
}" | jq '.hits.hits[]._source' > logs/fetched_errors.jsonl
echo "Fetched $(wc -l < logs/fetched_errors.jsonl) error entries from Elasticsearch"
Pulling from CloudWatch
#!/bin/bash
# scripts/fetch_from_cloudwatch.sh
# Fetch recent error logs from AWS CloudWatch
LOG_GROUP="/aws/ecs/myapp"
SINCE=$((($(date +%s) - 3600) * 1000))
aws logs filter-log-events \
--log-group-name "$LOG_GROUP" \
--start-time "$SINCE" \
--filter-pattern "ERROR" \
--query 'events[*].message' \
--output text > logs/cloudwatch_errors.log
echo "Fetched error logs from CloudWatch"
Conclusion
Building a log analysis agent with Claude Code transforms how your team handles incidents and monitors system health. By combining Claude's language understanding with practical scripting tools, you create an agent that can survey logs, detect error cascades, correlate events across services, and produce clear incident reports — all in a fraction of the time a human would need. The key to success lies in structured logging, well-crafted prompts, careful context management, and a workflow that balances automation with human oversight. Start with the sample logs and scripts in this guide, then adapt the agent to your real log sources and architecture. As you accumulate incident reports and known error patterns, the agent becomes increasingly effective, turning your log data from a passive record into an active intelligence system that helps you resolve incidents faster and prevent them from recurring.