Building an AI-Powered DevOps Agent — Part 2

From Local RCA to Bedrock Knowledge Base, SlackOps, Approval Workflow, and Continuous Learning

In Part 1, I built the foundation of an AI-powered DevOps Agent locally.

That first phase covered:

Amazon Nova Lite
Minikube
Local Kubernetes workloads
Local Markdown runbooks
FAISS vector search
Root Cause Analysis

The first version helped me prove one important idea:

Can an AI agent investigate Kubernetes issues and generate useful RCA locally?

Once that worked, the next question was bigger:

How do we evolve this into a more realistic DevOps workflow?

In real incidents, engineers do not only need an RCA.

They need:

Knowledge Base lookup
Issue classification
Remediation decision
Confidence level
Human approval
Slack interaction
Execution history
Verification
Incident tracking
Environment policy
Continuous learning

So in this second part, I will not repeat the local runbook and FAISS setup from Part 1.

Instead, I will continue from where Part 1 ended and explain how I expanded the local RCA agent into a more operational AI DevOps workflow.

The main goal was:

Let the AI agent investigate and recommend,
but keep humans in control of operational changes.

Where Part 2 Starts

At the end of Part 1, the flow looked like this:

Engineer runs local command
        ↓
Agent collects Kubernetes evidence
        ↓
Agent searches local runbooks
        ↓
Agent generates RCA
        ↓
Result is printed locally

That was enough for the first phase.

But a real DevOps workflow needs more than local RCA.

The next target architecture became:

Engineer or Slack asks for investigation
        ↓
Agent collects Kubernetes evidence
        ↓
Agent generates a semantic query
        ↓
Agent searches Amazon Bedrock Knowledge Base
        ↓
Agent classifies the issue
        ↓
Agent checks remediation catalog
        ↓
Agent calculates confidence
        ↓
Agent sends Slack RCA card
        ↓
Engineer approves or rejects remediation
        ↓
Agent executes approved action only
        ↓
Agent verifies recovery
        ↓
Agent records history
        ↓
Agent learns from human fixes

This article explains how I added each of those capabilities one by one.

Clean Project Structure for Part 2

During development, I created multiple versioned files while experimenting.

That is normal while building.

But for an article, it is better to explain the system using clean logical file names.

The Part 2 structure can be explained like this:

ai-devops-agent/
├── main.py
├── main_api.py
├── orchestrator.py
│
├── agents/
│   ├── query_agent.py
│   ├── knowledge_agent.py
│   ├── rca_agent.py
│   ├── issue_classifier_agent.py
│   ├── remediation_planner.py
│   ├── confidence_agent.py
│   ├── policy_agent.py
│   ├── health_scanner_agent.py
│   └── learning_agent.py
│
├── tools/
│   ├── kubernetes_tools.py
│   ├── bedrock_kb_tools.py
│   ├── remediation_catalog_tools.py
│   ├── deployment_inspector_tools.py
│   ├── command_executor.py
│   ├── slack_tools.py
│   └── cluster_health_tools.py
│
├── remediation_catalog/
│   └── remediations.yaml
│
├── observability/
│   ├── traces.json
│   ├── agent_tracer.py
│   ├── active_incidents.json
│   └── incident_store.py
│
├── execution_history/
│   ├── executions.json
│   └── execution_store.py
│
├── slack/
│   └── incident_card_builder.py
│
├── config/
│   └── cluster_registry.yaml
│
└── learning/
    ├── pending_learnings.json
    ├── learning_store.py
    ├── drafts/
    └── promoted/

The purpose of this structure is separation of responsibility.

orchestrator.py                  → Controls the full workflow
agents/                          → AI reasoning and decision components
tools/                           → Kubernetes, Slack, Bedrock, and execution utilities
remediation_catalog/             → Approved remediation definitions
observability/                   → RCA traces and active incidents
execution_history/               → Past remediation outcomes
slack/                           → Slack message rendering
config/                          → Environment and cluster policy
learning/                        → Human fixes and generated runbooks

Challenge 1: Move from Local Runbook Search to Amazon Bedrock Knowledge Base

Part 1 used local Markdown runbooks and FAISS.

That was useful for learning.

But for a more realistic platform, I wanted the agent to retrieve operational knowledge from a managed Knowledge Base.

In real teams, knowledge may come from:

Runbooks
Incident reports
Jira tickets
Confluence pages
Postmortems
Application support notes
Architecture documents

So the next step was to introduce Amazon Bedrock Knowledge Base.

New File: tools/bedrock_kb_tools.py

This file is responsible for calling Amazon Bedrock Knowledge Base.

Its job is simple:

Receive a query
Search Bedrock Knowledge Base
Return matched content, score, and source

import os
import boto3
KNOWLEDGE_BASE_ID = os.getenv("BEDROCK_KB_ID")
AWS_REGION = os.getenv("AWS_REGION", "us-east-1")
bedrock_agent_runtime = boto3.client(
    "bedrock-agent-runtime",
    region_name=AWS_REGION,
)
def retrieve_from_bedrock_kb(query: str, top_k: int = 3) -> str:
    response = bedrock_agent_runtime.retrieve(
        knowledgeBaseId=KNOWLEDGE_BASE_ID,
        retrievalQuery={
            "text": query,
        },
        retrievalConfiguration={
            "vectorSearchConfiguration": {
                "numberOfResults": top_k,
            }
        },
    )
    results = []
    for item in response.get("retrievalResults", []):
        content = item.get("content", {}).get("text", "")
        score = item.get("score", "N/A")
        location = item.get("location", {})
        s3_uri = location.get("s3Location", {}).get("uri", "unknown")
        results.append(
            f"""
Matched Source: {s3_uri}
Score: {score}
Content:
{content}
"""
        )
    if not results:
        return "No matching Bedrock Knowledge Base result found."
    return "\n---\n".join(results)

New File: agents/knowledge_agent.py

The Knowledge Agent wraps the Bedrock KB tool.

from tools.bedrock_kb_tools import retrieve_from_bedrock_kb
def search_knowledge_base(query: str) -> str:
    return retrieve_from_bedrock_kb(
        query=query,
        top_k=3,
    )

