How to Set Up AI-Powered Chatbots with Claude Code 3.5 and Slack API 2.0 for Internal Teams

anthropic==0.15.0
slack-sdk==3.22.0
python-socket-mode==3.2.1
python-dotenv==1.0.0
fastapi==0.104.1
uvicorn==0.24.0
pinecone-client==2.2.1
cohere==4.34
presidio-analyzer==2.2.0
presidio-anonymizer==2.2.0
redis==5.0.0
pytest==7.4.0

import os
import logging
from slack_sdk.web import WebClient
from slack_sdk.socket_mode import SocketModeClient
from slack_sdk.socket_mode.response import SocketModeResponse
from slack_sdk.socket_mode.request import SocketModeRequest
from dotenv import load_dotenv
import re

# Configure logging for production debugging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

load_dotenv()

class SlackBotClient:
    def __init__(self):
        # Initialize Slack Web Client for posting messages
        self.web_client = WebClient(token=os.getenv("SLACK_BOT_TOKEN"))
        # Initialize Socket Mode Client for real-time event listening (Slack API 2.0)
        self.socket_client = SocketModeClient(
            app_token=os.getenv("SLACK_APP_TOKEN"),
            web_client=self.web_client
        )
        # Regex to extract mentions and query text from Slack messages
        self.mention_regex = re.compile(r"<@(|[WU].+?)>(.*)", re.IGNORECASE)
        self._register_handlers()

    def _register_handlers(self):
        """Register event handlers for Slack API 2.0 socket events"""
        self.socket_client.socket_mode_request_listeners.append(self._handle_request)

    def _handle_request(self, client: SocketModeClient, req: SocketModeRequest):
        """Process incoming Slack events with error handling"""
        try:
            # Only process message events that mention the bot
            if req.type == "events_api" and req.payload.get("event", {}).get("type") == "message":
                event = req.payload["event"]
                # Ignore messages from the bot itself to prevent loops
                if event.get("bot_id"):
                    return
                text = event.get("text", "")
                match = self.mention_regex.match(text)
                if not match:
                    return
                query = match.group(2).strip()
                channel = event["channel"]
                thread_ts = event.get("thread_ts", event["ts"])
                # Acknowledge receipt to Slack within 3 seconds to avoid timeout
                response = SocketModeResponse(envelope_id=req.envelope_id)
                client.send_socket_mode_response(response)
                # Process query asynchronously (in production, use Celery or FastAPI background tasks)
                self._process_query(query, channel, thread_ts)
        except Exception as e:
            logger.error(f"Failed to process Slack request: {str(e)}", exc_info=True)
            # Send error response to Slack if possible
            if req.envelope_id:
                error_response = SocketModeResponse(envelope_id=req.envelope_id)
                client.send_socket_mode_response(error_response)

    def _process_query(self, query: str, channel: str, thread_ts: str):
        """Placeholder for query processing - integrated with Claude client in later steps"""
        logger.info(f"Processing query: {query} in channel {channel}")
        # Will be replaced with Claude 3.5 integration in Step 3
        self.web_client.chat_postMessage(
            channel=channel,
            thread_ts=thread_ts,
            text=f"Processing your query: {query}"
        )

    def start(self):
        """Start listening for Slack events"""
        logger.info("Starting Slack Socket Mode client...")
        self.socket_client.connect()
        # Keep the main thread alive
        import time
        while True:
            time.sleep(1)

if __name__ == "__main__":
    # Validate required environment variables
    required_vars = ["SLACK_BOT_TOKEN", "SLACK_APP_TOKEN"]
    missing = [var for var in required_vars if not os.getenv(var)]
    if missing:
        raise ValueError(f"Missing required environment variables: {missing}")
    bot = SlackBotClient()
    bot.start()

import os
import logging
from anthropic import Anthropic, HUMAN_PROMPT, AI_PROMPT
from anthropic.types import Message
from dotenv import load_dotenv
from typing import List, Dict, Optional

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

load_dotenv()

