| 1 | from typing import Optional |
| 2 | |
| 3 | from fastapi import APIRouter |
| 4 | from fastapi import Depends |
| 5 | from fastapi import HTTPException |
| 6 | from fastapi import Query |
| 7 | from fastapi import Security |
| 8 | from loguru import logger |
| 9 | from sqlalchemy.ext.asyncio import AsyncSession |
| 10 | |
| 11 | from app.auth.routes.auth import AuthHandler |
| 12 | from app.db.db_session import get_db |
| 13 | from app.integrations.copilot_mcp.schema.copilot_mcp import AvailableMCPServersResponse |
| 14 | from app.integrations.copilot_mcp.schema.copilot_mcp import ExampleQuestionsResponse |
| 15 | from app.integrations.copilot_mcp.schema.copilot_mcp import MCPQueryRequest |
| 16 | from app.integrations.copilot_mcp.schema.copilot_mcp import MCPQueryResponse |
| 17 | from app.integrations.copilot_mcp.schema.copilot_mcp import MCPServerType |
| 18 | from app.integrations.copilot_mcp.services.copilot_mcp import MCPService |
| 19 | from app.integrations.copilot_mcp.services.example_questions import ( |
| 20 | ExampleQuestionsService, |
| 21 | ) |
| 22 | from app.middleware.license import get_license |
| 23 | from app.middleware.license import is_feature_enabled |
| 24 | |
| 25 | copilot_mcp_router = APIRouter() |
| 26 | auth_handler = AuthHandler() |
| 27 | |
| 28 | |
| 29 | @copilot_mcp_router.get( |
| 30 | "/servers", |
| 31 | response_model=AvailableMCPServersResponse, |
| 32 | description="Get list of available MCP servers", |
| 33 | dependencies=[Security(AuthHandler().require_any_scope("admin", "analyst"))], |
| 34 | ) |
| 35 | async def get_available_mcp_servers() -> AvailableMCPServersResponse: |
| 36 | """ |
| 37 | Retrieve a list of all available MCP servers with their capabilities. |
| 38 | |
| 39 | This endpoint provides information about all MCP servers that can be |
| 40 | queried, including their descriptions and capabilities to help users |
| 41 | understand what each server can do. |
| 42 | |
| 43 | Returns: |
| 44 | AvailableMCPServersResponse: List of available MCP servers with metadata |
| 45 | """ |
| 46 | logger.info("Fetching available MCP servers") |
| 47 | |
| 48 | try: |
| 49 | servers = ExampleQuestionsService.get_available_servers() |
| 50 | |
| 51 | logger.info(f"Found {len(servers)} available MCP servers") |
| 52 | |
| 53 | return AvailableMCPServersResponse( |
| 54 | servers=servers, |
| 55 | total_servers=len(servers), |
| 56 | message="Successfully retrieved available MCP servers", |
| 57 | success=True, |
| 58 | ) |
| 59 | |
| 60 | except Exception as e: |
| 61 | logger.error(f"Error fetching available MCP servers: {str(e)}") |
| 62 | return AvailableMCPServersResponse(servers=[], total_servers=0, message=f"Error retrieving MCP servers: {str(e)}", success=False) |
| 63 | |
| 64 | |
| 65 | @copilot_mcp_router.get( |
| 66 | "/servers/{mcp_server}", |
| 67 | response_model=dict, |
| 68 | description="Get detailed information about a specific MCP server", |
| 69 | dependencies=[Security(AuthHandler().require_any_scope("admin", "analyst"))], |
| 70 | ) |
| 71 | async def get_mcp_server_details(mcp_server: MCPServerType) -> dict: |
| 72 | """ |
| 73 | Get detailed information about a specific MCP server. |
| 74 | |
| 75 | Args: |
| 76 | mcp_server: The MCP server to get details for |
| 77 | |
| 78 | Returns: |
| 79 | Dictionary containing detailed server information |
| 80 | """ |
| 81 | logger.info(f"Fetching details for MCP server: {mcp_server.value}") |
| 82 | |
| 83 | try: |
| 84 | server_info = ExampleQuestionsService.get_server_info(mcp_server) |
| 85 | |
| 86 | if not server_info: |
| 87 | return {"server": mcp_server.value, "message": f"Server information not found for {mcp_server.value}", "success": False} |
| 88 | |
| 89 | # Get additional context like available categories and question count |
| 90 | categories = ExampleQuestionsService.get_available_categories(mcp_server) |
| 91 | total_questions = len(ExampleQuestionsService.get_example_questions(mcp_server)) |
| 92 | |
| 93 | return { |
| 94 | "server": server_info.model_dump(), |
| 95 | "available_categories": categories, |
| 96 | "total_example_questions": total_questions, |
| 97 | "message": f"Successfully retrieved details for {mcp_server.value}", |
| 98 | "success": True, |
| 99 | } |
| 100 | |
| 101 | except Exception as e: |
| 102 | logger.error(f"Error fetching details for {mcp_server.value}: {str(e)}") |
| 103 | return {"server": mcp_server.value, "message": f"Error retrieving server details: {str(e)}", "success": False} |
| 104 | |
| 105 | |
| 106 | @copilot_mcp_router.get( |
| 107 | "/example-questions", |
| 108 | response_model=ExampleQuestionsResponse, |
| 109 | description="Get example questions for a specific MCP server", |
| 110 | dependencies=[Security(AuthHandler().require_any_scope("admin", "analyst"))], |
| 111 | ) |
| 112 | async def get_example_questions( |
| 113 | mcp_server: MCPServerType = Query(..., description="MCP server to get example questions for"), |
| 114 | category: Optional[str] = Query(None, description="Filter questions by category"), |
| 115 | ) -> ExampleQuestionsResponse: |
| 116 | """ |
| 117 | Retrieve example questions that users can ask for a specific MCP server. |
| 118 | |
| 119 | This endpoint provides users with sample queries they can use to interact |
| 120 | with different MCP servers, helping them understand the capabilities and |
| 121 | available operations for each server type. |
| 122 | |
| 123 | Args: |
| 124 | mcp_server: The MCP server type to get questions for |
| 125 | category: Optional category filter (e.g., 'alerts', 'agents', 'health') |
| 126 | |
| 127 | Returns: |
| 128 | ExampleQuestionsResponse: List of example questions with metadata |
| 129 | """ |
| 130 | logger.info(f"Fetching example questions for MCP server: {mcp_server.value}") |
| 131 | |
| 132 | try: |
| 133 | # Get questions, optionally filtered by category |
| 134 | if category: |
| 135 | questions = ExampleQuestionsService.get_questions_by_category(mcp_server, category) |
| 136 | logger.info(f"Found {len(questions)} questions in category '{category}' for {mcp_server.value}") |
| 137 | else: |
| 138 | questions = ExampleQuestionsService.get_example_questions(mcp_server) |
| 139 | logger.info(f"Found {len(questions)} total questions for {mcp_server.value}") |
| 140 | |
| 141 | return ExampleQuestionsResponse( |
| 142 | mcp_server=mcp_server, |
| 143 | questions=questions, |
| 144 | total_questions=len(questions), |
| 145 | message=f"Successfully retrieved example questions for {mcp_server.value}", |
| 146 | success=True, |
| 147 | ) |
| 148 | |
| 149 | except Exception as e: |
| 150 | logger.error(f"Error fetching example questions for {mcp_server.value}: {str(e)}") |
| 151 | return ExampleQuestionsResponse( |
| 152 | mcp_server=mcp_server, |
| 153 | questions=[], |
| 154 | total_questions=0, |
| 155 | message=f"Error retrieving example questions: {str(e)}", |
| 156 | success=False, |
| 157 | ) |
| 158 | |
| 159 | |
| 160 | @copilot_mcp_router.get( |
| 161 | "/categories", |
| 162 | description="Get available question categories for a specific MCP server", |
| 163 | dependencies=[Security(AuthHandler().require_any_scope("admin", "analyst"))], |
| 164 | ) |
| 165 | async def get_question_categories(mcp_server: MCPServerType = Query(..., description="MCP server to get categories for")) -> dict: |
| 166 | """ |
| 167 | Get available question categories for a specific MCP server. |
| 168 | |
| 169 | Args: |
| 170 | mcp_server: The MCP server type to get categories for |
| 171 | |
| 172 | Returns: |
| 173 | Dictionary containing the available categories |
| 174 | """ |
| 175 | logger.info(f"Fetching question categories for MCP server: {mcp_server.value}") |
| 176 | |
| 177 | try: |
| 178 | categories = ExampleQuestionsService.get_available_categories(mcp_server) |
| 179 | |
| 180 | return { |
| 181 | "mcp_server": mcp_server.value, |
| 182 | "categories": categories, |
| 183 | "total_categories": len(categories), |
| 184 | "message": f"Successfully retrieved categories for {mcp_server.value}", |
| 185 | "success": True, |
| 186 | } |
| 187 | |
| 188 | except Exception as e: |
| 189 | logger.error(f"Error fetching categories for {mcp_server.value}: {str(e)}") |
| 190 | return { |
| 191 | "mcp_server": mcp_server.value, |
| 192 | "categories": [], |
| 193 | "total_categories": 0, |
| 194 | "message": f"Error retrieving categories: {str(e)}", |
| 195 | "success": False, |
| 196 | } |
| 197 | |
| 198 | |
| 199 | @copilot_mcp_router.post( |
| 200 | "/query", |
| 201 | description="Process a query to the appropriate MCP server", |
| 202 | dependencies=[Security(AuthHandler().require_any_scope("admin", "analyst"))], |
| 203 | ) |
| 204 | async def query_mcp(request: MCPQueryRequest, session: AsyncSession = Depends(get_db)) -> MCPQueryResponse: |
| 205 | """ |
| 206 | Process a query to the MCP agent and return structured response. |
| 207 | Routes the query to the appropriate MCP server based on the selected server type. |
| 208 | """ |
| 209 | logger.info(f"Processing MCP query for server: {request.mcp_server.value}") |
| 210 | |
| 211 | license_key = None |
| 212 | |
| 213 | # Check if it's a cloud service and get license key if needed |
| 214 | if MCPService.is_cloud_service(request.mcp_server): |
| 215 | try: |
| 216 | await is_feature_enabled("SOCFORTRESS AI", session=session) |
| 217 | |
| 218 | # Will raise HTTPException(404) if no license record exists |
| 219 | license_info = await get_license(session) |
| 220 | license_key = license_info.license_key |
| 221 | |
| 222 | # If a license record exists but the key is missing, mirror get_license behavior |
| 223 | if not license_key: |
| 224 | raise HTTPException(status_code=404, detail="No license found. A license must be created first.") |
| 225 | |
| 226 | logger.info(f"Retrieved license key for cloud service {request.mcp_server.value}") |
| 227 | |
| 228 | except HTTPException as http_exc: |
| 229 | # Surface the HTTPException unchanged |
| 230 | raise http_exc |
| 231 | except Exception as e: |
| 232 | # Unexpected errors -> 500 |
| 233 | logger.error(f"Failed to get license key for cloud service: {str(e)}") |
| 234 | raise HTTPException(status_code=500, detail=f"License validation failed: {str(e)}") |
| 235 | |
| 236 | # Use the modular service to execute the query |
| 237 | return await MCPService.execute_query(request, license_key=license_key) |