This keeps Bedrock-specific code outside the orchestrator.

The orchestrator should not care whether the knowledge comes from local files, Bedrock KB, Confluence, or another source.

It only asks:

Give me relevant operational knowledge for this issue.

Update to orchestrator.py

In Part 1, the orchestrator already collected Kubernetes evidence and generated RCA.

Now we add Knowledge Base lookup before RCA generation.

from agents.knowledge_agent import search_knowledge_base
def investigate_application(app_name: str, namespace: str):
    evidence = collect_application_evidence(
        app_name=app_name,
        namespace=namespace,
    )
    search_query = generate_search_query(evidence)
    kb_context = search_knowledge_base(search_query)
    rca = generate_rca(
        evidence=evidence,
        knowledge_context=kb_context,
    )
    return {
        "application": app_name,
        "namespace": namespace,
        "evidence": evidence,
        "search_query": search_query,
        "knowledge_context": kb_context,
        "rca": rca,
    }

Outcome

The agent can now enrich RCA using managed operational knowledge.

Example:

Generated Query:
Kubernetes ImagePullBackOff ErrImagePull invalid image tag

Matched Source:
s3://ai-devops-kb/runbooks/imagepullbackoff.md

Score:
0.84

The RCA is no longer based only on live Kubernetes evidence.

It now combines:

Kubernetes evidence
+
Knowledge Base context
+
Amazon Nova Lite reasoning

Challenge 2: Generate Better Knowledge Base Queries Dynamically

After adding Bedrock Knowledge Base, the next issue was query quality.

Vector search is only useful when the query is good.

A weak query returns weak runbook matches.

So I introduced a Query Agent.

New File: agents/query_agent.py

The Query Agent reads Kubernetes evidence and creates a short semantic query.

import os
import boto3
import json
BEDROCK_MODEL_ID = os.getenv(
    "BEDROCK_MODEL_ID",
    "us.amazon.nova-lite-v1:0",
)
bedrock_runtime = boto3.client(
    "bedrock-runtime",
    region_name=os.getenv("AWS_REGION", "us-east-1"),
)
def generate_search_query(evidence: dict) -> str:
    prompt = f"""
Create one short Kubernetes troubleshooting search query from this evidence.Rules:
- Return only the query.
- Do not explain.
- Do not use markdown.
Evidence:
{json.dumps(evidence, indent=2)}
"""
    response = bedrock_runtime.converse(
        modelId=BEDROCK_MODEL_ID,
        messages=[
            {
                "role": "user",
                "content": [{"text": prompt}],
            }
        ],
    )
    return response["output"]["message"]["content"][0]["text"].strip()

Update to orchestrator.py

The orchestrator now calls the query agent before searching Bedrock KB.

from agents.query_agent import generate_search_query
from agents.knowledge_agent import search_knowledge_base
search_query = generate_search_query(evidence)
kb_context = search_knowledge_base(search_query)

Outcome

Example evidence:

Pod status: CrashLoopBackOff
Exit code: 137
Logs: memory allocation error

Generated query:

container exited with code 137 memory allocation out of memory kubernetes

This improves KB retrieval quality because the query is based on actual evidence, not application name.

Challenge 3: Classify the Issue Type from Evidence

Once the agent had Kubernetes evidence and Knowledge Base context, I needed a reliable issue type.

The issue type is important because it controls:

Remediation catalog lookup
Risk level
Confidence score
Slack message
Learning workflow

The issue type should not be hardcoded by application name.

So I introduced the Issue Classifier Agent.

New File: agents/issue_classifier_agent.py

import os
import boto3
import json
BEDROCK_MODEL_ID = os.getenv(
    "BEDROCK_MODEL_ID",
    "us.amazon.nova-lite-v1:0",
)
bedrock_runtime = boto3.client(
    "bedrock-runtime",
    region_name=os.getenv("AWS_REGION", "us-east-1"),
)
def classify_issue(evidence: dict, rca: str, kb_context: str) -> str:
    prompt = f"""
You are a Kubernetes issue classifier.
Classify the issue into exactly one of these values:
- ImagePullBackOff
- MemoryIssue
- DependencyFailure
- PermissionIssue
- TLSIssue
- UnknownIssue
Rules:
- ImagePullBackOff if evidence contains ImagePullBackOff, ErrImagePull, image not found, or manifest unknown.
- MemoryIssue if evidence contains OOMKilled, exit code 137, or memory allocation error.
- DependencyFailure if logs show database connection failed, connection refused, Redis failure, or upstream service unavailable.
- PermissionIssue if logs show permission denied, access denied, or read-only file system.
- TLSIssue if logs show certificate, truststore, SSL, or CA failure.
- UnknownIssue if unclear.
Return only the issue type.
Evidence:
{json.dumps(evidence, indent=2)}
RCA:
{rca}
Knowledge Base Context:
{kb_context}
"""
    response = bedrock_runtime.converse(
        modelId=BEDROCK_MODEL_ID,
        messages=[
            {
                "role": "user",
                "content": [{"text": prompt}],
            }
        ],
    )
    return response["output"]["message"]["content"][0]["text"].strip()

Update to orchestrator.py

from agents.issue_classifier_agent import classify_issue
issue_type = classify_issue(
    evidence=evidence,
    rca=rca,
    kb_context=kb_context,
)

Return issue type as part of the final result:

return {
    "application": app_name,
    "namespace": namespace,
    "evidence": evidence,
    "search_query": search_query,
    "knowledge_context": kb_context,
    "rca": rca,
    "issue_type": issue_type,
}

Outcome

The agent can now return:

Application: memory-app
Issue Type: MemoryIssue

or:

Application: payment-api
Issue Type: ImagePullBackOff

This makes the system application-independent.

Any workload can fail with any supported issue type.

Challenge 4: Add a Remediation Catalog

After classification, the next question is:

Can we safely suggest a fix?

This is where I introduced the remediation catalog.

The key safety rule is:

The LLM should not generate and execute arbitrary shell commands.

Instead:

LLM identifies the issue
        ↓
System checks remediation catalog
        ↓
Catalog decides whether remediation is executable
        ↓
Human approves
        ↓
System executes catalog-approved command only