class ClaudeClient:
    def __init__(self):
        # Initialize Anthropic client with Claude 3.5 Sonnet model
        self.client = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
        self.model = "claude-3-5-sonnet-20240620"  # Claude 3.5 Sonnet GA version
        # System prompt to enforce internal governance and context
        self.system_prompt = """You are an internal engineering support chatbot for our team. 
        Only answer questions related to our internal tooling, CI/CD pipelines, AWS infrastructure, and internal documentation.
        Do not share PII, internal IP addresses, or confidential project details.
        If you don't know the answer, say "I don't have enough context to answer that. Please check our internal wiki or tag a human."
        Always cite the source of your answer if it's from internal docs."""
        self.max_tokens = 1024
        self.temperature = 0.1  # Low temperature for factual engineering responses

    def generate_response(self, query: str, context: Optional[List[Dict]] = None) -> str:
        """
        Generate a response from Claude 3.5 with optional context from vector DB.

        Args:
            query: User's Slack query
            context: List of relevant doc chunks from Pinecone, each with 'text' and 'source' keys
        """
        try:
            # Build messages list with context if available
            messages = []
            if context:
                context_str = "\n\n".join([f"Source: {c['source']}\n{c['text']}" for c in context])
                messages.append({"role": "user", "content": f"Context:\n{context_str}\n\nQuery: {query}"})
            else:
                messages.append({"role": "user", "content": query})

            # Call Claude 3.5 API with error handling for rate limits
            response: Message = self.client.messages.create(
                model=self.model,
                max_tokens=self.max_tokens,
                temperature=self.temperature,
                system=self.system_prompt,
                messages=messages
            )
            # Extract text from response, handle empty responses
            if response.content and len(response.content) > 0:
                return response.content[0].text
            logger.warning(f"Claude returned empty response for query: {query}")
            return "I didn't get a valid response. Please try rephrasing your query."
        except Exception as e:
            logger.error(f"Claude API call failed: {str(e)}", exc_info=True)
            # Handle rate limit errors specifically
            if "rate_limit" in str(e).lower():
                return "I'm currently rate-limited. Please wait 30 seconds and try again."
            return "Sorry, I encountered an error processing your request. Tag a human if this persists."

    def get_token_count(self, text: str) -> int:
        """Utility to count tokens for context window management"""
        return self.client.count_tokens(text)

if __name__ == "__main__":
    # Test Claude client with a sample query
    required_vars = ["ANTHROPIC_API_KEY"]
    missing = [var for var in required_vars if not os.getenv(var)]
    if missing:
        raise ValueError(f"Missing required environment variables: {missing}")
    client = ClaudeClient()
    test_query = "How do I rotate AWS IAM keys for the production EKS cluster?"
    response = client.generate_response(test_query)
    print(f"Test response: {response}")

import os
import logging
from pinecone import Pinecone, Index
from cohere import Client as CohereClient
from dotenv import load_dotenv
from typing import List, Dict

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

load_dotenv()

class VectorDBClient:
    def __init__(self):
        # Initialize Pinecone
        self.pc = Pinecone(api_key=os.getenv("PINECONE_API_KEY"))
        self.index_name = os.getenv("PINECONE_INDEX_NAME", "internal-docs")
        # Create index if it doesn't exist (1024 dimensions for Cohere embed-english-v3-0)
        if self.index_name not in self.pc.list_indexes().names():
            self.pc.create_index(
                name=self.index_name,
                dimension=1024,
                metric="cosine"
            )
        self.index: Index = self.pc.Index(self.index_name)
        # Initialize Cohere for embeddings
        self.cohere = CohereClient(api_key=os.getenv("COHERE_API_KEY"))
        self.embedding_model = "embed-english-v3-0"

    def chunk_document(self, text: str, chunk_size: int = 512, overlap: int = 64) -> List[str]:
        """Split long documents into overlapping chunks optimized for Slack queries"""
        chunks = []
        for i in range(0, len(text), chunk_size - overlap):
            chunk = text[i:i + chunk_size]
            chunks.append(chunk)
        return chunks

    def index_document(self, doc_id: str, text: str, source: str):
        """Index a document with metadata"""
        chunks = self.chunk_document(text)
        embeddings = self.cohere.embed(
            texts=chunks,
            model=self.embedding_model,
            input_type="search_document"
        ).embeddings
        vectors = []
        for i, (chunk, embedding) in enumerate(zip(chunks, embeddings)):
            vectors.append({
                "id": f"{doc_id}-{i}",
                "values": embedding,
                "metadata": {"text": chunk, "source": source}
            })
        # Upsert in batches of 100 to avoid rate limits
        batch_size = 100
        for i in range(0, len(vectors), batch_size):
            batch = vectors[i:i + batch_size]
            self.index.upsert(vectors=batch)
        logger.info(f"Indexed {len(chunks)} chunks for document {doc_id}")

    def query_context(self, query: str, top_k: int = 3) -> List[Dict]:
        """Retrieve top k relevant chunks for a query"""
        try:
            query_embedding = self.cohere.embed(
                texts=[query],
                model=self.embedding_model,
                input_type="search_query"
            ).embeddings[0]
            results = self.index.query(
                vector=query_embedding,
                top_k=top_k,
                include_metadata=True
            )
            return [{"text": match["metadata"]["text"], "source": match["metadata"]["source"]} for match in results["matches"]]
        except Exception as e:
            logger.error(f"Vector DB query failed: {str(e)}", exc_info=True)
            return []

