Spec3 MCP Server

"""Spec3 MCP Server implementation using FastMCP.""" import logging from typing import Any import base64 from io import BytesIO import boto3 from fastmcp import FastMCP from pdf2image import convert_from_bytes from PIL import Image import pypdf # Configure logging logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" ) logger = logging.getLogger(__name__) # AWS Configuration AWS_REGION = "us-east-1" S3_BUCKET = "spec3-chatbot-data-091702001436-us-east-1" # Car-specific context CAR_CONTEXT = """ # My Spec3 E36 Build ## Vehicle Information - Year/Model: 1994 BMW E36 325is - Build Status: [In-progress] """ # Available documents mapping AVAILABLE_DOCS = { "spec3_constructor_guide": { "name": "Spec3 E36 Race Car Constructor's Guide", "s3_key": "Spec3 E36 Race Car Contsructor's Guide.pdf", "description": "Comprehensive guide for building a Spec3 E36 race car" }, "bentley_manual_general": { "name": "Bentley General Manual", "s3_key": "bentley_general.pdf", "description": "Bentley BMW E36 Manual - GENERAL SECTION" }, "nasa_ccr": { "name": "2025 NASA Competition Comp Rules (CCR)", "s3_key": "2025.4_NASACCR.pdf", "description": "2025 NASA Club Championship Racing rules" }, "spec3_rules": { "name": "2025 Spec3 Rules", "s3_key": "2025_Spec3_Rules.pdf", "description": "2025 Spec3 racing class specific rules and regulations" } } # Initialize AWS clients s3_client = boto3.client("s3", region_name=AWS_REGION) def create_server() -> FastMCP: """Create and configure the Spec3 MCP server.""" logger.info("Creating Spec3 MCP Server") # Initialize FastMCP server mcp = FastMCP("spec3-mcp-server") @mcp.tool() async def get_car_context() -> dict[str, Any]: """ Get information about the user's 1994 BMW E36 325is Spec3 race car build. Returns current configuration, build status, modifications, and car-specific details. Call this tool when providing personalized advice, troubleshooting, or planning modifications. Returns: dict: Car configuration, history, and current state """ logger.info("get_car_context called") return { "context": CAR_CONTEXT, "last_updated": "2025-10-05" } @mcp.tool() async def list_documents() -> dict[str, Any]: """ List all available Spec3 racing reference documents. Available documents include: Spec3 Constructor's Guide, Bentley E36 Manual, 2025 NASA CCR rules, and 2025 Spec3 class rules. Returns: dict: Document IDs, names, and descriptions """ logger.info("list_documents called") docs_list = [] for doc_id, doc_info in AVAILABLE_DOCS.items(): docs_list.append({ "id": doc_id, "name": doc_info["name"], "description": doc_info["description"] }) return { "documents": docs_list, "count": len(docs_list) } @mcp.tool() async def get_document( document_id: str, page_start: int = 1, page_end: int | None = None, include_images: bool = True ) -> dict[str, Any]: """ Retrieve full text and visual content of Spec3 racing reference documents. Fetches complete PDF content from S3 including text and page images. Page images preserve diagrams, tables, and formatting that text extraction cannot capture. Args: document_id: Document ID from list_documents (e.g., "spec3_rules") page_start: Starting page number (default: 1) page_end: Ending page number (default: None for all remaining pages) include_images: Include page images for diagrams/tables (default: True) Returns: dict: Document text, page images (base64), metadata, and page range """ logger.info(f"get_document called for: {document_id}, pages {page_start}-{page_end}, images={include_images}") if document_id not in AVAILABLE_DOCS: return { "error": f"Document ID '{document_id}' not found. Use list_documents to see available documents.", "available_ids": list(AVAILABLE_DOCS.keys()) } try: doc_info = AVAILABLE_DOCS[document_id] s3_key = doc_info["s3_key"] # Download PDF from S3 logger.info(f"Downloading {s3_key} from S3") response = s3_client.get_object(Bucket=S3_BUCKET, Key=s3_key) pdf_content = response['Body'].read() # Parse PDF for text pdf_file = BytesIO(pdf_content) pdf_reader = pypdf.PdfReader(pdf_file) total_pages = len(pdf_reader.pages) # Validate and adjust page range page_start = max(1, page_start) if page_end is None: page_end = total_pages else: page_end = min(page_end, total_pages) if page_start > total_pages: return { "error": f"page_start ({page_start}) exceeds total pages ({total_pages})", "total_pages": total_pages } # Extract text from specified pages text_content = [] for page_num in range(page_start - 1, page_end): page = pdf_reader.pages[page_num] page_text = page.extract_text() text_content.append(f"--- Page {page_num + 1} ---\n{page_text}") full_text = "\n\n".join(text_content) # Extract page images if requested page_images = [] if include_images: logger.info(f"Converting pages {page_start}-{page_end} to images") # Convert PDF pages to images images = convert_from_bytes( pdf_content, first_page=page_start, last_page=page_end, dpi=150 # Balance between quality and size ) for idx, img in enumerate(images): # Convert to base64 buffered = BytesIO() img.save(buffered, format="PNG", optimize=True) img_base64 = base64.b64encode(buffered.getvalue()).decode('utf-8') page_images.append({ "page_number": page_start + idx, "image": img_base64, "format": "png" }) result = { "document_name": doc_info["name"], "document_id": document_id, "total_pages": total_pages, "pages_retrieved": f"{page_start}-{page_end}", "text": full_text, "images": page_images, "num_images": len(page_images), "size_bytes": len(pdf_content) } logger.info(f"Successfully retrieved {page_end - page_start + 1} pages ({len(page_images)} images) from {doc_info['name']}") return result except Exception as e: logger.error(f"Error retrieving document: {str(e)}") return { "error": f"Error retrieving document: {str(e)}", "document_id": document_id } logger.info("Spec3 MCP Server created successfully") return mcp