SlackBolt is Underrated: Building an AI Knowledge Bot for Your Team

# slack-bolt.py
import os
import logging
from slack_bolt import App
from slack_bolt.adapter.socket_mode import SocketModeHandler

# Initialize your app with your bot token and socket mode handler
app = App(token=os.environ.get("SLACK_BOT_TOKEN"))

@app.command("/ai")
def handle_ai_command(ack, respond, command):
    ack()
    user_query = command['text']
    thread_ts = command.get('thread_ts', command['ts'])
    
    # Process the query (we'll implement this next)
    response = generate_response(user_query, thread_ts)
    respond(response, thread_ts=thread_ts)

if __name__ == "__main__":
    handler = SocketModeHandler(app, os.environ["SLACK_APP_TOKEN"])
    handler.start()

# bot.py
import openai
from typing import Dict, Any

# Define your categories
CATEGORIES = {
    "general": {
        "prompt_file": "prompts/general.txt",
        "kb_file": None,
        "description": "For general company questions and policies",
        "aliases": ["general", "company", "policy", "process"],
        "context_prefix": "**Answering a general company question.**"
    },
    "technical": {
        "prompt_file": "prompts/technical.txt",
        "kb_file": "knowledge_bases/tech_kb.txt",
        "description": "For technical documentation and development questions",
        "aliases": ["tech", "development", "api", "code", "deployment"],
        "context_prefix": "**Answering a technical question. Here is relevant context:**"
    }
}

async def get_query_category(user_message: str) -> str:
    """Route the user query to the appropriate category"""
    
    # Build router prompt dynamically
    router_prompt = """
You are a query router. Classify the user's question into one of the following categories and respond with ONLY the category name:
"""
    
    for category, config in CATEGORIES.items():
        router_prompt += f"\n- '{category}': {config['description']}"
    
    response = await openai.AsyncOpenAI().chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": router_prompt},
            {"role": "user", "content": user_message}
        ],
        temperature=0,
    )
    
    category = response.choices[0].message.content.strip().lower()
    return category if category in CATEGORIES else "general"

pip install mcp-agent

# mcp_integration.py
from mcp_agent import MCPApp, Agent
from mcp_agent.models import Settings, MCPSettings, MCPServerSettings, OpenAISettings

def setup_mcp_agent():
    """Configure MCP agent with Confluence tools"""
    
    # Configure Atlassian/Confluence server
    atlassian_args = [
        "mcp-atlassian",
        f"--confluence-url={os.environ.get('CONFLUENCE_URL')}",
        f"--confluence-username={os.environ.get('CONFLUENCE_USER')}",
        f"--confluence-api-token={os.environ.get('CONFLUENCE_TOKEN')}"
    ]
    
    settings = Settings(
        execution_engine="asyncio",
        mcp=MCPSettings(
            servers={
                "atlassian": MCPServerSettings(
                    command="uvx",
                    args=atlassian_args
                )
            }
        ),
        openai=OpenAISettings(
            api_key=os.environ.get("OPENAI_API_KEY"),
            default_model="gpt-4o"
        ),
    )
    
    return MCPApp(name="knowledge_bot", settings=settings)

# memory.py
from typing import Dict, List, NamedTuple
from datetime import datetime, timedelta
import threading

class ConversationMessage(NamedTuple):
    role: str
    content: str
    timestamp: datetime

class ConversationMemory:
    def __init__(self, max_messages: int = 10, ttl_hours: int = 24):
        self.conversations: Dict[str, List[ConversationMessage]] = {}
        self.max_messages = max_messages
        self.ttl_hours = ttl_hours
        self.lock = threading.Lock()
    
    def add_message(self, thread_id: str, role: str, content: str):
        """Add a message to the conversation history"""
        with self.lock:
            if thread_id not in self.conversations:
                self.conversations[thread_id] = []
            
            message = ConversationMessage(
                role=role,
                content=content,
                timestamp=datetime.now()
            )
            
            self.conversations[thread_id].append(message)
            
            # Keep only the most recent messages
            if len(self.conversations[thread_id]) > self.max_messages:
                self.conversations[thread_id] = self.conversations[thread_id][-self.max_messages:]
    
    def get_conversation_history(self, thread_id: str) -> str:
        """Get formatted conversation history"""
        if thread_id not in self.conversations:
            return ""
        
        history = []
        for msg in self.conversations[thread_id]:
            history.append(f"{msg.role}: {msg.content}")
        
        return "\n".join(history)

# Global memory instance
memory = ConversationMemory()

# response_generator.py
async def generate_response(message: str, thread_id: str) -> str:
    """Generate AI response to user query"""
    
    # Get conversation history
    conversation_history = memory.get_conversation_history(thread_id)
    
    # Route the query
    category = await get_query_category(message)
    config = CATEGORIES[category]
    
    # Load prompt and knowledge base
    prompt = load_prompt(config["prompt_file"])
    kb_content = load_knowledge_base(config["kb_file"]) if config["kb_file"] else ""
    
    # Build final message with context
    final_message = f"""
{config['context_prefix']}

{kb_content}

Conversation History:
{conversation_history}

Current Question: {message}
"""
    
    # Execute with MCP agent
    mcp_app = setup_mcp_agent()
    async with mcp_app:
        finder_agent = Agent(
            name="knowledge_finder",
            instruction=prompt,
            server_names=["atlassian"]
        )
        
        async with finder_agent:
            from mcp_agent.models import OpenAIAugmentedLLM
            llm = await finder_agent.attach_llm(OpenAIAugmentedLLM)
            result = await llm.generate_str(message=final_message)
    
    # Store the interaction in memory
    memory.add_message(thread_id, "User", message)
    memory.add_message(thread_id, "Assistant", result)
    
    return result

def load_prompt(filename: str) -> str:
    """Load prompt from file"""
    try:
        with open(filename, 'r') as f:
            return f.read()
    except FileNotFoundError:
        return "You are a helpful assistant that answers questions based on available knowledge."

def load_knowledge_base(filename: str) -> str:
    """Load knowledge base content"""
    if not filename:
        return ""
    
    try:
        with open(filename, 'r') as f:
            return f.read()
    except FileNotFoundError:
        return ""

"sales": {
    "prompt_file": "prompts/sales.txt",
    "kb_file": "knowledge_bases/sales_kb.txt",
    "description": "For sales process and customer questions",
    "aliases": ["sales", "customer", "pricing", "demo"],
    "context_prefix": "**Answering a sales question. Here is relevant context:**"
}