Building LLM Agents with Tool Use for DevOps Automation Using Claude API

import anthropic
import json
import subprocess
 
client = anthropic.Anthropic()
 
 
# Define the tools Claude can use
DEVOPS_TOOLS = [
    {
        "name": "kubectl_get",
        "description": "Run kubectl get to list Kubernetes resources. Returns the output.",
        "input_schema": {
            "type": "object",
            "properties": {
                "resource_type": {
                    "type": "string",
                    "description": "Resource type: pods, deployments, services, nodes, etc."
                },
                "namespace": {
                    "type": "string",
                    "description": "Kubernetes namespace. Use 'all-namespaces' to query all."
                },
                "flags": {
                    "type": "string",
                    "description": "Additional kubectl flags like '-o wide' or '-l app=myapp'"
                }
            },
            "required": ["resource_type"]
        }
    },
    {
        "name": "kubectl_describe",
        "description": "Describe a specific Kubernetes resource to get detailed information and events.",
        "input_schema": {
            "type": "object",
            "properties": {
                "resource_type": {"type": "string"},
                "name": {"type": "string"},
                "namespace": {"type": "string", "default": "default"}
            },
            "required": ["resource_type", "name"]
        }
    },
    {
        "name": "kubectl_logs",
        "description": "Get logs from a Kubernetes pod or deployment.",
        "input_schema": {
            "type": "object",
            "properties": {
                "pod_name": {"type": "string"},
                "namespace": {"type": "string", "default": "default"},
                "container": {"type": "string"},
                "lines": {"type": "integer", "default": 50},
                "previous": {"type": "boolean", "default": False}
            },
            "required": ["pod_name"]
        }
    },
    {
        "name": "aws_cli",
        "description": "Run an AWS CLI command. Use for querying AWS resources like EC2, RDS, costs.",
        "input_schema": {
            "type": "object",
            "properties": {
                "command": {
                    "type": "string",
                    "description": "Full AWS CLI command starting with 'aws', e.g. 'aws ec2 describe-instances --query ...'"
                }
            },
            "required": ["command"]
        }
    },
    {
        "name": "run_shell",
        "description": "Run a safe, read-only shell command for diagnostics.",
        "input_schema": {
            "type": "object",
            "properties": {
                "command": {
                    "type": "string",
                    "description": "Shell command to run. Only read-only commands allowed."
                }
            },
            "required": ["command"]
        }
    }
]

SAFE_SHELL_COMMANDS = {"curl", "dig", "nslookup", "ping", "nc", "telnet", "cat", "ls", "df", "ps", "top"}
BLOCKED_KUBECTL_VERBS = {"delete", "apply", "patch", "exec", "port-forward"}
 
 
def execute_tool(tool_name: str, tool_input: dict) -> str:
    """Execute a tool call and return the result as a string."""
    try:
        if tool_name == "kubectl_get":
            return _kubectl_get(**tool_input)
        elif tool_name == "kubectl_describe":
            return _kubectl_describe(**tool_input)
        elif tool_name == "kubectl_logs":
            return _kubectl_logs(**tool_input)
        elif tool_name == "aws_cli":
            return _aws_cli(**tool_input)
        elif tool_name == "run_shell":
            return _run_shell(**tool_input)
        else:
            return f"Unknown tool: {tool_name}"
    except Exception as e:
        return f"Tool error: {str(e)}"
 
 
def _kubectl_get(resource_type: str, namespace: str = "default", flags: str = "") -> str:
    cmd = ["kubectl", "get", resource_type]
    if namespace == "all-namespaces":
        cmd.append("--all-namespaces")
    else:
        cmd.extend(["-n", namespace])
    if flags:
        cmd.extend(flags.split())
    cmd.append("--no-headers")
 
    result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
    output = result.stdout or result.stderr
    return output[:3000] if output else "No output"
 
 
def _kubectl_describe(resource_type: str, name: str, namespace: str = "default") -> str:
    cmd = ["kubectl", "describe", resource_type, name, "-n", namespace]
    result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
    output = result.stdout or result.stderr
    return output[:4000] if output else "No output"
 
 
