Local Skills with Claude API

Build a production log analyzer skill and integrate it with Claude for interactive diagnostics.

Time: 30 minutes

Prerequisites: Complete Hello World first.

What You'll Build

A log analysis skill that:

• Fetches logs from a REST API
• Parses and filters log entries
• Analyzes patterns and errors
• Provides diagnostic recommendations

Understanding Skill Structure

Before building, let's understand how skills are organized:

log-analyzer/
├── SKILL.md              # Skill definition (required)
├── scripts/              # Executable scripts
│   ├── fetch_logs.py     # Fetch logs from API
│   ├── parse_logs.py     # Parse log formats
│   └── analyze.py        # Analyze patterns
└── references/           # Documentation for Claude
    └── log_formats.md    # Supported formats

SKILL.md Components

The SKILL.md file has two parts:

1. YAML Frontmatter - Metadata for skill discovery:

---
name: log-analyzer
description: |
  Production log analysis skill. Use when asked to:
  - Fetch and analyze logs
  - Diagnose production issues
license: MIT
compatibility: python3, requests
allowed-tools: Bash(python3:*) Read Write
---

2. Markdown Body - Instructions for Claude:

# Log Analyzer Skill

## Available Scripts

### fetch_logs.py
Fetches logs from REST API...

### analyze.py
Analyzes log patterns...

Step 1: Create the Skill Directory

mkdir -p priv/skills/log-analyzer/{scripts,references}

Step 2: Create SKILL.md

Create priv/skills/log-analyzer/SKILL.md:

---
name: log-analyzer
description: |
  Production log analysis and diagnostics skill. Use this skill when asked to:
  - Fetch logs from a monitoring API
  - Analyze log patterns and errors
  - Diagnose production issues
  - Generate log summaries
license: MIT
compatibility: python3, requests
allowed-tools: Bash(python3:*) Read Write
---

# Log Analyzer Skill

A production monitoring skill for analyzing application logs.

## Available Scripts

### 1. Fetch Logs