New Folder: remediation_catalog/

remediation_catalog/
└── remediations.yaml

File: remediation_catalog/remediations.yaml

ImagePullBackOff:
  executable: true
  risk: LOW
  requires_approval: true
  requires_inspection: true
  remediation:
    - Verify image tag exists
    - Detect deployment container name
    - Update image tag safely
    - Verify rollout
  command_template: |
    kubectl set image deployment/{deployment_name} {container_name}=nginx:latest -n {namespace}MemoryIssue:
  executable: false
  risk: MEDIUM
  requires_approval: true
  requires_inspection: true
  remediation:
    - Inspect current memory requests and limits
    - Review OOMKilled events
    - Generate patch recommendation only
    - Human review required
DependencyFailure:
  executable: false
  risk: MEDIUM
  requires_approval: true
  requires_inspection: true
  remediation:
    - Validate dependency configuration
    - Validate secrets
    - Validate service endpoints
    - Validate network policies
PermissionIssue:
  executable: false
  risk: MEDIUM
  requires_approval: true
  requires_inspection: true
  remediation:
    - Inspect securityContext
    - Inspect volume mounts
    - Check readOnlyRootFilesystem
    - Validate filesystem permissions
TLSIssue:
  executable: false
  risk: MEDIUM
  requires_approval: true
  requires_inspection: true
  remediation:
    - Validate certificate chain
    - Validate internal CA bundle
    - Check truststore configuration
    - Review mounted certificate path
UnknownIssue:
  executable: false
  risk: HIGH
  requires_approval: false
  requires_inspection: true
  remediation:
    - Manual investigation required
    - Record actual fix after resolution
    - Generate draft runbook

New File: tools/remediation_catalog_tools.py

from pathlib import Path
import yaml
CATALOG_PATH = Path("remediation_catalog/remediations.yaml")
def load_remediation_catalog() -> dict:
    return yaml.safe_load(
        CATALOG_PATH.read_text()
    )
def get_remediation(issue_type: str) -> dict:
    catalog = load_remediation_catalog()
    return catalog.get(
        issue_type,
        catalog["UnknownIssue"],
    )

Update to orchestrator.py

from tools.remediation_catalog_tools import get_remediation
remediation = get_remediation(issue_type)

Outcome

The agent can now make safer decisions.

Example:

Issue Type: ImagePullBackOff
Risk: LOW
Executable: true
Approval Required: true

Another example:

Issue Type: MemoryIssue
Risk: MEDIUM
Executable: false
Action: Investigation only

This separates reasoning from action.

The AI explains the issue.

The remediation catalog controls what can be executed.

Challenge 5: Inspect Deployment Before Building a Fix

For executable remediation, the system needs real deployment details.

For example:

Deployment name
Container name
Current image
Namespace

We should not assume the container name.

So I introduced deployment inspection.

New File: tools/deployment_inspector_tools.py

import json
from tools.kubernetes_tools import run_command
def inspect_deployment(app_name: str, namespace: str) -> dict:
    output = run_command([
        "kubectl",
        "get",
        "deployment",
        app_name,
        "-n",
        namespace,
        "-o",
        "json",
    ])
    data = json.loads(output)
    containers = data["spec"]["template"]["spec"]["containers"]
    return {
        "deployment_name": data["metadata"]["name"],
        "namespace": namespace,
        "replicas": data["spec"].get("replicas", 1),
        "containers": [
            {
                "name": container["name"],
                "image": container["image"],
                "resources": container.get("resources", {}),
            }
            for container in containers
        ],
    }
def get_primary_container_name(app_name: str, namespace: str) -> str:
    deployment = inspect_deployment(
        app_name=app_name,
        namespace=namespace,
    )
    return deployment["containers"][0]["name"]

Update to orchestrator.py

from tools.deployment_inspector_tools import (
    inspect_deployment,
    get_primary_container_name,
)

deployment_info = inspect_deployment(
    app_name=app_name,
    namespace=namespace,
)

container_name = get_primary_container_name(
    app_name=app_name,
    namespace=namespace,
)

Outcome

The remediation plan can use real deployment details.

Example:

{
  "deployment_name": "payment-api",
  "namespace": "dev",
  "containers": [
    {
      "name": "nginx",
      "image": "nginx:wrongtag"
    }
  ]
}

This makes remediation safer and avoids hardcoded assumptions.

Challenge 6: Build a Remediation Plan

Now the agent has:

Issue type
Remediation catalog entry
Deployment name
Container name
Namespace

The next step is to combine those details into a remediation plan.

New File: agents/remediation_planner.py

def build_remediation_plan(
    issue_type: str,
    remediation: dict,
    deployment_info: dict,
    container_name: str,
    namespace: str,
) -> dict:
    command = None
    if remediation.get("executable"):
        command_template = remediation.get("command_template", "")
        command = command_template.format(
            deployment_name=deployment_info["deployment_name"],
            container_name=container_name,
            namespace=namespace,
        ).strip()
    return {
        "issue_type": issue_type,
        "risk": remediation.get("risk"),
        "executable": remediation.get("executable"),
        "requires_approval": remediation.get("requires_approval"),
        "steps": remediation.get("remediation", []),
        "command": command,
    }

Update to orchestrator.py

from agents.remediation_planner import build_remediation_plan
remediation_plan = build_remediation_plan(
    issue_type=issue_type,
    remediation=remediation,
    deployment_info=deployment_info,
    container_name=container_name,
    namespace=namespace,
)

Outcome

For ImagePullBackOff, the plan may look like this:

{
  "issue_type": "ImagePullBackOff",
  "risk": "LOW",
  "executable": true,
  "requires_approval": true,
  "command": "kubectl set image deployment/payment-api nginx=nginx:latest -n dev"
}

For MemoryIssue, the plan may look like this:

{
  "issue_type": "MemoryIssue",
  "risk": "MEDIUM",
  "executable": false,
  "requires_approval": true,
  "steps": [
    "Inspect current memory requests and limits",
    "Review OOMKilled events",
    "Generate patch recommendation only",
    "Human review required"
  ]
}

Challenge 7: Add Confidence Scoring

Not every recommendation should be trusted equally.

The agent should consider:

