FAQ RAG MCP Server

import os import logging from typing import Dict from mcp.server.fastmcp import FastMCP from dotenv import load_dotenv from rag_core import ask_faq_core, FAQ_DIR from pathlib import Path load_dotenv() # Configure logging logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s") logger = logging.getLogger(__name__) # Fail fast on missing key so errors are crisp if not os.getenv("OPENAI_API_KEY"): logger.error("OPENAI_API_KEY is not set") raise RuntimeError("OPENAI_API_KEY is not set") mcp = FastMCP("faq-rag") @mcp.tool() def ask_faq(question: str, top_k: int = 4) -> Dict[str, object]: """Answer a question from the FAQ corpus and cite at least two files.""" try: q = (question or "").strip() if not q: raise ValueError("`question` is required") if top_k <= 0 or top_k > 10: top_k = 4 logger.info(f"Tool 'ask_faq' called with question: {q}") return ask_faq_core(q, top_k=top_k) except Exception as e: logger.error(f"Error in 'ask_faq' tool: {e}") return {"error": str(e)} @mcp.resource("faq://{filename}") def get_faq_content(filename: str) -> str: """Read specific FAQ file content.""" try: logger.info(f"Reading resource: faq::{filename}") # Securely join path to prevent traversal base_path = Path(FAQ_DIR).resolve() file_path = (base_path / filename).resolve() if not file_path.is_relative_to(base_path): raise ValueError("Access denied: Invalid file path") if not file_path.exists(): raise FileNotFoundError(f"File not found: {filename}") return file_path.read_text() except Exception as e: logger.error(f"Error reading resource {filename}: {e}") raise @mcp.prompt() def ask_faq_expert(question: str) -> str: """Prepare a prompt for an FAQ expert.""" return f"""You are an expert on the company FAQs. Please answer the following question accurately using the ask_faq tool: Question: {question} """ if __name__ == "__main__": # STDIO transport (client launches this as a subprocess) try: logger.info("Starting MCP server...") mcp.run(transport="stdio") except Exception as e: logger.critical(f"Server crashed: {e}") raise