\`\`\`bash
python3 scripts/fetch_logs.py --endpoint "http://api.example.com/logs" --limit 100
\`\`\`

Options:
- \`--endpoint\` - REST API URL (required)
- \`--limit\` - Max logs to fetch
- \`--level\` - Filter by level (DEBUG, INFO, WARN, ERROR)
- \`--output\` - Save to file

### 2. Analyze Logs

\`\`\`bash
python3 scripts/analyze.py logs.json --summary
python3 scripts/analyze.py logs.json --errors-only
python3 scripts/analyze.py logs.json --diagnostics
\`\`\`

## Workflow

1. Fetch recent logs
2. Analyze for patterns
3. Generate diagnostics

Step 3: Create the Fetch Script

Create priv/skills/log-analyzer/scripts/fetch_logs.py:

#!/usr/bin/env python3
"""Fetch logs from REST API."""

import argparse
import json
import sys
from datetime import datetime


def simulate_logs(limit: int, level: str = None) -> list:
    """Simulate log API response for demo."""
    samples = [
        ("INFO", "Request received: GET /api/users"),
        ("WARN", "Slow query: 2.5s for user lookup"),
        ("ERROR", "Connection timeout to payment service"),
        ("INFO", "Request completed: 200 OK"),
        ("ERROR", "Database connection failed"),
    ]

    logs = []
    for i in range(min(limit, 50)):
        log_level, message = samples[i % len(samples)]
        if level and log_level != level:
            continue

        logs.append({
            "timestamp": datetime.now().isoformat(),
            "level": log_level,
            "service": "api-gateway",
            "message": message,
            "host": f"prod-server-{(i % 3) + 1}",
        })

    return logs


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--endpoint", required=True)
    parser.add_argument("--limit", type=int, default=100)
    parser.add_argument("--level", choices=["DEBUG", "INFO", "WARN", "ERROR"])
    parser.add_argument("--output")
    args = parser.parse_args()

    logs = simulate_logs(args.limit, args.level)

    output = json.dumps(logs, indent=2)
    if args.output:
        with open(args.output, "w") as f:
            f.write(output)
        print(f"Saved {len(logs)} logs to {args.output}")
    else:
        print(output)


if __name__ == "__main__":
    main()

Make it executable:

chmod +x priv/skills/log-analyzer/scripts/fetch_logs.py

Step 4: Create the Analyze Script

Create priv/skills/log-analyzer/scripts/analyze.py:

#!/usr/bin/env python3
"""Analyze logs for patterns and errors."""

import argparse
import json
from collections import Counter


def load_logs(filepath: str) -> list:
    with open(filepath) as f:
        return json.loads(f.read())


def analyze_summary(logs: list) -> dict:
    levels = Counter(log.get("level", "INFO") for log in logs)
    error_rate = levels.get("ERROR", 0) / len(logs) * 100 if logs else 0

    return {
        "total_logs": len(logs),
        "level_breakdown": dict(levels),
        "error_rate": f"{error_rate:.1f}%",
        "health": "CRITICAL" if error_rate > 10 else "HEALTHY",
    }


def analyze_errors(logs: list) -> dict:
    errors = [l for l in logs if l.get("level") == "ERROR"]
    messages = Counter(e.get("message", "")[:50] for e in errors)

    return {
        "total_errors": len(errors),
        "top_errors": messages.most_common(5),
    }


def generate_diagnostics(logs: list) -> list:
    errors = " ".join(l.get("message", "") for l in logs if l.get("level") == "ERROR")
    diagnostics = []

    if "timeout" in errors.lower():
        diagnostics.append({"issue": "Timeout errors", "priority": "HIGH"})
    if "connection" in errors.lower():
        diagnostics.append({"issue": "Connection failures", "priority": "HIGH"})

    return diagnostics or [{"issue": "No critical issues", "priority": "LOW"}]


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("file")
    parser.add_argument("--summary", action="store_true")
    parser.add_argument("--errors-only", action="store_true")
    parser.add_argument("--diagnostics", action="store_true")
    args = parser.parse_args()

    logs = load_logs(args.file)

    if args.summary:
        print(json.dumps(analyze_summary(logs), indent=2))
    elif args.errors_only:
        print(json.dumps(analyze_errors(logs), indent=2))
    elif args.diagnostics:
        print(json.dumps(generate_diagnostics(logs), indent=2))
    else:
        print(json.dumps({
            "summary": analyze_summary(logs),
            "errors": analyze_errors(logs),
            "diagnostics": generate_diagnostics(logs),
        }, indent=2))


if __name__ == "__main__":
    main()

Make it executable:

chmod +x priv/skills/log-analyzer/scripts/analyze.py

Step 5: Test the Skill Manually

# Fetch logs
python3 priv/skills/log-analyzer/scripts/fetch_logs.py \
  --endpoint "http://localhost/api/logs" \
  --limit 20 \
  --output /tmp/logs.json

# Analyze
python3 priv/skills/log-analyzer/scripts/analyze.py /tmp/logs.json --summary

Step 6: Create the Agent

Create lib/my_app/log_agent.ex:

defmodule MyApp.LogAgent do
  @moduledoc """
  An agent that uses the log-analyzer skill for production diagnostics.
  """

  @api_url "https://api.anthropic.com/v1/messages"

  def diagnose(question) do
    {:ok, skills} = Conjure.load("priv/skills")
    session = Conjure.Session.new_local(skills)

    case Conjure.Session.chat(session, question, &api_callback/1) do
      {:ok, response, _session} ->
        {:ok, extract_text(response)}

      {:error, error} ->
        {:error, error}
    end
  end

  defp api_callback(messages) do
    {:ok, skills} = Conjure.load("priv/skills")

    body = %{
      model: "claude-sonnet-4-5-20250929",
      max_tokens: 4096,
      system: system_prompt(skills),
      messages: messages,
      tools: Conjure.tool_definitions()
    }

    case Req.post(@api_url, json: body, headers: headers()) do
      {:ok, %{status: 200, body: body}} -> {:ok, body}
      {:ok, %{body: body}} -> {:error, body}
      {:error, reason} -> {:error, reason}
    end
  end

  defp system_prompt(skills) do
    """
    You are a production support engineer with access to log analysis tools.
    When diagnosing issues:
    1. First fetch recent logs
    2. Analyze for patterns
    3. Provide clear diagnostics

    #{Conjure.system_prompt(skills)}
    """
  end

  defp headers do
    [
      {"x-api-key", api_key()},
      {"anthropic-version", "2023-06-01"},
      {"content-type", "application/json"}
    ]
  end

  defp extract_text(%{"content" => content}) do
    content
    |> Enum.filter(&(&1["type"] == "text"))
    |> Enum.map_join("\n", & &1["text"])
  end

  defp api_key do
    System.get_env("ANTHROPIC_API_KEY") ||
      raise "ANTHROPIC_API_KEY not set"
  end
end

Step 7: Run a Diagnostic Session

# In IEx
MyApp.LogAgent.diagnose("""
We're seeing slow response times in production.
Please fetch the last 50 logs from http://monitoring.example.com/api/logs
and analyze them for issues.
""")

Expected interaction:

[Claude reads SKILL.md]
[Claude calls: fetch_logs.py --endpoint "..." --limit 50 --output /tmp/logs.json]
[Claude calls: analyze.py /tmp/logs.json --summary]
[Claude calls: analyze.py /tmp/logs.json --diagnostics]

Based on my analysis of the last 50 logs:

**Summary:**
- Total logs: 50
- Error rate: 20%
- Health status: CRITICAL

**Issues Found:**
1. Connection timeout errors to payment service (HIGH priority)
2. Database connection failures (HIGH priority)

**Recommendations:**
1. Check payment service connectivity
2. Review database connection pool settings
3. Consider increasing connection timeouts

Multi-Turn Conversations

The Session API maintains conversation state:

{:ok, skills} = Conjure.load("priv/skills")
session = Conjure.Session.new_local(skills)

# First turn
{:ok, response1, session} = Conjure.Session.chat(
  session,
  "Fetch the last 100 logs from http://api.example.com/logs",
  &api_callback/1
)

# Second turn (continues the conversation)
{:ok, response2, session} = Conjure.Session.chat(
  session,
  "Now show me only the errors",
  &api_callback/1
)

# Third turn
{:ok, response3, _session} = Conjure.Session.chat(
  session,
  "What's causing these errors?",
  &api_callback/1
)

Adding Callbacks for Visibility

Monitor tool calls in real-time:

opts = [
  on_tool_call: fn call ->
    IO.puts("Tool: #{call.name}")
    IO.puts("Input: #{inspect(call.input)}")
  end,
  on_tool_result: fn result ->
    IO.puts("Result: #{String.slice(result.content, 0, 100)}...")
  end
]

Conjure.Conversation.run_loop(messages, skills, &api_callback/1, opts)

Troubleshooting

"Skill not found"

Verify the skill structure:

ls -la priv/skills/log-analyzer/
# Should show: SKILL.md, scripts/, references/

Script execution fails

Test scripts manually:

python3 priv/skills/log-analyzer/scripts/fetch_logs.py --help

"No module named requests"

For production, install dependencies:

pip install requests

The demo scripts simulate API responses, so requests isn't required for learning.

Next Steps

• Anthropic Skills API - Generate incident reports as spreadsheets
• Native Elixir Skills - Fetch logs directly from Elixir
• Unified Backends - Combine all approaches