Evidence quality
Knowledge Base match
Issue type
Remediation catalog match
Previous execution history
Risk level
Verification availability

So I added confidence scoring.

New Folder: execution_history/

execution_history/
└── executions.json

Example:

[
  {
    "application": "payment-api",
    "namespace": "dev",
    "issue_type": "ImagePullBackOff",
    "result": "success",
    "timestamp": "2026-06-10T10:00:00"
  }
]

New File: execution_history/execution_store.py

from pathlib import Path
import json
from datetime import datetime
EXECUTION_FILE = Path("execution_history/executions.json")
def load_executions() -> list[dict]:
    if not EXECUTION_FILE.exists():
        return []
    return json.loads(
        EXECUTION_FILE.read_text()
    )
def save_execution(record: dict) -> None:
    executions = load_executions()
    record["timestamp"] = datetime.now().isoformat()
    executions.append(record)
    EXECUTION_FILE.write_text(
        json.dumps(executions, indent=2)
    )
def get_execution_history(issue_type: str) -> list[dict]:
    executions = load_executions()
    return [
        item for item in executions
        if item.get("issue_type") == issue_type
    ]

New File: agents/confidence_agent.py

from execution_history.execution_store import get_execution_history
def calculate_confidence(
    issue_type: str,
    kb_context: str,
    remediation_plan: dict,
) -> dict:
    score = 0
    reasons = []
    if issue_type != "UnknownIssue":
        score += 25
        reasons.append("Known issue type detected")
    if kb_context and "No matching" not in kb_context:
        score += 25
        reasons.append("Knowledge Base match found")
    if remediation_plan.get("executable") is not None:
        score += 20
        reasons.append("Remediation catalog entry found")
    history = get_execution_history(issue_type)
    successful = [
        item for item in history
        if item.get("result") == "success"
    ]
    if len(successful) >= 2:
        score += 20
        reasons.append("Previous successful remediation history found")
    if remediation_plan.get("risk") == "HIGH":
        score -= 20
        reasons.append("High risk issue")
    if issue_type == "UnknownIssue":
        score -= 30
        reasons.append("Unknown issue type")
    score = max(0, min(score, 100))
    if score >= 85:
        level = "HIGH"
    elif score >= 60:
        level = "MEDIUM"
    else:
        level = "LOW"
    return {
        "score": score,
        "level": level,
        "reasons": reasons,
    }

Update to orchestrator.py

from agents.confidence_agent import calculate_confidence

confidence = calculate_confidence(
    issue_type=issue_type,
    kb_context=kb_context,
    remediation_plan=remediation_plan,
)

Outcome

The agent can now explain whether it has enough confidence to suggest a fix.

Example:

Issue Type: ImagePullBackOff
KB Match: Yes
Catalog Match: Yes
Previous Success: Yes
Risk: LOW
Confidence: HIGH

Another example:

Issue Type: UnknownIssue
KB Match: Weak
Catalog Match: No
Risk: HIGH
Confidence: LOW

This is important because a safe DevOps agent must know when not to act.

Challenge 8: Add Human Approval Before Execution

Even if confidence is high, I do not want the agent to directly change Kubernetes without approval.

So the next step is approval.

In the local version, approval can start with terminal input.

Update to orchestrator.py

def should_request_approval(remediation_plan: dict, confidence: dict) -> bool:
    return (
        remediation_plan.get("executable") is True
        and remediation_plan.get("requires_approval") is True
        and confidence["level"] in ["HIGH", "MEDIUM"]
    )

Then:

if should_request_approval(remediation_plan, confidence):
    approval = input("Approve remediation? (Y/N): ").strip().lower()

    if approval != "y":
        save_execution({
            "application": app_name,
            "namespace": namespace,
            "issue_type": issue_type,
            "result": "rejected",
            "approved": False,
        })

        return {
            "status": "REJECTED",
            "message": "Remediation rejected by human",
        }

Outcome

The local workflow now follows an important rule:

AI recommends.
Human approves.
Automation executes.
System verifies.

This same approval model later moves into Slack.

Challenge 9: Execute Only Approved Remediation

After approval, the system can execute the remediation command.

But the command must come from the remediation catalog.

It must not come directly from the LLM.

New File: tools/command_executor.py

import subprocess
def execute_command(command: str) -> dict:
    result = subprocess.run(
        command,
        shell=True,
        capture_output=True,
        text=True,
        timeout=60,
    )
    return {
        "return_code": result.returncode,
        "stdout": result.stdout,
        "stderr": result.stderr,
        "success": result.returncode == 0,
    }

Update to orchestrator.py

from tools.command_executor import execute_command

execution_result = execute_command(
    remediation_plan["command"]
)

Outcome

At this point, the system can execute approved remediation.

But execution alone is not enough.

We still need verification.

Challenge 10: Verify Recovery After Execution

A command can succeed, but the application can still be unhealthy.

For example:

kubectl set image deployment/payment-api nginx=nginx:latest -n dev

may complete successfully, but the rollout may fail.

So I added verification.

Update to orchestrator.py

def verify_rollout(app_name: str, namespace: str) -> dict:
    command = (
        f"kubectl rollout status deployment/{app_name} "
        f"-n {namespace} --timeout=60s"
    )
    result = execute_command(command)
    return {
        "command": command,
        "success": result["success"],
        "output": result["stdout"] or result["stderr"],
    }

Then:

verification = verify_rollout(
    app_name=app_name,
    namespace=namespace,
)

Save execution result:

save_execution({
    "application": app_name,
    "namespace": namespace,
    "issue_type": issue_type,
    "command": remediation_plan["command"],
    "result": "success" if verification["success"] else "failed",
    "approved": True,
})

Outcome

The system now has a complete local remediation loop:

Investigate
        ↓
Generate RCA
        ↓
Classify issue
        ↓
Find remediation
        ↓
Calculate confidence
        ↓
Ask approval
        ↓
Execute
        ↓
Verify
        ↓
Save result

Challenge 11: Store RCA and Execution Traces

Once the agent starts investigating and executing actions, an audit trail becomes important.

Slack messages and terminal output are not enough.

So I added a trace store.

New Folder: observability/

observability/
└── traces.json

