I Replaced 47 DevOps Scripts With One AI Agent — Here’s What Happened

aws ecs create-cluster --cluster-name prod
aws ecs register-task-definition --family myapp ...
# [50+ commands, 3 hours, manual debugging]

agent.run("Deploy FastAPI with PostgreSQL, auto-scale to 20, "
          "minimal IAM perms, CloudWatch monitoring")

# ✅ Done in minutes for most cases, with significantly reduced debugging

"Deploy my app on 5 ECS tasks, auto-scale to 20 on high CPU"

dispatcher = {
    "ecs__create_cluster": ecs.create_cluster,
    "ecs__register_task": ecs.register_task_definition,
    "ecs__create_service": ecs.create_service,
    "iam__create_role": iam.create_role,  # For permissions
}

"User wants ECS deployment.
 I need to:
 1. Check if cluster exists
 2. Register task definition
 3. Create service with 5 tasks
 4. Setup auto-scaling
 5. Verify it's running"

# All tools in one place
tools = {
    "docker__run": run_container,
    "ec2__create": create_instance,
    "ecs__deploy": create_service,
    "iam__create_role": create_role,
    # ... more tools
}

# Let Claude orchestrate
agent.run("Deploy with Docker locally, then scale to ECS production")
# Claude figures out: Docker first, then ECS, then IAM for permissions

┌──────────────────────────────┐
│ User Goal (natural language) │  
│ "Deploy scalable production  │
│  stack with auto-scaling"    │
└──────────┬───────────────────┘
           │
           ▼
    ┌─────────────────────────────────────────────┐
    │  Claude (Bedrock)                           │
    │  ← Reads goal + available tools             │
    │  ← Decides sequence of actions              │
    │  ← Adapts when things fail                  │
    └──────────────┬────────────────────────────┘
                   │
    ┌──────────────┴──────────────┐
    │                             │
    ▼                             ▼
┌─────────────┐  ┌──────────────────┐
│ AWS APIs    │  │ Conversation     │
│ (via boto3) │  │ Memory (DynamoDB)│
│ ← Executes  │  │ ← Recalls setup  │
│   decisions │  │   from last week │
└─────────────┘  └──────────────────┘
    │                    │
    └────────┬───────────┘
             │
    ┌────────▼─────────┐
    │ Actual Resources │
    │ EC2, ECS, Docker │
    │      IAM, etc    │
    └──────────────────┘

import boto3
from agents.memory import load_history, save_message

MODEL = "apac.amazon.nova-lite-v1:0"  # Fast, cheap 

class BaseAgent:
    """Foundation for all agents (Docker, EC2, ECS, IAM)"""
    AGENT_KEY = None  # "docker", "ec2", etc.
    SYSTEM_PROMPT = ""  # Agent's personality
    CAPABILITIES = []  # What this agent can do

    def __init__(self, name: str, region: str = "us-east-1"):
        self.name = name
        self.session = boto3.Session(region_name=region)
        self._bedrock = None

    @property
    def bedrock(self):
        """Lazy load Bedrock connection (only when needed)"""
        if not self._bedrock:
            self._bedrock = self.session.client("bedrock-runtime")
        return self._bedrock

    def run(self, task: str, session_id: str = None) -> dict:
        """The agentic loop: think → decide → act → repeat"""
        tools = self.get_tools()
        dispatcher = self.get_dispatcher()
        messages = load_history(session_id) + [{"role": "user", "content": [{"text": task}]}]

        for _ in range(10):  # Max 10 iterations to prevent infinite loops
            # Ask Claude what to do
            response = self.bedrock.converse(
                modelId=MODEL,
                system=self.SYSTEM_PROMPT,
                messages=messages,
                tools=tools,
                toolUseDepth=2,
            )

            # Is Claude done?
            if response["stopReason"] == "endTurn":
                return {"status": "SUCCESS", "message": response["content"]}

            # Execute tools Claude wants to call
            tool_results = []
            for tool_use in response["content"]:
                tool_name = tool_use["toolUse"]["name"]
                try:
                    result = dispatcher<a href="tool_use["toolUse"]["input"]">tool_name</a>
                except Exception as e:
                    result = {"error": str(e), "status": "failed"}

                tool_results.append({"toolUseId": tool_use["toolUse"]["toolUseId"],
                                    "content": [{"json": result}]})

            # Add Claude's decision + results to conversation
            messages.append({"role": "assistant", "content": response["content"]})
            messages.append({"role": "user", "content": [{"toolResult": tr} for tr in tool_results]})

            # Save for future reference
            if session_id:
                save_message(session_id, messages[-1])

        return {"status": "FAILED", "reason": "Max iterations reached"}

