#!/usr/bin/env python3 """ HTTP server for MCP codebase RAG with REST API """ import uvicorn import logging from fastapi import FastAPI, HTTPException from fastapi.responses import JSONResponse from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel from typing import Optional import signal import sys import json from enhanced_toon import EnhancedToon from mcp_codebase import ( startup, search_codebase as _search_codebase, find_code_references as _find_code_references, read_file_lines, rebuild_index as _rebuild_index ) logger = logging.getLogger("rag-mcp") def signal_handler(sig, frame): """Handle graceful shutdown on Ctrl+C.""" logger.info("\n=== HTTP Server Shutdown ===") logger.info("Goodbye!\n") sys.exit(0) # Request/Response models class SearchRequest(BaseModel): query: str top_k: int = 5 rerank: bool = True class ReferenceRequest(BaseModel): symbol: str top_k: int = 20 class ReadFileRequest(BaseModel): path: str start: int = 1 end: Optional[int] = None class SymbolRequest(BaseModel): symbol: str def create_app(): """Create FastAPI app with MCP tool endpoints""" app = FastAPI( title="MCP Codebase RAG Server", description="Codebase search and analysis via MCP tools over HTTP", version="1.0.0" ) # Add CORS middleware for remote access app.add_middleware( CORSMiddleware, allow_origins=["*"], # Configure this appropriately for production allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Health check endpoint @app.get("/") async def root(): return { "status": "running", "service": "MCP Codebase RAG Server", "version": "1.0.0", "endpoints": { "docs": "/docs", "health": "/health", "tools": "/tools", "search": "POST /search", "references": "POST /references", "read_file": "POST /read_file", "rebuild": "POST /rebuild" } } @app.get("/health") async def health(): """Get server health and statistics""" try: # Import here to avoid circular import issues from mcp_codebase import health_check result = health_check() # Parse JSON string to dict for better API response try: result_dict = json.loads(result) return result_dict except: return {"result": result} except Exception as e: logger.exception("Health check failed") raise HTTPException(status_code=500, detail=str(e)) @app.get("/tools") async def list_tools(): """List available MCP tools""" return { "tools": [ { "name": "search_codebase", "description": "Search the entire codebase using hybrid RAG", "endpoint": "POST /search", "parameters": { "query": "string (required)", "top_k": "int (default: 5)", "rerank": "bool (default: true)" } }, { "name": "find_code_references", "description": "Find all references to a specific symbol", "endpoint": "POST /references", "parameters": { "symbol": "string (required)", "top_k": "int (default: 20)" } }, { "name": "read_file_lines", "description": "Read specific lines from a file with context", "endpoint": "POST /read_file", "parameters": { "path": "string (required)", "start": "int (default: 1)", "end": "int (optional)" } }, { "name": "rebuild_index", "description": "Force rebuild of search indexes", "endpoint": "POST /rebuild", "parameters": {} } ] } @app.post("/search") async def search(request: SearchRequest): """Search codebase - returns enhanced Toon format""" try: result = _search_codebase( query=request.query, top_k=request.top_k, rerank=request.rerank ) enhanced_result = EnhancedToon.search_results(result, request.query) return { "result": enhanced_result, "query": request.query, "top_k": request.top_k } except Exception as e: logger.exception("Search failed") raise HTTPException(status_code=500, detail=str(e)) @app.post("/references") async def references(request: ReferenceRequest): """Find all references to a symbol - returns enhanced Toon format""" try: result = _find_code_references( symbol=request.symbol, top_k=request.top_k ) enhanced_result = EnhancedToon.reference_results(result, request.symbol) return { "result": enhanced_result, "symbol": request.symbol, "top_k": request.top_k } except Exception as e: logger.exception("Reference search failed") raise HTTPException(status_code=500, detail=str(e)) @app.post("/read_file") async def read_file(request: ReadFileRequest): """Read file lines with context - returns enhanced Toon format""" try: result = read_file_lines( path=request.path, start=request.start, end=request.end ) enhanced_result = EnhancedToon.file_content_results(result, request.path) return { "result": enhanced_result, "path": request.path, "start_line": request.start, "end_line": request.end } except Exception as e: logger.exception("Read file failed") raise HTTPException(status_code=500, detail=str(e)) @app.post("/rebuild") async def rebuild(): """Rebuild search indexes""" try: result = _rebuild_index() # For rebuild, we might not need enhanced format as it's usually a simple status message return {"result": result} except Exception as e: logger.exception("Rebuild failed") raise HTTPException(status_code=500, detail=str(e)) return app if __name__ == "__main__": # Register signal handlers signal.signal(signal.SIGINT, signal_handler) signal.signal(signal.SIGTERM, signal_handler) try: # Initialize the indexes (this will load or build them) startup() logger.info("="*60) logger.info("🚀 Starting HTTP server on: http://0.0.0.0:8000") logger.info("📚 REST API endpoints:") logger.info(" GET / - Service info") logger.info(" GET /health - Server health & stats") logger.info(" GET /tools - List available tools") logger.info(" POST /search - Search codebase") logger.info(" POST /references - Find symbol references") logger.info(" POST /read_file - Read file with context") logger.info(" POST /rebuild - Rebuild indexes") logger.info("📖 API docs available at: http://0.0.0.0:8000/docs") logger.info("⏚ī¸ Press Ctrl+C to stop") logger.info("="*60) # Create and run app app = create_app() uvicorn.run( app, host="0.0.0.0", port=8000, log_level="info", access_log=True ) except KeyboardInterrupt: signal_handler(signal.SIGINT, None) except Exception as e: logger.exception("Fatal error during HTTP server startup") sys.exit(1)