New File: observability/agent_tracer.py

from pathlib import Path
import json
from datetime import datetime
TRACE_FILE = Path("observability/traces.json")
def save_trace(record: dict) -> None:
    if TRACE_FILE.exists():
        traces = json.loads(
            TRACE_FILE.read_text()
        )
    else:
        traces = []
    record["timestamp"] = datetime.now().isoformat()
    traces.append(record)
    TRACE_FILE.write_text(
        json.dumps(traces, indent=2)
    )

Update to orchestrator.py

At the end of the investigation:

from observability.agent_tracer import save_trace

save_trace({
    "application": app_name,
    "namespace": namespace,
    "issue_type": issue_type,
    "search_query": search_query,
    "confidence": confidence,
    "rca": rca,
    "remediation_plan": remediation_plan,
    "execution_result": execution_result,
    "verification": verification,
})

Outcome

Now every investigation can be reviewed later.

This helps with:

Audit
Debugging
Incident review
Confidence improvement
Runbook generation

Challenge 12: Bring the Agent into Slack

Until now, the workflow was mostly terminal-based.

But in real incidents, engineers work in Slack.

So the next goal was:

Engineer asks in Slack
        ↓
Agent investigates
        ↓
Agent posts RCA back to Slack

This is where main_api.py comes in.

New File: main_api.py

This file exposes a local FastAPI server for Slack.

from fastapi import FastAPI, Request
from orchestrator import investigate_application
from tools.slack_tools import send_slack_message
app = FastAPI()
@app.post("/slack/events")
async def slack_events(request: Request):
    body = await request.json()
    if body.get("type") == "url_verification":
        return {
            "challenge": body["challenge"]
        }
    event = body.get("event", {})
    text = event.get("text", "")
    if "investigate" in text:
        app_name = extract_app_name(text)
        result = investigate_application(
            app_name=app_name,
            namespace="dev",
        )
        send_slack_message(
            text=format_investigation_result(result)
        )
    return {"status": "ok"}
def extract_app_name(text: str) -> str:
    words = text.split()
    if "investigate" in words:
        index = words.index("investigate")
        return words[index + 1]
    raise ValueError("Application name not found")
def format_investigation_result(result: dict) -> str:
    return f"""
🚨 AI DevOps Investigation
Application: {result["application"]}
Namespace: {result["namespace"]}
Issue Type: {result["issue_type"]}
Confidence: {result["confidence"]["level"]}
Root Cause:
{result["rca"]}
"""

New File: tools/slack_tools.py

import os
import requests

SLACK_BOT_TOKEN = os.getenv("SLACK_BOT_TOKEN")
SLACK_CHANNEL_ID = os.getenv("SLACK_CHANNEL_ID")

def send_slack_message(text: str):
    response = requests.post(
        "https://slack.com/api/chat.postMessage",
        headers={
            "Authorization": f"Bearer {SLACK_BOT_TOKEN}",
            "Content-Type": "application/json",
        },
        json={
            "channel": SLACK_CHANNEL_ID,
            "text": text,
        },
        timeout=30,
    )
    response.raise_for_status()
    return response.json()

Run the API

uvicorn main_api:app --host 0.0.0.0 --port 8000

Expose it with ngrok:

ngrok http 8000

Configure Slack:

Event Subscription URL:
https://your-ngrok-url.ngrok-free.app/slack/events

Now the engineer can ask:

@devopsbot investigate memory-app

Outcome

The agent is now available from Slack.

Before:

Engineer runs Python command in terminal

After:

Engineer asks from Slack

This is the beginning of SlackOps.

Challenge 13: Add Slack Approval Buttons

After Slack investigation worked, the next step was approval.

The engineer should be able to approve or reject remediation from Slack.

For that, Slack interactivity is required.

Update Slack App Configuration

Enable Interactivity in Slack App settings.

Set request URL:

https://your-ngrok-url.ngrok-free.app/slack/actions

Update to main_api.py

Add a new endpoint:

import json
from fastapi import Form
from tools.command_executor import execute_command

@app.post("/slack/actions")
async def slack_actions(payload: str = Form(...)):
    data = json.loads(payload)

    action = data["actions"][0]
    action_value = action["value"]
    user = data["user"]["username"]

    if action_value.startswith("approve:"):
        command = action_value.replace("approve:", "")

        result = execute_command(command)

        send_slack_message(
            text=f"""
✅ Remediation Approved

Approved By: {user}

Command:
{command}

Result:
{result["stdout"] or result["stderr"]}
"""
        )

    if action_value.startswith("reject:"):
        send_slack_message(
            text=f"❌ Remediation rejected by {user}"
        )

    return {"status": "ok"}

New File: slack/incident_card_builder.py

Instead of sending plain text, we can build a Slack Block Kit card.

def build_incident_card(result: dict) -> list[dict]:
    remediation_plan = result.get("remediation_plan", {})
    command = remediation_plan.get("command")
    summary_text = f"""
*🚨 AI DevOps Investigation*
*Application:* {result["application"]}
*Environment:* {result.get("environment", result["namespace"])}
*Issue Type:* {result["issue_type"]}
*Confidence:* {result["confidence"]["level"]}
*Risk:* {remediation_plan.get("risk")}
*Root Cause:*
{result["rca"]}
*Suggested Action:*
{format_steps(remediation_plan.get("steps", []))}
"""
    blocks = [
        {
            "type": "section",
            "text": {
                "type": "mrkdwn",
                "text": summary_text,
            },
        }
    ]
    if command:
        blocks.append({
            "type": "actions",
            "elements": [
                {
                    "type": "button",
                    "text": {"type": "plain_text", "text": "Approve Fix"},
                    "style": "primary",
                    "value": f"approve:{command}",
                },
                {
                    "type": "button",
                    "text": {"type": "plain_text", "text": "Reject"},
                    "style": "danger",
                    "value": "reject",
                },
                {
                    "type": "button",
                    "text": {"type": "plain_text", "text": "Record Actual Fix"},
                    "value": "record_actual_fix",
                },
            ],
        })
    return blocks
def format_steps(steps: list[str]) -> str:
    return "\n".join([f"• {step}" for step in steps])

