Skills Monitoring and Observability
After a Skill is published, how do you know if it's working well? Is it failing frequently? Which steps are the slowest?
This article introduces how to build observability for Skills, using logs and metrics to understand runtime status.
Three Dimensions of Observability
| Dimension | Question Answered | Implementation |
|---|---|---|
| Logs | What happened? What went wrong? | Structured JSON log files |
| Metrics | How many times was it executed? How long did it take? What is the success rate? | Append to metrics file |
| Trace | Which steps did each execution go through? How long did each step take? | Step-level timestamp records |
Structured Log Output Standards
Skill scripts should write runtime logs to a file at a fixed path for easy later review.
Example
# File path: scripts/skill_logger.py
import json
import os
import sys
from datetime import datetime, timezone
LOG_FILE = "/home/claude/skill_run.log"
def log(level: str, event: str, skill: str = "", **context):
"""
Write structured log entries
Parameters:
level: Log level (info / warning / error)
event: Event name (snake_case)
skill: Skill name
context: Any additional fields
"""
entry = {
"ts": datetime.now(timezone.utc).isoformat(),
"level": level,
"event": event,
"skill": skill,
**context
}
line = json.dumps(entry, ensure_ascii=False)
# Output to both stderr (real-time visibility) and log file (persistence)
print(line, file=sys.stderr)
with open(LOG_FILE, "a", encoding="utf-8") as f:
f.write(line + "\n")
# Usage example
if __name__ == "__main__":
log("info", "skill_start", skill="data-analyzer",
file="/mnt/user-data/uploads/example.csv")
log("info", "step_complete", skill="data-analyzer",
step="clean", rows_removed=3, elapsed_ms=240)
log("info", "skill_complete", skill="data-analyzer",
output="/mnt/user-data/outputs/report.xlsx", elapsed_ms=1830)
log("error", "step_failed", skill="data-analyzer",
step="generate_chart", error="openpyxl not found")
import json
import os
import sys
from datetime import datetime, timezone
LOG_FILE = "/home/claude/skill_run.log"
def log(level: str, event: str, skill: str = "", **context):
"""
Write structured log entries
Parameters:
level: Log level (info / warning / error)
event: Event name (snake_case)
skill: Skill name
context: Any additional fields
"""
entry = {
"ts": datetime.now(timezone.utc).isoformat(),
"level": level,
"event": event,
"skill": skill,
**context
}
line = json.dumps(entry, ensure_ascii=False)
# Output to both stderr (real-time visibility) and log file (persistence)
print(line, file=sys.stderr)
with open(LOG_FILE, "a", encoding="utf-8") as f:
f.write(line + "\n")
# Usage example
if __name__ == "__main__":
log("info", "skill_start", skill="data-analyzer",
file="/mnt/user-data/uploads/example.csv")
log("info", "step_complete", skill="data-analyzer",
step="clean", rows_removed=3, elapsed_ms=240)
log("info", "skill_complete", skill="data-analyzer",
output="/mnt/user-data/outputs/report.xlsx", elapsed_ms=1830)
log("error", "step_failed", skill="data-analyzer",
step="generate_chart", error="openpyxl not found")
{"ts": "2026-05-18T10:23:05+00:00", "level": "info", "event": "skill_start", "skill": "data-analyzer", "file": "/mnt/user-data/uploads/example.csv"}
{"ts": "2026-05-18T10:23:05+00:00", "level": "info", "event": "step_complete", "skill": "data-analyzer", "step": "clean", "rows_removed": 3, "elapsed_ms": 240}
{"ts": "2026-05-18T10:23:07+00:00", "level": "info", "event": "skill_complete", "skill": "data-analyzer", "output": "/mnt/user-data/outputs/report.xlsx", "elapsed_ms": 1830}
{"ts": "2026-05-18T10:23:07+00:00", "level": "error", "event": "step_failed", "skill": "data-analyzer", "step": "generate_chart", "error": "openpyxl not found"}
Metric Aggregation: Counting Executions and Success Rate
Extract key metrics from log files to understand the overall health of the Skill.
Example
# File path: scripts/metrics_report.py
# Calculate Skill runtime metrics from log files
import json
import os
from collections import defaultdict
LOG_FILE = "/home/claude/skill_run.log"
def generate_metrics_report(log_file: str) -> dict:
"""Read log files and generate a metrics summary"""
if not os.path.exists(log_file):
return {"error": f"Log file does not exist: {log_file}"}
counts = defaultdict(int) # Count of various events
errors = [] # Error records
elapsed = [] # Execution time list (ms)
with open(log_file, encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line:
continue
try:
entry = json.loads(line)
except json.JSONDecodeError:
continue
event = entry.get("event", "")
counts[event] += 1
if entry.get("level") == "error":
errors.append({
"ts": entry.get("ts"),
"event": event,
"error": entry.get("error", ""),
"step": entry.get("step", "")
})
if event == "skill_complete" and "elapsed_ms" in entry:
elapsed.append(entry["elapsed_ms"])
total_runs = counts.get("skill_start", 0)
total_errors = len(errors)
success_rate = round((total_runs - total_errors) / total_runs * 100, 1) \
if total_runs > 0 else 0
return {
"total_runs": total_runs,
"total_errors": total_errors,
"success_rate": f"{success_rate}%",
"avg_elapsed_ms": round(sum(elapsed) / len(elapsed)) if elapsed else 0,
"recent_errors": errors[-5:] # Last 5 errors
}
if __name__ == "__main__":
report = generate_metrics_report(LOG_FILE)
print(json.dumps(report, ensure_ascii=False, indent=2))
# Calculate Skill runtime metrics from log files
import json
import os
from collections import defaultdict
LOG_FILE = "/home/claude/skill_run.log"
def generate_metrics_report(log_file: str) -> dict:
"""Read log files and generate a metrics summary"""
if not os.path.exists(log_file):
return {"error": f"Log file does not exist: {log_file}"}
counts = defaultdict(int) # Count of various events
errors = [] # Error records
elapsed = [] # Execution time list (ms)
with open(log_file, encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line:
continue
try:
entry = json.loads(line)
except json.JSONDecodeError:
continue
event = entry.get("event", "")
counts[event] += 1
if entry.get("level") == "error":
errors.append({
"ts": entry.get("ts"),
"event": event,
"error": entry.get("error", ""),
"step": entry.get("step", "")
})
if event == "skill_complete" and "elapsed_ms" in entry:
elapsed.append(entry["elapsed_ms"])
total_runs = counts.get("skill_start", 0)
total_errors = len(errors)
success_rate = round((total_runs - total_errors) / total_runs * 100, 1) \
if total_runs > 0 else 0
return {
"total_runs": total_runs,
"total_errors": total_errors,
"success_rate": f"{success_rate}%",
"avg_elapsed_ms": round(sum(elapsed) / len(elapsed)) if elapsed else 0,
"recent_errors": errors[-5:] # Last 5 errors
}
if __name__ == "__main__":
report = generate_metrics_report(LOG_FILE)
print(json.dumps(report, ensure_ascii=False, indent=2))
{
"total_runs": 42,
"total_errors": 3,
"success_rate": "92.9%",
"avg_elapsed_ms": 1650,
"recent_errors": [
{"ts": "2026-05-18T09:12:00+00:00", "event": "step_failed", "step": "generate_chart", "error": "openpyxl not found"}
]
}
Execution Tracing: Recording Time for Each Step
When a particular execution is especially slow, step-level timing records can quickly pinpoint the bottleneck.
Example
# File path: scripts/tracer.py
import time
import json
import sys
from skill_logger import log
class ExecutionTracer:
"""Record the time taken by each step in a Skill execution"""
def __init__(self, skill_name: str):
self.skill_name = skill_name
self.steps = []
self.start_time = time.perf_counter()
def step(self, step_name: str):
"""Mark the start of a step, automatically recording the previous step's duration"""
now = time.perf_counter()
if self.steps:
# Calculate the previous step's duration
self.steps[-1]["elapsed_ms"] = int((now - self.steps[-1]["_start"]) * 1000)
del self.steps[-1]["_start"]
self.steps.append({"name": step_name, "_start": now})
log("info", "step_start", skill=self.skill_name, step=step_name)
def finish(self):
"""Mark the end of execution and output a complete trace report"""
now = time.perf_counter()
if self.steps:
self.steps[-1]["elapsed_ms"] = int((now - self.steps[-1]["_start"]) * 1000)
del self.steps[-1]["_start"]
total_ms = int((now - self.start_time) * 1000)
report = {"skill": self.skill_name,
"total_ms": total_ms, "steps": self.steps}
log("info", "trace_complete", skill=self.skill_name,
total_ms=total_ms, steps=self.steps)
return report
# Usage example
if __name__ == "__main__":
tracer = ExecutionTracer("data-analyzer")
tracer.step("Read file")
time.sleep(0.3) # Simulate a time-consuming operation
tracer.step("Data cleaning")
time.sleep(0.1)
tracer.step("Generate report")
time.sleep(0.8)
report = tracer.finish()
print(json.dumps(report, ensure_ascii=False, indent=2))
import time
import json
import sys
from skill_logger import log
class ExecutionTracer:
"""Record the time taken by each step in a Skill execution"""
def __init__(self, skill_name: str):
self.skill_name = skill_name
self.steps = []
self.start_time = time.perf_counter()
def step(self, step_name: str):
"""Mark the start of a step, automatically recording the previous step's duration"""
now = time.perf_counter()
if self.steps:
# Calculate the previous step's duration
self.steps[-1]["elapsed_ms"] = int((now - self.steps[-1]["_start"]) * 1000)
del self.steps[-1]["_start"]
self.steps.append({"name": step_name, "_start": now})
log("info", "step_start", skill=self.skill_name, step=step_name)
def finish(self):
"""Mark the end of execution and output a complete trace report"""
now = time.perf_counter()
if self.steps:
self.steps[-1]["elapsed_ms"] = int((now - self.steps[-1]["_start"]) * 1000)
del self.steps[-1]["_start"]
total_ms = int((now - self.start_time) * 1000)
report = {"skill": self.skill_name,
"total_ms": total_ms, "steps": self.steps}
log("info", "trace_complete", skill=self.skill_name,
total_ms=total_ms, steps=self.steps)
return report
# Usage example
if __name__ == "__main__":
tracer = ExecutionTracer("data-analyzer")
tracer.step("Read file")
time.sleep(0.3) # Simulate a time-consuming operation
tracer.step("Data cleaning")
time.sleep(0.1)
tracer.step("Generate report")
time.sleep(0.8)
report = tracer.finish()
print(json.dumps(report, ensure_ascii=False, indent=2))
{
"skill": "data-analyzer",
"total_ms": 1203,
"steps": [
{"name": "读取文件", "elapsed_ms": 312},
{"name": "数据清洗", "elapsed_ms": 108},
{"name": "生成报告", "elapsed_ms": 783}
]
}
Other ExtensionsObservability isn't something only large-scale systems need. Even for a Skill used personally, a clear log file can often reduce troubleshooting time from hours to minutes when problems arise.