class DockerAgent(BaseAgent):
    AGENT_KEY = "docker"
    SYSTEM_PROMPT = """You manage Docker containers.
Rules: Pull image first, check if container exists, use sensible defaults."""

    CAPABILITIES = [
        {"name": "list_containers", "description": "List running containers"},
        {"name": "run_container", "description": "Pull and run image"},
        {"name": "stop_container", "description": "Stop a container"},
    ]

docker_agent = DockerAgent()
docker_agent.run("Deploy FastAPI on port 8000 with health check")
# Claude calls: list_containers → run_container → confirms it's running

class EC2Agent(BaseAgent):
    AGENT_KEY = "ec2"
    SYSTEM_PROMPT = """You manage EC2 instances.
Rules: Tag for organization, verify security groups, create new→test→retire old."""

    CAPABILITIES = [
        {"name": "describe_instances", "description": "List instances"},
        {"name": "create_instance", "description": "Launch new instance"},
        {"name": "stop_instance", "description": "Stop instance"},
    ]

ec2_agent = EC2Agent()
ec2_agent.run("Create 2 t2.micro instances, tag as app-server")
# Claude: checks if instances exist → creates 2 → returns IPs → confirms running

class ECSAgent(BaseAgent):
    AGENT_KEY = "ecs"
    SYSTEM_PROMPT = """You manage ECS services at production scale.
Rules: Task definition first, use Fargate, always configure auto-scaling."""

    CAPABILITIES = [
        {"name": "create_cluster", "description": "Create ECS cluster"},
        {"name": "register_task_definition", "description": "Register blueprint"},
        {"name": "create_service", "description": "Deploy service"},
        {"name": "setup_autoscaling", "description": "Configure scaling rules"},
    ]

ecs_agent = ECSAgent()
ecs_agent.run("Deploy with 5 tasks, auto-scale to 20 on high CPU, health monitoring")
# Claude orchestrates: cluster → task definition → service → autoscaling → verification

docker_agent.run("Run myapp:latest on port 8000", session_id="prod_001")
# ✅ Docker container running

ec2_agent.run("Create 2 instances for load balancing", session_id="prod_001")
# ✅ 2 EC2 instances up (same session = Claude remembers port 8000)

ecs_agent.run("Deploy with 5-20 auto-scaling, CloudWatch monitoring", session_id="prod_001")
# ✅ 5 ECS tasks running, auto-scaling 5-20 based on CPU

# Day 1
agent.run("Deploy on Docker with port 8000", session_id="user_123")

# Week later
agent.run("Move this to ECS for production", session_id="user_123")
# Claude reads history from DynamoDB:
# "I remember this app used port 8000. I'll keep that for ECS too."

# ❌ Wrong: crashes on error
result = dispatcher[tool_name](tool_input)

# ✅ Right: tell Claude about the error
try:
    result = dispatcher[tool_name](tool_input)
except Exception as e:
    result = {
        "error": str(e),
        "status": "failed",
        "request_id": request_id,  # For debugging
    }
# Claude sees the error and tries a different approach

from functools import wraps
import signal