Outcome

The workflow now becomes:

Slack investigation request
        ↓
Agent investigates
        ↓
Slack RCA card appears
        ↓
Engineer clicks Approve Fix
        ↓
Command executes
        ↓
Slack shows execution result

This keeps human approval in the loop.

Challenge 14: Simulate Alert-Driven Investigation Locally

In production, I would not replace monitoring tools.

Most teams already use:

Prometheus
Grafana Alertmanager
CloudWatch
Datadog
PagerDuty

Those tools detect symptoms.

The AI DevOps Agent should investigate the symptom.

At this stage, I had not yet integrated Prometheus or Grafana.

So I added a local health scanner to simulate alert-driven investigation.

New File: tools/cluster_health_tools.py

from tools.kubernetes_tools import run_command
def get_all_pods(namespace: str) -> str:
    return run_command([
        "kubectl",
        "get",
        "pods",
        "-n",
        namespace,
    ])
def get_warning_events(namespace: str) -> str:
    return run_command([
        "kubectl",
        "get",
        "events",
        "-n",
        namespace,
        "--field-selector",
        "type=Warning",
        "--sort-by=.lastTimestamp",
    ])

New File: agents/health_scanner_agent.py

from tools.cluster_health_tools import (
    get_all_pods,
    get_warning_events,
)
def scan_namespace(namespace: str) -> list[dict]:
    pods_output = get_all_pods(namespace)
    warning_events = get_warning_events(namespace)
    findings = []
    for line in pods_output.splitlines():
        if "ImagePullBackOff" in line:
            findings.append({
                "namespace": namespace,
                "application": line.split()[0].split("-")[0],
                "pod": line.split()[0],
                "issue_type": "ImagePullBackOff",
                "reason": "ImagePullBackOff",
                "events": warning_events,
            })
        if "CrashLoopBackOff" in line:
            findings.append({
                "namespace": namespace,
                "application": line.split()[0].split("-")[0],
                "pod": line.split()[0],
                "issue_type": "CrashLoopBackOff",
                "reason": "CrashLoopBackOff",
                "events": warning_events,
            })
    return findings

Update to main.py

Now main.py can support scanning.

from agents.health_scanner_agent import scan_namespace
from orchestrator import investigate_application

def run_health_scan():
    namespaces = ["dev", "uat", "prod"]

    for namespace in namespaces:
        findings = scan_namespace(namespace)

        for finding in findings:
            result = investigate_application(
                app_name=finding["application"],
                namespace=namespace,
            )

            print(result["rca"])

Outcome

The current local flow becomes:

python main.py
        ↓
Health scanner checks namespaces
        ↓
Unhealthy pod found
        ↓
Orchestrator investigates
        ↓
Slack RCA card is sent

Future production flow:

Prometheus / Grafana / CloudWatch
        ↓
Alert webhook
        ↓
main_api.py
        ↓
orchestrator.py
        ↓
Same investigation workflow

The trigger changes, but the investigation engine stays the same.

Challenge 15: Avoid Repeated Slack Alerts

If the scanner runs multiple times, the same issue may be detected again and again.

Without state, Slack can get noisy.

So I added an incident store.

New File: observability/incident_store.py

from pathlib import Path
import json
from datetime import datetime
INCIDENT_FILE = Path("observability/active_incidents.json")
def load_incidents() -> dict:
    if not INCIDENT_FILE.exists():
        return {}
    return json.loads(
        INCIDENT_FILE.read_text()
    )
def save_incidents(incidents: dict) -> None:
    INCIDENT_FILE.write_text(
        json.dumps(incidents, indent=2)
    )
def build_incident_id(namespace: str, application: str, issue_type: str) -> str:
    return f"{namespace}:{application}:{issue_type}"
def incident_exists(incident_id: str) -> bool:
    incidents = load_incidents()
    return incident_id in incidents
def create_or_update_incident(incident_id: str, finding: dict) -> bool:
    incidents = load_incidents()
    is_new = incident_id not in incidents
    if is_new:
        finding["first_seen"] = datetime.now().isoformat()
        finding["status"] = "OPEN"
    finding["last_seen"] = datetime.now().isoformat()
    incidents[incident_id] = finding
    save_incidents(incidents)
    return is_new

Update to main.py

from observability.incident_store import (
    build_incident_id,
    create_or_update_incident,
)

for finding in findings:
    incident_id = build_incident_id(
        namespace=finding["namespace"],
        application=finding["application"],
        issue_type=finding["issue_type"],
    )
    is_new = create_or_update_incident(
        incident_id=incident_id,
        finding=finding,
    )
    if not is_new:
        continue
    result = investigate_application(
        app_name=finding["application"],
        namespace=finding["namespace"],
    )
    send_slack_incident(result)

Outcome

Now the agent sends Slack messages only for new incidents.

This keeps the Slack channel cleaner.

The agent can still update the incident timestamp internally.

Challenge 16: Add Dev, UAT, and Prod Policies

After the agent worked in one namespace, I wanted to test environment-specific behavior.

In real life, Dev, UAT, and Prod should not have the same policy.

For example:

Dev  → approved remediation allowed
UAT  → approved remediation allowed with strict verification
Prod → recommendation only, no automated remediation

New File: config/cluster_registry.yaml

clusters:
  dev:
    context: minikube
    namespace: dev
    environment: dev
    remediation_allowed: true
    verification_required: true

  uat:
    context: minikube
    namespace: uat
    environment: uat
    remediation_allowed: true
    verification_required: true

  prod:
    context: minikube
    namespace: prod
    environment: prod
    remediation_allowed: false
    verification_required: true

New File: agents/policy_agent.py

def apply_environment_policy(
    environment: str,
    remediation_plan: dict,
) -> dict:
    if environment == "prod":
        return {
            **remediation_plan,
            "remediation_allowed": False,
            "decision": "RECOMMENDATION_ONLY",
        }
    if remediation_plan.get("executable"):
        return {
            **remediation_plan,
            "remediation_allowed": True,
            "decision": "APPROVAL_RECOMMENDED",
        }
    return {
        **remediation_plan,
        "remediation_allowed": False,
        "decision": "INVESTIGATION_ONLY",
    }

