Build an AI DevOps Onboarding Assistant with Claude API

pip install anthropic fastapi uvicorn numpy scikit-learn python-dotenv

export ANTHROPIC_API_KEY=sk-ant-...

# retriever.py
import os
import numpy as np
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
 
class DocumentRetriever:
    def __init__(self, docs_dir: str):
        self.docs = []
        self.doc_names = []
        self._load_docs(docs_dir)
        self._build_index()
 
    def _load_docs(self, docs_dir: str):
        for filename in os.listdir(docs_dir):
            if filename.endswith(".txt"):
                path = os.path.join(docs_dir, filename)
                with open(path, "r") as f:
                    content = f.read()
                self.docs.append(content)
                self.doc_names.append(filename)
        print(f"Loaded {len(self.docs)} documents")
 
    def _build_index(self):
        self.vectorizer = TfidfVectorizer(stop_words="english")
        self.tfidf_matrix = self.vectorizer.fit_transform(self.docs)
 
    def retrieve(self, query: str, top_k: int = 3) -> list[dict]:
        query_vec = self.vectorizer.transform([query])
        scores = cosine_similarity(query_vec, self.tfidf_matrix)[0]
        top_indices = np.argsort(scores)[::-1][:top_k]
 
        results = []
        for idx in top_indices:
            if scores[idx] > 0.05:  # filter out irrelevant docs
                results.append({
                    "source": self.doc_names[idx],
                    "content": self.docs[idx],
                    "score": float(scores[idx])
                })
        return results

# main.py
import os
import anthropic
from fastapi import FastAPI
from pydantic import BaseModel
from retriever import DocumentRetriever
from dotenv import load_dotenv
 
load_dotenv()
 
app = FastAPI(title="DevOps Onboarding Assistant")
client = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
retriever = DocumentRetriever(docs_dir="docs/")
 
class QuestionRequest(BaseModel):
    question: str
 
class AnswerResponse(BaseModel):
    answer: str
    sources: list[str]
 
@app.post("/ask", response_model=AnswerResponse)
async def ask_question(request: QuestionRequest):
    # Retrieve relevant docs
    relevant_docs = retriever.retrieve(request.question, top_k=3)
 
    if not relevant_docs:
        return AnswerResponse(
            answer="I couldn't find relevant documentation for your question. Try asking in #platform-team.",
            sources=[]
        )
 
    # Build context from retrieved docs
    context_parts = []
    sources = []
    for doc in relevant_docs:
        context_parts.append(f"[{doc['source']}]\n{doc['content']}")
        sources.append(doc['source'])
 
    context = "\n\n---\n\n".join(context_parts)
 
    # Call Claude
    message = client.messages.create(
        model="claude-haiku-4-5-20251001",
        max_tokens=1024,
        system="""You are a helpful DevOps onboarding assistant for our engineering team.
Answer questions based only on the provided documentation.
Be concise and practical. If the docs don't cover something, say so and suggest asking in Slack.
Format commands with backticks. Use numbered steps for procedures.""",
        messages=[
            {
                "role": "user",
                "content": f"Documentation:\n\n{context}\n\nQuestion: {request.question}"
            }
        ]
    )
 
    return AnswerResponse(
        answer=message.content[0].text,
        sources=sources
    )
 
@app.get("/health")
async def health():
    return {"status": "ok", "docs_loaded": len(retriever.docs)}

uvicorn main:main --reload --port 8000

curl -X POST http://localhost:8000/ask \
  -H "Content-Type: application/json" \
  -d '{"question": "How do I deploy to staging?"}'

{
  "answer": "To deploy to staging:\n\n1. Build and push your Docker image to ECR using the `devops` AWS profile...",
  "sources": ["deploy-staging.txt"]
}

pip install slack-bolt

# slack_bot.py
import os
import requests
from slack_bolt import App
from slack_bolt.adapter.socket_mode import SocketModeHandler
 
app = App(token=os.environ["SLACK_BOT_TOKEN"])
API_URL = "http://localhost:8000/ask"
 
@app.event("app_mention")
def handle_mention(event, say):
    question = event["text"].replace(f"<@{event['user']}>", "").strip()
 
    response = requests.post(API_URL, json={"question": question})
    data = response.json()
 
    sources_text = ", ".join(data["sources"]) if data["sources"] else "no docs matched"
    say(f"{data['answer']}\n\n_Sources: {sources_text}_")
 
if __name__ == "__main__":
    SocketModeHandler(app, os.environ["SLACK_APP_TOKEN"]).start()