if __name__ == "__main__":
    required_vars = ["PINECONE_API_KEY", "COHERE_API_KEY"]
    missing = [var for var in required_vars if not os.getenv(var)]
    if missing:
        raise ValueError(f"Missing required environment variables: {missing}")
    # Test indexing a sample doc
    db = VectorDBClient()
    sample_doc = "To rotate AWS IAM keys for EKS: 1. Go to AWS Console → IAM → Users → [user] → Security Credentials → Create Access Key. 2. Update kubeconfig with new key. 3. Delete old key after 24h."
    db.index_document("iam-rotation-1", sample_doc, "Internal AWS Wiki")
    # Test query
    context = db.query_context("How do I rotate EKS IAM keys?")
    print(f"Retrieved context: {context}")

import os
import logging
from fastapi import FastAPI, BackgroundTasks
from dotenv import load_dotenv
from src.slack_client import SlackBotClient
from src.claude_client import ClaudeClient
from src.vector_db import VectorDBClient
import threading

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

load_dotenv()

app = FastAPI(title="Claude Slack Bot")
slack_client = SlackBotClient()
claude_client = ClaudeClient()
vector_db = VectorDBClient()

def process_query(query: str, channel: str, thread_ts: str):
    """Full query processing pipeline"""
    try:
        # Retrieve context from vector DB
        context = vector_db.query_context(query)
        # Generate response from Claude 3.5
        response = claude_client.generate_response(query, context)
        # Post response to Slack
        slack_client.web_client.chat_postMessage(
            channel=channel,
            thread_ts=thread_ts,
            text=response
        )
        logger.info(f"Posted response to {channel} for query: {query}")
    except Exception as e:
        logger.error(f"Failed to process query: {str(e)}", exc_info=True)
        slack_client.web_client.chat_postMessage(
            channel=channel,
            thread_ts=thread_ts,
            text="Sorry, I encountered an error. Please tag a human for help."
        )

# Override Slack client's query processing to use full pipeline
slack_client._process_query = lambda query, channel, thread_ts: process_query(query, channel, thread_ts)

@app.get("/health")
def health_check():
    return {"status": "healthy"}

if __name__ == "__main__":
    # Start Slack Socket Mode client in a separate thread
    slack_thread = threading.Thread(target=slack_client.start, daemon=True)
    slack_thread.start()
    # Start FastAPI server
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

import redis
import os

class RateLimiter:
    def __init__(self):
        self.redis = redis.Redis.from_url(os.getenv("REDIS_URL", "redis://localhost:6379"))
        self.max_per_user = 10  # queries per minute per user
        self.window_seconds = 60

    def is_rate_limited(self, user_id: str) -> bool:
        key = f"rate_limit:{user_id}"
        current = self.redis.incr(key)
        if current == 1:
            self.redis.expire(key, self.window_seconds)
        return current > self.max_per_user

self.system_prompt = """You are an internal engineering support chatbot. 
BANNED TOPICS: PII (names, emails, phone numbers), internal IP addresses (10.x.x.x, 172.16.x.x), AWS secret keys, project codenames (e.g., Project Phoenix).
If a query asks for banned content, respond: "I can't share that information. Please contact your team lead for access."
Always cite sources from internal docs. If no context is available, direct users to internal wiki."""

def chunk_document(self, text: str, chunk_size: int = 512, overlap: int = 64) -> List[str]:
    """Chunk size optimized for 12-word average Slack queries"""
    chunks = []
    words = text.split()
    current_chunk = []
    current_length = 0
    for word in words:
        current_chunk.append(word)
        current_length += 1
        if current_length >= chunk_size:
            chunks.append(" ".join(current_chunk))
            current_chunk = current_chunk[-overlap:]  # keep overlap words
            current_length = overlap
    if current_chunk:
        chunks.append(" ".join(current_chunk))
    return chunks

claude-slack-bot/
├── src/
│   ├── __init__.py
│   ├── slack_client.py       # Slack API 2.0 Socket Mode integration
│   ├── claude_client.py      # Anthropic Claude 3.5 SDK wrapper
│   ├── vector_db.py          # Pinecone indexing and query logic
│   ├── rate_limiter.py       # Redis-backed rate limiting
│   ├── pii_redactor.py       # Presidio PII detection
│   ├── config.py             # Environment variable management
│   └── main.py               # FastAPI entrypoint
├── tests/
│   ├── __init__.py
│   ├── test_slack.py
│   ├── test_claude.py
│   └── test_vector_db.py
├── docker/
│   ├── Dockerfile
│   └── docker-compose.yml
├── .env.example              # Environment variable template
├── requirements.txt          # Python dependencies
└── README.md                 # Setup and deployment instructions