Update to orchestrator.py

from agents.policy_agent import apply_environment_policy

policy_decision = apply_environment_policy(
    environment=environment,
    remediation_plan=remediation_plan,
)

Return:

return {
    "application": app_name,
    "namespace": namespace,
    "environment": environment,
    "issue_type": issue_type,
    "rca": rca,
    "confidence": confidence,
    "remediation_plan": policy_decision,
}

Outcome

The same issue now behaves differently by environment.

For Dev:

Approve Fix button is shown.

For Prod:

Recommendation only.
No automated fix button.

This is closer to real enterprise operations.

Challenge 17: Record the Actual Human Fix

Not every issue will be solved by the agent.

Sometimes the agent may have low confidence, weak KB match, or no remediation catalog entry.

In that case, the engineer investigates manually.

But after the engineer fixes it, that knowledge should not be lost.

So I introduced a learning workflow.

New Folder: learning/

learning/
├── pending_learnings.json
├── drafts/
└── promoted/

New File: learning/learning_store.py

from pathlib import Path
import json
from datetime import datetime

LEARNING_FILE = Path("learning/pending_learnings.json")

def load_pending_learnings() -> list[dict]:
    if not LEARNING_FILE.exists():
        return []

    return json.loads(
        LEARNING_FILE.read_text()
    )

def record_actual_fix(
    application: str,
    environment: str,
    root_cause: str,
    actual_fix: str,
    verification: str,
) -> dict:
    records = load_pending_learnings()

    record = {
        "application": application,
        "environment": environment,
        "root_cause": root_cause,
        "actual_fix": actual_fix,
        "verification": verification,
        "status": "pending_review",
        "created_at": datetime.now().isoformat(),
    }

    records.append(record)

    LEARNING_FILE.write_text(
        json.dumps(records, indent=2)
    )

    return record

CLI Example

python main.py \
  --record-fix \
  --application tls-trust-app \
  --environment dev \
  --root-cause "Internal CA not in JVM truststore" \
  --actual-fix "Mounted internal-ca.pem and updated JAVA_OPTS truststore path" \
  --verification "Pod reached Running and TLS handshake succeeded"

Slack Flow

The same action can be triggered from Slack:

[Record Actual Fix]

The engineer can provide:

Root Cause
Actual Fix
Verification Method

Challenge 18: Generate a Draft Runbook from the Human Fix

Once the actual fix is recorded, the agent can generate a draft runbook.

This is where the system starts becoming a learning platform.

New File: agents/learning_agent.py

import boto3
import os
import json
from pathlib import Path

BEDROCK_MODEL_ID = os.getenv(
    "BEDROCK_MODEL_ID",
    "us.amazon.nova-lite-v1:0",
)
bedrock_runtime = boto3.client(
    "bedrock-runtime",
    region_name=os.getenv("AWS_REGION", "us-east-1"),
)
def generate_runbook_draft(learning_record: dict) -> str:
    prompt = f"""
You are a DevOps runbook writer.
Create a clear Kubernetes runbook from this resolved incident.
Include:
1. Symptoms
2. Root Cause
3. Actual Fix
4. Verification
5. Prevention
Learning Record:
{json.dumps(learning_record, indent=2)}
"""
    response = bedrock_runtime.converse(
        modelId=BEDROCK_MODEL_ID,
        messages=[
            {
                "role": "user",
                "content": [{"text": prompt}],
            }
        ],
    )

    return response["output"]["message"]["content"][0]["text"]

def save_runbook_draft(application: str, content: str) -> str:
    path = Path(f"learning/drafts/{application}-runbook.md")
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(content)

    return str(path)

Example Generated Runbook

# Runbook: TLS Trust Failure

## Symptoms

- Application fails to connect to internal HTTPS endpoint
- Logs show certificate verification failure
- Pod enters CrashLoopBackOff

## Root Cause

The application did not trust the internal CA certificate used by the upstream service.

## Actual Fix

Mounted the internal CA certificate and configured the JVM truststore path using JAVA_OPTS.

## Verification

- Pod reached Running state
- TLS handshake succeeded
- No further certificate errors in logs

## Prevention

- Include internal CA bundle in base image
- Validate truststore during CI
- Add synthetic TLS connectivity checks

Outcome

Human troubleshooting is now converted into reusable documentation.

This is one of the most valuable parts of the platform.

Challenge 19: Promote Knowledge Only After Human Review

The AI can generate a runbook draft, but it should not automatically become trusted knowledge.

A human should review it first.

New File: learning/runbook_promoter.py

from pathlib import Path
import shutil

def promote_runbook(draft_path: str) -> str:
    draft = Path(draft_path)
    promoted_dir = Path("learning/promoted")
    promoted_dir.mkdir(parents=True, exist_ok=True)
    promoted_path = promoted_dir / draft.name
    shutil.copyfile(
        draft,
        promoted_path,
    )
    return str(promoted_path)

Future Knowledge Base Sync

After promotion, the runbook can be uploaded to the S3 source bucket used by Bedrock Knowledge Base.

Future flow:

Draft runbook
        ↓
Human review
        ↓
Promoted runbook
        ↓
Upload to S3
        ↓
Bedrock Knowledge Base sync
        ↓
Future incidents use the new knowledge

Draft runbook

# Permission Denied Writing to /tmp/app.log Runbook

## Symptoms

- The pod `permission-app-5b988455f7-z6mh7` in the `dev` namespace is in a failing state.
- The failure reason is explicitly stated as "ERROR: permission denied writing to /tmp/app.log".

## Evidence

### Kubernetes Findings

- The pod `permission-app-5b988455f7-z6mh7` in the `dev` namespace is in a failing state.
- The failure reason is "ERROR: permission denied writing to /tmp/app.log".

### Log Findings

- The current logs show an error message: "ERROR: permission denied writing to /tmp/app.log".

## Common Root Causes

- The application does not have the necessary write permissions to the `/tmp/app.log` directory.
- Kubernetes Pod Security Policies (PSP) or Security Contexts might be restricting write permissions for the pod.
- The application's configuration may be set to write logs to a directory where it does not have write permissions.

## Safe Remediation