def _kubectl_logs(
    pod_name: str,
    namespace: str = "default",
    container: str = None,
    lines: int = 50,
    previous: bool = False
) -> str:
    cmd = ["kubectl", "logs", pod_name, "-n", namespace, f"--tail={lines}"]
    if container:
        cmd.extend(["-c", container])
    if previous:
        cmd.append("--previous")
 
    result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
    output = result.stdout or result.stderr
    return output[-3000:] if output else "No logs found"
 
 
def _aws_cli(command: str) -> str:
    # Safety: only allow read operations
    blocked = ["delete", "terminate", "remove", "put", "create", "update", "modify"]
    cmd_lower = command.lower()
    for blocked_word in blocked:
        if f" {blocked_word}" in cmd_lower:
            return f"Blocked: '{blocked_word}' operations not allowed in read-only mode"
 
    parts = command.split()
    result = subprocess.run(parts, capture_output=True, text=True, timeout=60)
    output = result.stdout or result.stderr
    return output[:3000] if output else "No output"
 
 
def _run_shell(command: str) -> str:
    first_word = command.split()[0]
    if first_word not in SAFE_SHELL_COMMANDS:
        return f"Blocked: '{first_word}' not in allowed commands list"
 
    result = subprocess.run(
        command, shell=True, capture_output=True, text=True, timeout=30
    )
    output = result.stdout or result.stderr
    return output[:2000] if output else "No output"

def run_devops_agent(user_request: str, max_iterations: int = 10) -> str:
    """
    Run an agentic loop that uses tools to answer DevOps questions.
    """
    print(f"Agent starting: {user_request}")
 
    messages = [{"role": "user", "content": user_request}]
 
    system_prompt = """You are an expert DevOps engineer assistant with access to tools
to query Kubernetes clusters, AWS resources, and run diagnostics.
 
When investigating issues:
1. Start by gathering information — do not assume, verify with tools
2. Use specific queries rather than broad ones to avoid overwhelming output
3. Follow the evidence — if logs show an error, dig into that error
4. Provide a clear diagnosis and specific remediation steps at the end
 
You have read-only access — you cannot modify resources. Focus on diagnosis."""
 
    for iteration in range(max_iterations):
        response = client.messages.create(
            model="claude-sonnet-5",
            max_tokens=4096,
            system=system_prompt,
            tools=DEVOPS_TOOLS,
            messages=messages
        )
 
        # Add assistant response to history
        messages.append({"role": "assistant", "content": response.content})
 
        # Check stop reason
        if response.stop_reason == "end_turn":
            # Extract the final text response
            for block in response.content:
                if hasattr(block, "text"):
                    return block.text
            return "Agent completed without text response"
 
        if response.stop_reason != "tool_use":
            break
 
        # Process tool calls
        tool_results = []
        for block in response.content:
            if block.type == "tool_use":
                print(f"  → Calling tool: {block.name}({json.dumps(block.input)[:100]})")
                result = execute_tool(block.name, block.input)
                print(f"  ← Result: {result[:100]}...")
 
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": result
                })
 
        # Add tool results to conversation
        messages.append({"role": "user", "content": tool_results})
 
    return "Agent reached iteration limit without completing"

if __name__ == "__main__":
    # Example 1: Investigate a failing deployment
    result = run_devops_agent(
        "My payment-api deployment in the production namespace has been showing "
        "increased error rates for the last 30 minutes. Can you investigate what's "
        "happening and tell me the root cause?"
    )
    print(result)
 
    # Example 2: Cost investigation
    result = run_devops_agent(
        "Our AWS bill spiked by $2000 this week. Can you check what EC2 instances "
        "are running and if there are any unusual resources that might explain the increase?"
    )
    print(result)
 
    # Example 3: Capacity planning
    result = run_devops_agent(
        "How many pods are running in each namespace in our cluster? "
        "Which namespaces are using the most resources?"
    )
    print(result)