42Crunch MCP Server

"""LangChain tools for 42crunch MCP Server.

Token Configuration:
-------------------
1. 42C_TOKEN: Required by the MCP server (server-side)
   - Set in the MCP server's environment (.env file or environment variable)
   - Used by the MCP server to authenticate with 42crunch API
   - NOT passed from the UI tools - the server already has it

2. LLM API Key: Required by the UI (client-side)
   - OpenAI: OPENAI_API_KEY
   - Claude: ANTHROPIC_API_KEY  
   - Gemini: GOOGLE_API_KEY
   - Set in ~/.ai_tokens, .env file, or environment variable
   - Used by LangChain agent in app.py (not by mcp_tools.py)

The MCP tools only need the MCP server URL - they call the server via HTTP,
and the server uses its own 42C_TOKEN to authenticate with 42crunch API.
"""

import requests
import os
from typing import Optional, Dict, Any, List
from langchain.tools import BaseTool
from pydantic import BaseModel, Field


class MCPTool(BaseTool):
    """Base class for MCP tools.
    
    These tools call the MCP HTTP server which handles authentication
    with 42crunch API using 42C_TOKEN (set server-side).
    
    The UI tools don't need to pass tokens - they just call the MCP server.
    """
    
    mcp_server_url: str = Field(
        default="http://localhost:8000",
        description="URL of the MCP HTTP server (must have 42C_TOKEN configured)"
    )
    
    def _call_mcp_server(self, tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
        """Call MCP server tool via JSON-RPC.
        
        Note: The MCP server must have 42C_TOKEN set in its environment
        to authenticate with 42crunch API. This method doesn't pass tokens.
        """
        try:
            response = requests.post(
                f"{self.mcp_server_url}/jsonrpc",
                json={
                    "jsonrpc": "2.0",
                    "id": 1,
                    "method": "tools/call",
                    "params": {
                        "name": tool_name,
                        "arguments": arguments
                    }
                },
                headers={"Content-Type": "application/json"},
                timeout=30
            )
            response.raise_for_status()
            result = response.json()
            
            if "error" in result:
                return {"success": False, "error": result["error"]}
            
            return result.get("result", {})
        except requests.exceptions.ConnectionError as e:
            return {
                "success": False,
                "error": f"Cannot connect to MCP server at {self.mcp_server_url}. Make sure the server is running and has 42C_TOKEN configured."
            }
        except requests.exceptions.HTTPError as e:
            return {
                "success": False,
                "error": f"HTTP error from MCP server: {e.response.status_code} - {e.response.text[:200]}"
            }
        except Exception as e:
            return {"success": False, "error": str(e)}


class ListCollectionsInput(BaseModel):
    """Input for list_collections tool."""
    page: Optional[int] = Field(default=1, description="Page number (default: 1)")
    per_page: Optional[int] = Field(default=10, description="Items per page (default: 10)")


class ListCollectionsTool(MCPTool):
    """Tool to list 42crunch collections."""
    
    name: str = "list_collections"
    description: str = """
    List all collections in 42crunch.
    Use this to see available collections and get collection IDs.
    
    Args:
        page: Page number (default: 1)
        per_page: Number of items per page (default: 10)
    
    Returns:
        List of collections with their IDs, names, and metadata.
    """
    args_schema: type[BaseModel] = ListCollectionsInput
    
    def _run(self, page: int = 1, per_page: int = 10) -> str:
        """Execute the tool."""
        result = self._call_mcp_server("list_collections", {
            "page": page,
            "per_page": per_page
        })
        
        if result.get("success"):
            data = result.get("data", {})
            # Format the response for LLM
            # API returns collections in 'list' key
            if isinstance(data, dict):
                collections = data.get("list") or data.get("collections") or data.get("items", [])
            elif isinstance(data, list):
                collections = data
            else:
                collections = []
            
            if not collections:
                return "No collections found."
            
            formatted = f"Found {len(collections)} collection(s):\n\n"
            for i, coll in enumerate(collections, 1):
                coll_id = coll.get("id") or coll.get("collection_id") or coll.get("uuid", "N/A")
                name = coll.get("name") or coll.get("title", "Unnamed")
                api_count = coll.get("apiCount", 0)
                formatted += f"{i}. {name} (ID: {coll_id}, APIs: {api_count})\n"
            
            return formatted
        else:
            error = result.get("error", "Unknown error")
            return f"Error listing collections: {error}"


class GetCollectionAPIsInput(BaseModel):
    """Input for get_collection_apis tool."""
    collection_id: str = Field(description="Collection UUID")
    with_tags: Optional[bool] = Field(default=True, description="Include tags (default: True)")


class GetCollectionAPIsTool(MCPTool):
    """Tool to get APIs in a collection."""
    
    name: str = "get_collection_apis"
    description: str = """
    Get all APIs in a specific 42crunch collection.
    Use this after listing collections to get APIs within a collection.
    
    Args:
        collection_id: The UUID of the collection (required)
        with_tags: Include tags in response (default: True)
    
    Returns:
        List of APIs in the collection with their IDs, names, and metadata.
    """
    args_schema: type[BaseModel] = GetCollectionAPIsInput
    
    def _run(self, collection_id: str, with_tags: bool = True) -> str:
        """Execute the tool."""
        result = self._call_mcp_server("get_collection_apis", {
            "collection_id": collection_id,
            "with_tags": with_tags
        })
        
        if result.get("success"):
            data = result.get("data", {})
            # Format the response for LLM
            # API returns APIs in 'list' or 'apis' key
            if isinstance(data, dict):
                apis = data.get("list") or data.get("apis") or data.get("items", [])
            elif isinstance(data, list):
                apis = data
            else:
                apis = []
            
            if not apis:
                return f"No APIs found in collection {collection_id}."
            
            formatted = f"Found {len(apis)} API(s) in collection:\n\n"
            for i, api in enumerate(apis, 1):
                api_id = api.get("id") or api.get("api_id") or api.get("uuid", "N/A")
                name = api.get("name") or api.get("title", "Unnamed")
                formatted += f"{i}. {name} (ID: {api_id})\n"
            
            return formatted
        else:
            error = result.get("error", "Unknown error")
            return f"Error getting collection APIs: {error}"


class GetAPIDetailsInput(BaseModel):
    """Input for get_api_details tool."""
    api_id: str = Field(description="API UUID")
    branch: Optional[str] = Field(default="main", description="Git branch (default: main)")
    include_definition: Optional[bool] = Field(default=True, description="Include OpenAPI definition")
    include_assessment: Optional[bool] = Field(default=False, description="Include assessment data")
    include_scan: Optional[bool] = Field(default=False, description="Include scan results")


class GetAPIDetailsTool(MCPTool):
    """Tool to get detailed information about an API."""
    
    name: str = "get_api_details"
    description: str = """
    Get detailed information about a specific API in 42crunch.
    Use this after getting APIs from a collection to see full API details.
    
    Args:
        api_id: The UUID of the API (required)
        branch: Git branch name (default: main)
        include_definition: Include OpenAPI definition (default: True)
        include_assessment: Include security assessment data (default: False)
        include_scan: Include scan results (default: False)
    
    Returns:
        Detailed API information including definition, assessment, and scan results.
    """
    args_schema: type[BaseModel] = GetAPIDetailsInput
    
    def _run(
        self,
        api_id: str,
        branch: str = "main",
        include_definition: bool = True,
        include_assessment: bool = False,
        include_scan: bool = False
    ) -> str:
        """Execute the tool."""
        result = self._call_mcp_server("get_api_details", {
            "api_id": api_id,
            "branch": branch,
            "include_definition": include_definition,
            "include_assessment": include_assessment,
            "include_scan": include_scan
        })
        
        if result.get("success"):
            data = result.get("data", {})
            
            # Format the response for LLM
            formatted = f"API Details (ID: {api_id}):\n\n"
            
            # Extract key information
            name = data.get("name") or data.get("title", "Unnamed API")
            formatted += f"Name: {name}\n"
            
            if "definition" in data and include_definition:
                def_info = data["definition"]
                if isinstance(def_info, dict):
                    formatted += f"OpenAPI Version: {def_info.get('openapi', 'N/A')}\n"
                    formatted += f"Info: {def_info.get('info', {}).get('title', 'N/A')}\n"
            
            if "assessment" in data and include_assessment:
                assessment = data["assessment"]
                formatted += f"Assessment: Available\n"
            
            if "scan" in data and include_scan:
                scan = data["scan"]
                formatted += f"Scan Results: Available\n"
            
            return formatted
        else:
            error = result.get("error", "Unknown error")
            return f"Error getting API details: {error}"


def get_mcp_tools(mcp_server_url: Optional[str] = None) -> List[MCPTool]:
    """Get all MCP tools as LangChain tools.
    
    Args:
        mcp_server_url: URL of the MCP HTTP server. If None, uses:
            - MCP_SERVER_URL environment variable, or
            - Default: http://localhost:8000
    
    Returns:
        List of MCP tools configured to call the MCP server.
    
    Note:
        The MCP server must have 42C_TOKEN configured in its environment
        to authenticate with 42crunch API. The UI tools don't pass tokens.
    """
    if mcp_server_url is None:
        mcp_server_url = os.getenv("MCP_SERVER_URL", "http://localhost:8000")
    
    tools = [
        ListCollectionsTool(mcp_server_url=mcp_server_url),
        GetCollectionAPIsTool(mcp_server_url=mcp_server_url),
        GetAPIDetailsTool(mcp_server_url=mcp_server_url),
    ]
    return tools