def with_timeout(seconds=30):
    """Prevent tools from running forever"""
    def decorator(func):
        def wrapper(*args, **kwargs):
            def handler(signum, frame):
                raise TimeoutError(f"Tool execution exceeded {seconds}s")

            signal.signal(signal.SIGALRM, handler)
            signal.alarm(seconds)
            try:
                result = func(*args, **kwargs)
                signal.alarm(0)  # Cancel timeout
                return result
            except TimeoutError as e:
                return {"error": str(e), "status": "timeout"}
        return wrapper
    return decorator

@with_timeout(seconds=60)
def create_instance(params):
    """If EC2 creation hangs, timeout after 60 seconds"""
    ec2 = boto3.client("ec2")
    return ec2.create_instances(**params)

def create_instance(params):
    """Validate before executing"""
    # Check required fields
    required = ["ImageId", "InstanceType"]
    for field in required:
        if field not in params:
            return {"error": f"Missing required field: {field}", "status": "failed"}

    # Validate instance type
    valid_types = ["t2.micro", "t2.small", "t3.medium"]
    if params["InstanceType"] not in valid_types:
        return {
            "error": f"InstanceType must be one of {valid_types}",
            "status": "failed"
        }

    # Now safe to execute
    ec2 = boto3.client("ec2")
    return ec2.create_instances(**params)

import json
from datetime import datetime

def log_action(request_id: str, user: str, action: str, params: dict, result: dict):
    """Log everything for audits"""
    log_entry = {
        "request_id": request_id,
        "timestamp": datetime.utcnow().isoformat(),
        "user": user,
        "action": action,
        "params": json.dumps(params),  # What was requested
        "result_status": result.get("status"),
        "result_code": result.get("error"),
    }

    # Save to CloudWatch Logs or DynamoDB
    dynamodb.put_item(TableName="agent_audit", Item=log_entry)

# In the tool dispatcher:
result = dispatcher[tool_name](tool_input)
log_action(request_id, user_id, tool_name, tool_input, result)

MONTHLY_BUDGET = 1000  # $1000/month limit

def estimate_cost(action: str, params: dict) -> float:
    """Estimate AWS cost before executing"""
    if action == "create_instance":
        instance_type = params.get("InstanceType", "t2.micro")
        hourly_rates = {
            "t2.micro": 0.012,
            "t2.small": 0.023,
            "t3.medium": 0.042,
        }
        months_running = 1
        return hourly_rates.get(instance_type, 0) * 730 * months_running

    elif action == "create_rds":
        # RDS: ~$400/month for 20GB
        return 400

    return 0

# In the runnable:
estimated = estimate_cost(tool_name, tool_input)
if estimated > MONTHLY_BUDGET:
    return {
        "error": f"Cost ${estimated:.2f} exceeds budget ${MONTHLY_BUDGET}",
        "status": "rejected",
    }
# Only execute if under budget
result = dispatcher[tool_name](tool_input)

# Define who can do what
USER_PERMISSIONS = {
    "admin": ["create_instance", "delete_instance", "create_role", "delete_role"],
    "developer": ["create_instance", "stop_instance", "deploy_to_ecs"],
    "readonly": ["describe_instances", "list_containers", "get_logs"],
}

def check_permission(user_role: str, action: str) -> bool:
    """Prevent unauthorized actions"""
    allowed = USER_PERMISSIONS.get(user_role, [])
    if action not in allowed:
        return False
    return True

# Before executing:
if not check_permission(user_role, tool_name):
    return {
        "error": f"User role '{user_role}' cannot perform '{tool_name}'",
        "status": "permission_denied",
    }
result = dispatcher[tool_name](tool_input)

git clone https://github.com/jayakrishnayadav24/ai-agents
cd ai-agents
pip install -r requirements.txt

from agents.docker_agent import DockerAgent

agent = DockerAgent()
response = agent.run(
    "Deploy FastAPI app on port 8000 with health check"
)
print(response)