LLM Function Calling for DevOps Automation with Claude API

import anthropic
import subprocess
import json
 
client = anthropic.Anthropic()
 
# Define tools Claude can use
DEVOPS_TOOLS = [
    {
        "name": "run_kubectl",
        "description": "Execute a kubectl command and return the output. Use for Kubernetes operations.",
        "input_schema": {
            "type": "object",
            "properties": {
                "command": {
                    "type": "string",
                    "description": "The kubectl command to run (without the 'kubectl' prefix)"
                },
                "namespace": {
                    "type": "string",
                    "description": "Kubernetes namespace (optional, use when relevant)"
                }
            },
            "required": ["command"]
        }
    },
    {
        "name": "run_aws_cli",
        "description": "Execute an AWS CLI command and return JSON output.",
        "input_schema": {
            "type": "object",
            "properties": {
                "command": {
                    "type": "string",
                    "description": "The aws CLI command (without the 'aws' prefix)"
                }
            },
            "required": ["command"]
        }
    },
    {
        "name": "check_pod_logs",
        "description": "Get recent logs from a pod or deployment.",
        "input_schema": {
            "type": "object",
            "properties": {
                "target": {"type": "string", "description": "Pod name or deployment name"},
                "namespace": {"type": "string", "description": "Kubernetes namespace"},
                "lines": {"type": "integer", "description": "Number of lines (default 50)"},
                "filter": {"type": "string", "description": "Optional grep filter"}
            },
            "required": ["target", "namespace"]
        }
    },
    {
        "name": "scale_deployment",
        "description": "Scale a Kubernetes deployment to a specific number of replicas.",
        "input_schema": {
            "type": "object",
            "properties": {
                "deployment": {"type": "string", "description": "Deployment name"},
                "namespace": {"type": "string", "description": "Kubernetes namespace"},
                "replicas": {"type": "integer", "description": "Target replica count"}
            },
            "required": ["deployment", "namespace", "replicas"]
        }
    }
]
 
 
def execute_tool(tool_name: str, tool_input: dict) -> str:
    """Execute a tool call and return the result."""
 
    if tool_name == "run_kubectl":
        cmd = tool_input["command"].split()
        ns = tool_input.get("namespace")
        if ns and "-n" not in cmd and "--namespace" not in cmd:
            cmd = ["-n", ns] + cmd
        result = subprocess.run(
            ["kubectl"] + cmd,
            capture_output=True, text=True, timeout=30
        )
        return result.stdout or result.stderr or "No output"
 
    elif tool_name == "run_aws_cli":
        cmd = tool_input["command"].split()
        if "--output" not in cmd:
            cmd += ["--output", "json"]
        result = subprocess.run(
            ["aws"] + cmd,
            capture_output=True, text=True, timeout=60
        )
        return result.stdout or result.stderr or "No output"
 
    elif tool_name == "check_pod_logs":
        target = tool_input["target"]
        namespace = tool_input["namespace"]
        lines = tool_input.get("lines", 50)
        filter_str = tool_input.get("filter")
 
        cmd = ["kubectl", "logs", "-n", namespace, target, f"--tail={lines}"]
        result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
        logs = result.stdout or result.stderr
 
        if filter_str:
            filtered = [line for line in logs.split("\n") if filter_str.lower() in line.lower()]
            return "\n".join(filtered) or f"No lines matching '{filter_str}'"
        return logs
 
    elif tool_name == "scale_deployment":
        result = subprocess.run(
            ["kubectl", "scale", "deployment", tool_input["deployment"],
             "-n", tool_input["namespace"],
             f"--replicas={tool_input['replicas']}"],
            capture_output=True, text=True, timeout=30
        )
        return result.stdout or result.stderr
 
    return f"Unknown tool: {tool_name}"
 
 
def run_devops_agent(user_request: str, max_iterations: int = 10) -> str:
    """Run the DevOps agent with tool use loop."""
 
    messages = [{"role": "user", "content": user_request}]
 
    system = """You are an expert DevOps engineer with kubectl and AWS CLI access.
Diagnose and fix issues systematically:
1. Gather information first (don't make assumptions)
2. Check logs and events for errors
3. Apply fixes methodically
4. Verify the fix worked
Always explain what you're doing and why."""
 
    for iteration in range(max_iterations):
        response = client.messages.create(
            model="claude-sonnet-5",
            max_tokens=2000,
            system=system,
            tools=DEVOPS_TOOLS,
            messages=messages
        )
 
        messages.append({"role": "assistant", "content": response.content})
 
        # Check if Claude is done
        if response.stop_reason == "end_turn":
            # Extract final text response
            for block in response.content:
                if hasattr(block, "text"):
                    return block.text
            return "Task completed"
 
        # Execute all tool calls
        tool_results = []
        for block in response.content:
            if block.type == "tool_use":
                print(f"  → Calling {block.name}: {json.dumps(block.input)[:100]}...")
                result = execute_tool(block.name, block.input)
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": result[:3000]  # Limit result size
                })
 
        if tool_results:
            messages.append({"role": "user", "content": tool_results})
 
    return "Max iterations reached"
 
 
if __name__ == "__main__":
    # Example requests
    requests = [
        "The payment-service in production is showing high error rates. Diagnose the issue.",
        "Scale the api-gateway deployment in production to 5 replicas and verify it's healthy.",
        "Check if any pods in the production namespace are in CrashLoopBackOff and fix them.",
    ]
 
    for req in requests[:1]:  # Run first example
        print(f"\nRequest: {req}")
        print("-" * 60)
        result = run_devops_agent(req)
        print(f"\nResult:\n{result}")

DANGEROUS_OPERATIONS = ["delete", "drain", "scale --replicas=0", "rollout undo"]
 
def execute_tool_with_approval(tool_name: str, tool_input: dict) -> str:
    command = tool_input.get("command", "")
    
    if any(op in command for op in DANGEROUS_OPERATIONS):
        confirm = input(f"Confirm dangerous operation? '{command}' [y/N]: ")
        if confirm.lower() != "y":
            return "Operation cancelled by user"
    
    return execute_tool(tool_name, tool_input)