1. **Check and Adjust File System Permissions**

   - Execute the following command to check the permissions of the `/tmp` directory:
     ```sh
     kubectl exec -it permission-app-5b988455f7-z6mh7 -- ls -l /tmp

• If the permissions are incorrect, modify them using:

  kubectl exec -it permission-app-5b988455f7-z6mh7 -- chmod 755 /tmp

• Ensure that the application has write permissions to /tmp/app.log. You might need to create the file with appropriate permissions if it does not exist:

  kubectl exec -it permission-app-5b988455f7-z6mh7 -- touch /tmp/app.log && kubectl exec -it permission-app-5b988455f7-z6mh7 -- chmod 664 /tmp/app.log

2. Verify Pod Security Policies

   • Check if there are any Kubernetes Pod Security Policies (PSP) or Security Contexts restricting write permissions for the pod:

     kubectl get psp
     kubectl describe pod permission-app-5b988455f7-z6mh7

   • Ensure the application's security context allows write access to /tmp.

3. Inspect Application Configuration

   • Review the application's configuration to ensure it is set to write logs to a directory where it has write permissions.

Verification

• After making the necessary changes, verify that the application can write to /tmp/app.log without any permission issues.
• Check the pod status and ensure it is running correctly:

  kubectl get pod permission-app-5b988455f7-z6mh7

• Review the application logs to ensure the error message "ERROR: permission denied writing to /tmp/app.log" is resolved:

  kubectl logs permission-app-5b988455f7-z6mh7

Notes

• Ensure to adjust the file system permissions and security policies cautiously to avoid any potential security risks.
• If the issue persists after following these steps, consider consulting with the application development team for further assistance.

Final Flow After Part 2

By this stage, the workflow looks like this:

Engineer asks from Slack
        ↓
main_api.py receives event
        ↓
orchestrator.py starts investigation
        ↓
Kubernetes evidence is collected
        ↓
query_agent.py creates KB query
        ↓
knowledge_agent.py searches Bedrock KB
        ↓
rca_agent.py generates RCA
        ↓
issue_classifier_agent.py classifies issue
        ↓
remediation catalog is checked
        ↓
deployment inspector collects deployment details
        ↓
confidence_agent.py calculates confidence
        ↓
policy_agent.py applies environment policy
        ↓
Slack card is sent
        ↓
Engineer approves or rejects
        ↓
command_executor.py runs approved command
        ↓
Verification confirms recovery
        ↓
execution_store.py saves result
        ↓
agent_tracer.py saves RCA trace
        ↓
learning_store.py can record actual fix
        ↓
learning_agent.py creates draft runbook
        ↓
runbook_promoter.py promotes reviewed knowledge

Production View

For local testing, JSON files and local folders are enough.

For production, I would replace them with managed services.

Local JSON incident store      → DynamoDB
Local execution history        → DynamoDB or Aurora
Local RCA traces               → S3 + OpenSearch
Generated runbooks             → Confluence + S3
Remediation catalog            → Git repository with PR review
Slack tokens in .env           → AWS Secrets Manager
Local scanner                  → Prometheus / Grafana / CloudWatch webhook
Local API                      → ECS / EKS behind ALB or API Gateway

The production design should look like this:

Prometheus / Grafana / CloudWatch
        ↓
Alert Webhook
        ↓
AI DevOps Agent API
        ↓
orchestrator.py
        ↓
Kubernetes Evidence
        ↓
Bedrock Knowledge Base
        ↓
RCA
        ↓
Confidence
        ↓
Remediation Catalog
        ↓
Slack / Jira Approval
        ↓
Execution
        ↓
Verification
        ↓
Incident Store
        ↓
Learning Workflow

Security and Governance Lessons

A safe AI DevOps Agent should follow these rules:

Use read-only access by default
Do not execute raw LLM-generated commands
Use a remediation catalog
Require human approval
Block automated remediation in production
Verify after execution
Store audit history
Mask secrets before sending logs to the LLM
Review runbooks before promotion
Use Git PRs for remediation catalog changes
Use existing monitoring tools for production alert detection

The most important rule is:

AI can investigate.
AI can summarize.
AI can recommend.
But execution must be controlled.

Final Thoughts

Part 1 proved that the agent can investigate Kubernetes issues locally.

Part 2 evolved that foundation into a more realistic operational workflow.

At this stage, the agent can:

Receive investigation requests from Slack
Collect Kubernetes evidence
Search Amazon Bedrock Knowledge Base
Generate RCA
Classify the issue
Check remediation catalog
Calculate confidence
Ask for human approval
Execute approved remediation
Verify recovery
Record actual human fixes
Generate draft runbooks
Promote reviewed knowledge
Apply Dev / UAT / Prod policy

The biggest learning was:

AI should investigate with evidence,
recommend with confidence,
execute only with approval,
and learn only after human review.

This balance is important.

If we give AI too much control, it becomes risky.

If we give it no operational context, it becomes just another chatbot.

The useful middle ground is:

AI for investigation
AI for summarization
AI for runbook search
AI for recommendation
Human for approval
Automation for verified execution
Human review for learning

The current local scanner is useful for the home lab because it simulates alert-driven investigation.

But the scanner is not the final monitoring strategy.

The real value is the investigation engine behind it.

In Part 3, the next focus will be to connect this investigation engine with real observability and incident workflows.

Planned next improvements include:

Prometheus and Grafana Alertmanager webhook integration
CloudWatch alert integration
Alert parser and routing layer
Autonomous investigation from incoming alerts
Incident deduplication and state tracking
Jira-based incident workflow
Confluence or S3-based knowledge lifecycle
Bedrock Knowledge Base sync after approved learning
Multi-cluster and multi-environment support
Change risk and policy decision engine
Operational dashboards for incidents, confidence, and remediation history

The idea is simple:

Monitoring tools detect the symptom.
The AI DevOps Agent investigates the reason.
Engineers approve the action.
The platform verifies and learns.

So Part 3 will move the project closer to a real AIOps workflow by introducing alert ingestion, observability integration, incident lifecycle management, and stronger learning loops.

The journey started as a local RCA assistant.

Now it is becoming the foundation of a human-controlled, continuously learning AI DevOps platform.