mcp_server.py

#!/usr/bin/env python3
"""
OpenAPI to MCP Server
Converts OpenAPI specifications to MCP tools and exposes them via Model Context Protocol
Supports both stdio and SSE transports
"""

import asyncio
import json
import logging
import os
import sys
import glob
from typing import Dict, Any, Optional, List
from pathlib import Path
from urllib.parse import urlparse

import httpx
from mcp.server.fastmcp import FastMCP
from dotenv import load_dotenv

# Setup logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
    handlers=[logging.StreamHandler(), logging.FileHandler("mcp_server.log")],
)
logger = logging.getLogger(__name__)

# Load environment variables
load_dotenv()
logger.info("Environment variables loaded")

# Initialize FastMCP server
mcp = FastMCP("openapi-to-mcp")
logger.info("FastMCP server initialized")

# Configuration
SWAGGER_FOLDER = "swagger_files"
DEFAULT_TIMEOUT = 30.0
# SSE Configuration to help with connection issues
SSE_TIMEOUT = 60.0
SSE_KEEPALIVE = 30.0

# Global storage for OpenAPI spec and tools
openapi_spec: Optional[Dict[str, Any]] = None
api_base_url: Optional[str] = None


# Global variable to store user-provided 3DSpace URL
user_3dspace_url: Optional[str] = None


def get_3dspace_url() -> str:
    """Get the user-provided 3DSpace URL"""
    if user_3dspace_url is None:
        raise ValueError("3DSpace URL not set. Please provide the 3DSpace URL first.")
    return user_3dspace_url


def is_valid_swagger_file(file_path: str) -> bool:
    """Check if a JSON file is a valid OpenAPI/Swagger specification"""
    try:
        # Try UTF-8 first, then fallback to other encodings
        for encoding in ["utf-8", "utf-8-sig", "latin-1", "cp1252"]:
            try:
                with open(file_path, "r", encoding=encoding) as f:
                    data = json.load(f)
                break
            except UnicodeDecodeError:
                continue
        else:
            logger.debug(f"Could not decode file with any encoding: {file_path}")
            return False

        # Check for OpenAPI 3.x format
        if "openapi" in data and isinstance(data["openapi"], str):
            if data["openapi"].startswith("3."):
                logger.debug(f"Valid OpenAPI 3.x file: {file_path}")
                return True

        # Check for Swagger 2.x format
        if "swagger" in data and isinstance(data["swagger"], str):
            if data["swagger"].startswith("2."):
                logger.debug(f"Valid Swagger 2.x file: {file_path}")
                return True

        # Must have info section
        if "info" not in data:
            logger.debug(f"Invalid swagger file (no info section): {file_path}")
            return False

        # Must have paths section
        if "paths" not in data:
            logger.debug(f"Invalid swagger file (no paths section): {file_path}")
            return False

        logger.debug(f"Valid swagger file: {file_path}")
        return True

    except (json.JSONDecodeError, KeyError, TypeError) as e:
        logger.debug(f"Invalid swagger file {file_path}: {str(e)}")
        return False


def merge_swagger_specs(specs: List[Dict[str, Any]]) -> Dict[str, Any]:
    """Merge multiple OpenAPI/Swagger specifications into a single consolidated spec"""
    if not specs:
        raise ValueError("No specifications to merge")

    if len(specs) == 1:
        return specs[0]

    # Use the first spec as the base
    merged_spec = specs[0].copy()

    # Start with the first spec's info, but update title to indicate it's merged
    merged_spec["info"] = merged_spec.get("info", {}).copy()
    merged_spec["info"]["title"] = "Merged OpenAPI Specifications"
    merged_spec["info"][
        "description"
    ] = f"Consolidated from {len(specs)} OpenAPI specifications"

    # Initialize paths if not present
    if "paths" not in merged_spec:
        merged_spec["paths"] = {}

    # Merge paths from all specs
    for i, spec in enumerate(specs[1:], 1):
        spec_paths = spec.get("paths", {})
        for path, path_item in spec_paths.items():
            if path in merged_spec["paths"]:
                # Path already exists, merge the operations
                logger.warning(
                    f"Path {path} exists in multiple specs, merging operations"
                )
                for method, operation in path_item.items():
                    if method in merged_spec["paths"][path]:
                        # Operation already exists, add suffix to operationId to avoid conflicts
                        if "operationId" in operation:
                            operation["operationId"] = (
                                f"{operation['operationId']}_spec{i}"
                            )
                        logger.warning(
                            f"Duplicate operation {method} {path}, renamed operationId"
                        )
                    merged_spec["paths"][path][method] = operation
            else:
                merged_spec["paths"][path] = path_item

    # Merge components/definitions if present
    components_keys = [
        "components",
        "definitions",
        "parameters",
        "responses",
        "securityDefinitions",
    ]
    for key in components_keys:
        if key in merged_spec:
            for spec in specs[1:]:
                if key in spec:
                    if isinstance(merged_spec[key], dict) and isinstance(
                        spec[key], dict
                    ):
                        merged_spec[key].update(spec[key])

    # Merge servers (prefer first spec's servers, but add unique ones)
    all_servers = merged_spec.get("servers", [])
    existing_urls = {server.get("url") for server in all_servers}

    for spec in specs[1:]:
        for server in spec.get("servers", []):
            if server.get("url") not in existing_urls:
                all_servers.append(server)
                existing_urls.add(server.get("url"))

    if all_servers:
        merged_spec["servers"] = all_servers

    logger.info(f"Successfully merged {len(specs)} OpenAPI specifications")
    return merged_spec


async def load_openapi_spec() -> Dict[str, Any]:
    """Load OpenAPI specifications from swagger folder and merge them"""
    global openapi_spec, api_base_url

    if openapi_spec is not None:
        logger.debug("OpenAPI spec already loaded, returning cached version")
        return openapi_spec

    try:
        logger.info(f"Loading OpenAPI specifications from {SWAGGER_FOLDER}")

        # Create swagger folder if it doesn't exist
        swagger_folder_path = Path(SWAGGER_FOLDER)
        if not swagger_folder_path.exists():
            swagger_folder_path.mkdir(parents=True, exist_ok=True)
            logger.info(f"Created swagger folder: {SWAGGER_FOLDER}")

        # Find all JSON files in the swagger folder
        json_files = glob.glob(f"{SWAGGER_FOLDER}/*.json")
        logger.info(f"Found {len(json_files)} JSON files in {SWAGGER_FOLDER}")

        # Filter for valid swagger files
        valid_swagger_files = []
        for file_path in json_files:
            if is_valid_swagger_file(file_path):
                valid_swagger_files.append(file_path)
                logger.info(f"Valid swagger file: {file_path}")
            else:
                logger.warning(f"Skipping invalid swagger file: {file_path}")

        if not valid_swagger_files:
            logger.warning(
                f"No valid OpenAPI files found in {SWAGGER_FOLDER}, creating sample spec"
            )

            # Create a sample OpenAPI spec if none exists
            sample_spec = {
                "openapi": "3.0.0",
                "info": {
                    "title": "Sample API",
                    "version": "1.0.0",
                    "description": "A sample API for demonstration",
                },
                "servers": [
                    {
                        "url": "https://jsonplaceholder.typicode.com",
                        "description": "JSONPlaceholder - Free fake API for testing",
                    }
                ],
                "paths": {
                    "/posts": {
                        "get": {
                            "operationId": "getPosts",
                            "summary": "Get all posts",
                            "description": "Retrieve all posts from the API",
                            "responses": {"200": {"description": "List of posts"}},
                        },
                        "post": {
                            "operationId": "createPost",
                            "summary": "Create a new post",
                            "description": "Create a new post",
                            "requestBody": {
                                "required": True,
                                "content": {
                                    "application/json": {
                                        "schema": {
                                            "type": "object",
                                            "properties": {
                                                "title": {"type": "string"},
                                                "body": {"type": "string"},
                                                "userId": {"type": "integer"},
                                            },
                                        }
                                    }
                                },
                            },
                        },
                    },
                    "/posts/{id}": {
                        "get": {
                            "operationId": "getPostById",
                            "summary": "Get post by ID",
                            "parameters": [
                                {
                                    "name": "id",
                                    "in": "path",
                                    "required": True,
                                    "schema": {"type": "integer"},
                                }
                            ],
                        }
                    },
                    "/users": {
                        "get": {"operationId": "getUsers", "summary": "Get all users"}
                    },
                    "/users/{id}": {
                        "get": {
                            "operationId": "getUserById",
                            "summary": "Get user by ID",
                            "parameters": [
                                {
                                    "name": "id",
                                    "in": "path",
                                    "required": True,
                                    "schema": {"type": "integer"},
                                }
                            ],
                        }
                    },
                },
            }

            sample_file_path = swagger_folder_path / "sample_api.json"
            with open(sample_file_path, "w", encoding="utf-8") as f:
                json.dump(sample_spec, f, indent=2)

            logger.info(f"Created sample OpenAPI specification at {sample_file_path}")
            openapi_spec = sample_spec
        else:
            # Load all valid swagger files
            loaded_specs = []
            for file_path in valid_swagger_files:
                logger.info(f"Loading OpenAPI file: {file_path}")
                # Try multiple encodings for loading
                for encoding in ["utf-8", "utf-8-sig", "latin-1", "cp1252"]:
                    try:
                        with open(file_path, "r", encoding=encoding) as f:
                            spec = json.load(f)
                            loaded_specs.append(spec)
                            logger.debug(
                                f"Loaded spec from {file_path}: {spec.get('info', {}).get('title', 'Unknown')}"
                            )
                        break
                    except UnicodeDecodeError:
                        continue
                else:
                    logger.error(
                        f"Could not decode file with any encoding: {file_path}"
                    )
                    continue

            # Merge all specifications into one
            if len(loaded_specs) == 1:
                openapi_spec = loaded_specs[0]
                logger.info(
                    f"Using single OpenAPI specification: {openapi_spec.get('info', {}).get('title', 'Unknown')}"
                )
            else:
                openapi_spec = merge_swagger_specs(loaded_specs)
                logger.info(
                    f"Merged {len(loaded_specs)} OpenAPI specifications into one"
                )

        # Set a placeholder base URL - actual URL will be provided by user
        api_base_url = "{3DSpace}"
        logger.info(f"OpenAPI spec loaded successfully")

        return openapi_spec

    except Exception as e:
        logger.error(f"Failed to load OpenAPI specifications: {str(e)}")
        raise Exception(f"Failed to load OpenAPI specifications: {str(e)}")


def get_operation_base_url(operation_path: str, operation_method: str) -> str:
    """Get the base URL for a specific operation, using user-provided 3DSpace URL"""
    try:
        base_url = get_3dspace_url()
        logger.debug(
            f"Using user-provided 3DSpace URL for {operation_method} {operation_path}: {base_url}"
        )
        return base_url
    except ValueError:
        logger.warning("3DSpace URL not set, using placeholder URL")
        return "https://3dspace.mydomain.com/3dspace"


async def make_api_request(
    method: str,
    path: str,
    headers: Optional[Dict[str, str]] = None,
    params: Optional[Dict[str, Any]] = None,
    query_params: Optional[Dict[str, Any]] = None,
    body: Optional[Dict[str, Any]] = None,
    base_url_override: Optional[str] = None,
) -> Dict[str, Any]:
    """Make HTTP request to the API"""

    # Get the appropriate base URL for this operation
    if base_url_override:
        operation_base_url = base_url_override
        logger.info(f"Using base URL override: {operation_base_url}")
    else:
        operation_base_url = get_operation_base_url(path, method)

    # Replace path parameters
    formatted_path = path
    if params:
        for key, value in params.items():
            placeholder = "{" + key + "}"
            formatted_path = formatted_path.replace(placeholder, str(value))
        logger.debug(f"Formatted path with parameters: {formatted_path}")

    # Construct full URL
    if operation_base_url.endswith("/") and formatted_path.startswith("/"):
        url = operation_base_url[:-1] + formatted_path
    elif not operation_base_url.endswith("/") and not formatted_path.startswith("/"):
        url = operation_base_url + "/" + formatted_path
    else:
        url = operation_base_url + formatted_path

    # Prepare headers
    request_headers = headers or {}

    logger.info(f"Making {method.upper()} request to: {url}")
    logger.debug(f"Headers: {request_headers}")
    logger.debug(f"Query params: {query_params}")
    logger.debug(f"Body: {body}")

    try:
        timeout = httpx.Timeout(DEFAULT_TIMEOUT)
        async with httpx.AsyncClient(timeout=timeout) as client:
            if method.lower() == "get":
                response = await client.get(
                    url, headers=request_headers, params=query_params
                )
            elif method.lower() == "post":
                response = await client.post(
                    url, headers=request_headers, json=body, params=query_params
                )
            elif method.lower() == "put":
                response = await client.put(
                    url, headers=request_headers, json=body, params=query_params
                )
            elif method.lower() == "delete":
                response = await client.delete(
                    url, headers=request_headers, params=query_params
                )
            elif method.lower() == "patch":
                response = await client.patch(
                    url, headers=request_headers, json=body, params=query_params
                )
            else:
                response = await client.request(
                    method.upper(),
                    url,
                    headers=request_headers,
                    json=body,
                    params=query_params,
                )

        logger.info(
            f"API request completed: {method.upper()} {url} - Status: {response.status_code}"
        )

        # Handle different content types
        content_type = response.headers.get("content-type", "").lower()
        if "application/json" in content_type:
            try:
                response_body = response.json()
            except:
                response_body = response.text
                logger.warning(
                    f"Failed to parse JSON response, using text: {response.text[:200]}..."
                )
        else:
            response_body = response.text

        return {
            "status_code": response.status_code,
            "headers": dict(response.headers),
            "body": response_body,
            "url": url,
            "method": method.upper(),
            "success": 200 <= response.status_code < 300,
        }

    except Exception as e:
        logger.error(f"API request failed: {method.upper()} {url} - Error: {str(e)}")
        return {
            "error": str(e),
            "status_code": 500,
            "url": url,
            "method": method.upper(),
            "success": False,
        }


@mcp.resource("openapi://spec")
def get_openapi_spec() -> str:
    """Get the loaded OpenAPI specification"""
    if openapi_spec is None:
        return "OpenAPI specification not loaded"

    return json.dumps(
        {
            "info": openapi_spec.get("info", {}),
            "servers": openapi_spec.get("servers", []),
            "paths_count": len(openapi_spec.get("paths", {})),
            "base_url": api_base_url,
        },
        indent=2,
    )


@mcp.resource("openapi://paths")
def list_api_paths() -> str:
    """List all available API paths and methods"""
    if openapi_spec is None:
        return "OpenAPI specification not loaded"

    paths = openapi_spec.get("paths", {})
    api_paths = []

    for path, methods in paths.items():
        for method, details in methods.items():
            if method.lower() in [
                "get",
                "post",
                "put",
                "delete",
                "patch",
                "head",
                "options",
            ]:
                operation_id = details.get("operationId", f"{method}_{path}")
                summary = details.get("summary", f"{method.upper()} {path}")
                api_paths.append(
                    {
                        "operation_id": operation_id,
                        "method": method.upper(),
                        "path": path,
                        "summary": summary,
                    }
                )

    return json.dumps(api_paths, indent=2)


@mcp.tool()
async def set_3dspace_url(url: str) -> str:
    """Set the 3DSpace URL that will be used for all API calls

    Args:
        url: The 3DSpace URL (e.g., https://3dspace.mydomain.com/3dspace)

    Returns:
        Confirmation message
    """
    global user_3dspace_url

    # Clean up the URL - remove trailing slash if present
    clean_url = url.rstrip("/")

    # Validate URL format
    try:
        parsed = urlparse(clean_url)
        if not parsed.scheme or not parsed.netloc:
            return json.dumps(
                {
                    "error": "Invalid URL format. Please provide a complete URL like https://3dspace.mydomain.com/3dspace"
                }
            )
    except Exception as e:
        return json.dumps({"error": f"Invalid URL: {str(e)}"})

    user_3dspace_url = clean_url
    logger.info(f"3DSpace URL set to: {user_3dspace_url}")

    return json.dumps(
        {
            "success": True,
            "message": f"3DSpace URL set to: {user_3dspace_url}",
            "url": user_3dspace_url,
        }
    )


@mcp.tool()
async def list_available_tools() -> str:
    """
    List all available MCP tools with their descriptions and parameters
    """
    # Load spec to ensure we have the latest operations
    await load_openapi_spec()

    logger.info("Listing available tools")

    tools_info = {
        "mcp_tools": [
            {
                "name": "list_available_tools",
                "description": "List all available MCP tools with their descriptions",
                "parameters": [],
            },
            {
                "name": "api_request",
                "description": "Make API requests using OpenAPI operation IDs",
                "parameters": [
                    {
                        "name": "operation_id",
                        "type": "string",
                        "required": True,
                        "description": "The operation ID from OpenAPI spec",
                    },
                    {
                        "name": "path_params",
                        "type": "object",
                        "required": False,
                        "description": "Path parameters",
                    },
                    {
                        "name": "query_params",
                        "type": "object",
                        "required": False,
                        "description": "Query parameters",
                    },
                    {
                        "name": "headers",
                        "type": "object",
                        "required": False,
                        "description": "Additional headers",
                    },
                    {
                        "name": "body",
                        "type": "object",
                        "required": False,
                        "description": "Request body for POST/PUT/PATCH",
                    },
                ],
            },
            {
                "name": "generic_api_call",
                "description": "Make generic API calls to any endpoint",
                "parameters": [
                    {
                        "name": "method",
                        "type": "string",
                        "required": True,
                        "description": "HTTP method",
                    },
                    {
                        "name": "path",
                        "type": "string",
                        "required": True,
                        "description": "API path",
                    },
                    {
                        "name": "path_params",
                        "type": "object",
                        "required": False,
                        "description": "Path parameters",
                    },
                    {
                        "name": "query_params",
                        "type": "object",
                        "required": False,
                        "description": "Query parameters",
                    },
                    {
                        "name": "headers",
                        "type": "object",
                        "required": False,
                        "description": "Headers",
                    },
                    {
                        "name": "body",
                        "type": "object",
                        "required": False,
                        "description": "Request body",
                    },
                ],
            },
            {
                "name": "search_operations",
                "description": "Search for API operations by keyword",
                "parameters": [
                    {
                        "name": "query",
                        "type": "string",
                        "required": True,
                        "description": "Search term",
                    }
                ],
            },
            {
                "name": "get_security_context",
                "description": "Get security context information from 3DSpace and format as Role.Organization.CollabSpace strings",
                "parameters": [
                    {
                        "name": "headers",
                        "type": "object",
                        "required": False,
                        "description": "Authentication headers (Authorization and ENO_CSRF_TOKEN required)",
                    }
                ],
            },
            {
                "name": "get_csrf_token",
                "description": "Get CSRF token from 3DSpace for PUT, PATCH, POST, or DELETE operations",
                "parameters": [
                    {
                        "name": "headers",
                        "type": "object",
                        "required": False,
                        "description": "Authentication headers (Authorization required)",
                    }
                ],
            },
        ],
        "api_operations": [],
    }

    # Add available API operations
    if openapi_spec:
        paths = openapi_spec.get("paths", {})
        for path, methods in paths.items():
            for method, details in methods.items():
                if method.lower() in [
                    "get",
                    "post",
                    "put",
                    "delete",
                    "patch",
                    "head",
                    "options",
                ]:
                    operation_id = details.get("operationId", f"{method}_{path}")
                    summary = details.get("summary", f"{method.upper()} {path}")
                    description = details.get("description", "")

                    tools_info["api_operations"].append(
                        {
                            "operation_id": operation_id,
                            "method": method.upper(),
                            "path": path,
                            "summary": summary,
                            "description": description,
                        }
                    )

    tools_info["total_mcp_tools"] = len(tools_info["mcp_tools"])
    tools_info["total_api_operations"] = len(tools_info["api_operations"])

    logger.info(
        f"Listed {tools_info['total_mcp_tools']} MCP tools and {tools_info['total_api_operations']} API operations"
    )
    return json.dumps(tools_info, indent=2)


@mcp.tool()
async def api_request(
    operation_id: str,
    path_params: Optional[Dict[str, Any]] = None,
    query_params: Optional[Dict[str, Any]] = None,
    headers: Optional[Dict[str, str]] = None,
    body: Optional[Dict[str, Any]] = None,
) -> str:
    """
    Make an API request using operation ID from the OpenAPI specification

    Args:
        operation_id: The operationId from the OpenAPI spec
        path_params: Path parameters for the request (e.g., {"id": "123"})
        query_params: Query parameters for the request (e.g., {"limit": 10})
        headers: Additional headers for the request
        body: Request body for POST/PUT/PATCH requests
    """

    logger.info(f"Making API request with operation ID: {operation_id}")

    # Load spec if not already loaded
    spec = await load_openapi_spec()

    # Find the operation by operation_id
    paths = spec.get("paths", {})
    operation_found = False
    target_path = None
    target_method = None

    for path, methods in paths.items():
        for method, details in methods.items():
            if method.lower() not in [
                "get",
                "post",
                "put",
                "delete",
                "patch",
                "head",
                "options",
            ]:
                continue

            op_id = details.get("operationId")
            if not op_id:
                # Generate operation ID from method and path
                clean_path = (
                    path.replace("/", "_")
                    .replace("{", "")
                    .replace("}", "")
                    .replace(":", "_")
                )
                op_id = f"{method}_{clean_path}"

            if op_id == operation_id:
                target_path = path
                target_method = method
                operation_found = True
                logger.debug(f"Found operation: {target_method.upper()} {target_path}")
                break

        if operation_found:
            break

    if not operation_found:
        available_operations = []
        for path, methods in paths.items():
            for method, details in methods.items():
                if method.lower() in [
                    "get",
                    "post",
                    "put",
                    "delete",
                    "patch",
                    "head",
                    "options",
                ]:
                    op_id = details.get("operationId", f"{method}_{path}")
                    available_operations.append(op_id)

        logger.warning(
            f"Operation '{operation_id}' not found. Available operations: {available_operations[:5]}"
        )
        return f"Operation '{operation_id}' not found. Available operations: {', '.join(available_operations[:10])}"

    # Make the API request
    result = await make_api_request(
        method=target_method,
        path=target_path,
        headers=headers,
        params=path_params,
        query_params=query_params,
        body=body,
    )

    logger.info(f"API request completed for operation: {operation_id}")
    return json.dumps(result, indent=2)


@mcp.tool()
async def generic_api_call(
    method: str,
    path: str,
    path_params: Optional[Dict[str, Any]] = None,
    query_params: Optional[Dict[str, Any]] = None,
    headers: Optional[Dict[str, str]] = None,
    body: Optional[Dict[str, Any]] = None,
) -> str:
    """
    Make a generic API call to any path

    Args:
        method: HTTP method (GET, POST, PUT, DELETE, PATCH)
        path: API path (e.g., "/users/{id}")
        path_params: Path parameters (e.g., {"id": "123"})
        query_params: Query parameters (e.g., {"limit": 10})
        headers: Additional headers
        body: Request body for POST/PUT/PATCH requests
    """

    logger.info(f"Making generic API call: {method.upper()} {path}")

    # Ensure spec is loaded to get base URL
    await load_openapi_spec()

    # Validate method
    if method.upper() not in [
        "GET",
        "POST",
        "PUT",
        "DELETE",
        "PATCH",
        "HEAD",
        "OPTIONS",
    ]:
        logger.error(f"Invalid HTTP method: {method}")
        return f"Invalid HTTP method: {method}. Supported methods: GET, POST, PUT, DELETE, PATCH, HEAD, OPTIONS"

    # Make the API request
    result = await make_api_request(
        method=method,
        path=path,
        headers=headers,
        params=path_params,
        query_params=query_params,
        body=body,
    )

    logger.info(f"Generic API call completed: {method.upper()} {path}")
    return json.dumps(result, indent=2)


@mcp.tool()
async def search_operations(query: str) -> str:
    """
    Search for API operations by name, path, or description

    Args:
        query: Search term to find matching operations
    """

    logger.info(f"Searching operations with query: '{query}'")

    # Load spec if not already loaded
    spec = await load_openapi_spec()

    paths = spec.get("paths", {})
    matching_operations = []

    for path, methods in paths.items():
        for method, details in methods.items():
            if method.lower() not in [
                "get",
                "post",
                "put",
                "delete",
                "patch",
                "head",
                "options",
            ]:
                continue

            operation_id = details.get("operationId", f"{method}_{path}")
            summary = details.get("summary", "")
            description = details.get("description", "")

            # Search in operation_id, path, summary, and description
            search_text = f"{operation_id} {path} {summary} {description}".lower()

            if query.lower() in search_text:
                matching_operations.append(
                    {
                        "operation_id": operation_id,
                        "method": method.upper(),
                        "path": path,
                        "summary": summary,
                        "description": description,
                    }
                )

    logger.info(f"Found {len(matching_operations)} operations matching '{query}'")

    if not matching_operations:
        logger.warning(f"No operations found matching '{query}'")
        return f"No operations found matching '{query}'"

    return json.dumps(matching_operations, indent=2)


@mcp.tool()
async def get_security_context(
    headers: Optional[Dict[str, str]] = None,
) -> str:
    """
    Get security context information from 3DSpace and format it as Role.Organization.CollabSpace strings

    Args:
        headers: Authentication headers (Authorization and ENO_CSRF_TOKEN required)

    Returns:
        JSON response with formatted security context array and original collabspaces data
    """

    logger.info("Getting security context from 3DSpace")

    # Ensure spec is loaded to get base URL
    await load_openapi_spec()

    if not headers:
        logger.warning("Security context request failed: Missing headers")
        return json.dumps(
            {
                "error": "Headers required. Please provide Authorization and ENO_CSRF_TOKEN headers.",
                "success": False,
            },
            indent=2,
        )

    logger.debug(
        f"Making security context request with headers: {list(headers.keys())}"
    )

    # Use user-provided 3DSpace URL for security context requests
    try:
        security_base_url = get_3dspace_url()
        logger.debug(
            f"Using user-provided 3DSpace URL for security context request: {security_base_url}"
        )
    except ValueError as e:
        return json.dumps(
            {
                "error": str(e),
                "success": False,
            },
            indent=2,
        )

    # Make the API request to get user's collaborative spaces
    result = await make_api_request(
        method="GET",
        path="/resources/modeler/pno/person",
        headers=headers,
        query_params={"current": True, "select": "collabspaces"},
        base_url_override=security_base_url,
    )

    if not result.get("success", False):
        logger.error(
            f"Security context API request failed: {result.get('error', 'Unknown error')}"
        )
        return json.dumps(result, indent=2)

    try:
        # Parse string result to JSON if needed
        response_body = result.get("body", {})
        if isinstance(response_body, str):
            response_body = json.loads(response_body)

        # Extract user information
        user_pid = response_body.get("pid", "")
        user_name = response_body.get("name", "")
        collabspaces = response_body.get("collabspaces", [])

        logger.info(
            f"Processing security context for user: {user_name} (PID: {user_pid})"
        )
        logger.debug(f"Found {len(collabspaces)} collaborative spaces")

        # Process collabspaces to generate security context strings
        security_contexts = []

        for collabspace in collabspaces:
            collabspace_name = collabspace.get("name", "")
            couples = collabspace.get("couples", [])

            logger.debug(
                f"Processing collabspace: {collabspace_name} with {len(couples)} couples"
            )

            for couple in couples:
                organization = couple.get("organization", {})
                role = couple.get("role", {})

                organization_name = organization.get("name", "")
                role_name = role.get("name", "")

                if role_name and organization_name and collabspace_name:
                    security_context = (
                        f"{role_name}.{organization_name}.{collabspace_name}"
                    )
                    security_contexts.append(security_context)
                    logger.debug(f"Generated security context: {security_context}")

        # Remove duplicates while preserving order
        unique_security_contexts = []
        seen = set()
        for context in security_contexts:
            if context not in seen:
                unique_security_contexts.append(context)
                seen.add(context)

        logger.info(
            f"Generated {len(unique_security_contexts)} unique security contexts"
        )

        # Log the response URL for reference
        response_url = result.get("url", "")
        logger.debug(f"Security context API response URL: {response_url}")

        # Prepare formatted response
        formatted_response = {
            "success": True,
            "user": {"pid": user_pid, "name": user_name},
            "security_contexts": unique_security_contexts,
            "total_contexts": len(unique_security_contexts),
            "collabspaces": collabspaces,
            "original_response": response_body,
        }

        logger.info("Security context processing completed successfully")
        return json.dumps(formatted_response, indent=2)

    except Exception as e:
        logger.error(
            f"Failed to process security context data: {str(e)}", exc_info=True
        )
        return json.dumps(
            {
                "error": f"Failed to process security context data: {str(e)}",
                "success": False,
                "raw_response": result,
            },
            indent=2,
        )


@mcp.tool()
async def get_csrf_token(
    headers: Optional[Dict[str, str]] = None,
) -> str:
    """
    Get CSRF token from 3DSpace for PUT, PATCH, POST, or DELETE operations

    Generates a CSRF token valid for 6W and PN0 web services. The token is valid
    for the entire Java session and becomes invalid after a new 3DSpace login.

    Args:
        headers: Authentication headers (Authorization required)

    Returns:
        JSON response with CSRF token name and value, plus original response data
    """

    logger.info("Getting CSRF token from 3DSpace")

    # Ensure spec is loaded to get base URL
    await load_openapi_spec()

    if not headers:
        logger.warning("CSRF token request failed: Missing headers")
        return json.dumps(
            {
                "error": "Headers required. Please provide Authorization header for authentication.",
                "success": False,
            },
            indent=2,
        )

    logger.debug(f"Making CSRF token request with headers: {list(headers.keys())}")

    # Use user-provided 3DSpace URL for CSRF requests
    try:
        csrf_base_url = get_3dspace_url()
        logger.debug(
            f"Using user-provided 3DSpace URL for CSRF request: {csrf_base_url}"
        )
    except ValueError as e:
        return json.dumps(
            {
                "error": str(e),
                "success": False,
            },
            indent=2,
        )

    # Make the API request to get CSRF token
    result = await make_api_request(
        method="GET",
        path="/resources/v1/application/CSRF",
        headers=headers,
        base_url_override=csrf_base_url,
    )

    if not result.get("success", False):
        logger.error(
            f"CSRF token API request failed: {result.get('error', 'Unknown error')}"
        )
        return json.dumps(result, indent=2)

    try:
        # Parse string result to JSON if needed
        response_body = result.get("body", {})
        if isinstance(response_body, str):
            response_body = json.loads(response_body)

        # Extract CSRF token information
        csrf_data = response_body.get("csrf", {})
        csrf_name = csrf_data.get("name", "")
        csrf_value = csrf_data.get("value", "")

        logger.debug(
            f"CSRF token extracted - Name: {csrf_name}, Value length: {len(csrf_value) if csrf_value else 0}"
        )

        # Check if we got valid CSRF data
        if not csrf_name or not csrf_value:
            logger.error("Invalid CSRF response - missing name or value")
            return json.dumps(
                {
                    "error": "Invalid CSRF response - missing name or value",
                    "success": False,
                    "raw_response": response_body,
                },
                indent=2,
            )

        logger.info(f"CSRF token successfully obtained: {csrf_name}")

        # Prepare formatted response
        formatted_response = {
            "success": True,
            "csrf_token": {
                "name": csrf_name,
                "value": csrf_value,
                "header_format": f"{csrf_name}: {csrf_value}",
            },
            "usage_info": {
                "description": "Use this token for PUT, PATCH, POST, or DELETE requests",
                "validity": "Valid for the entire Java session until next 3DSpace login",
                "header_name": csrf_name,
                "header_value": csrf_value,
            },
            "original_response": response_body,
        }

        logger.info("CSRF token processing completed successfully")
        return json.dumps(formatted_response, indent=2)

    except Exception as e:
        logger.error(f"Failed to process CSRF token data: {str(e)}", exc_info=True)
        return json.dumps(
            {
                "error": f"Failed to process CSRF token data: {str(e)}",
                "success": False,
                "raw_response": result,
            },
            indent=2,
        )


async def startup_handler():
    """Initialize the server and load OpenAPI specification"""
    try:
        logger.info("[INIT] Starting OpenAPI to MCP Server initialization...")
        spec = await load_openapi_spec()

        # Display API info
        api_info = spec.get("info", {})
        api_title = api_info.get("title", "Unknown")
        api_version = api_info.get("version", "Unknown")
        api_description = api_info.get("description", "No description")

        logger.info(f"API: {api_title} v{api_version}")
        logger.info(f"Description: {api_description}")
        logger.info(f"Base URL: {api_base_url}")

        # Count available operations
        paths = spec.get("paths", {})
        operation_count = 0
        for path, methods in paths.items():
            for method in methods.keys():
                if method.lower() in [
                    "get",
                    "post",
                    "put",
                    "delete",
                    "patch",
                    "head",
                    "options",
                ]:
                    operation_count += 1

        logger.info(f"[SUCCESS] Successfully loaded {operation_count} API operations!")
        logger.info("[READY] MCP Server ready!")

        print(f"API: {api_title} v{api_version}")
        print(f"Description: {api_description}")
        print(f"Base URL: {api_base_url}")
        print(f"[SUCCESS] Successfully loaded {operation_count} API operations!")
        print("[READY] MCP Server ready!")

    except Exception as e:
        logger.error(f"[ERROR] Error during startup: {e}", exc_info=True)
        print(f"[ERROR] Error during startup: {e}")
        raise


def run_sse_server():
    """Run SSE server with better error handling"""
    try:
        # Set environment variables that might help with SSE issues
        os.environ.setdefault("UVICORN_TIMEOUT_KEEP_ALIVE", str(int(SSE_KEEPALIVE)))

        logger.info("[SSE] Starting SSE Server...")
        logger.info("[INFO] Server will be available at: http://localhost:8080/sse")
        print("[SSE] Starting SSE Server...")
        print("[INFO] Server will be available at: http://localhost:8080/sse")
        print(
            "[INFO] For SSE transport: Use http://localhost:8080/sse as the MCP server URL"
        )
        print("[TIP] To use stdio instead: python mcp_server.py --stdio")

        # Run with error handling
        mcp.run(transport="sse")

    except KeyboardInterrupt:
        logger.info("[STOP] Server stopped by user")
        print("\n[STOP] Server stopped by user")
    except Exception as e:
        logger.error(f"[ERROR] SSE transport failed: {e}", exc_info=True)
        print(f"[ERROR] SSE transport failed: {e}")
        print("[TIP] Common solutions:")
        print("   1. Try using stdio transport: python mcp_server.py --stdio")
        print("   2. Check if port 8080 is already in use")
        print("   3. Try restarting your terminal/IDE")
        return False
    return True


def main():
    """Main entry point with transport selection"""
    transport = "sse"  # Default to SSE transport for IDE integration

    logger.info("Starting OpenAPI to MCP Server")

    # Check command line arguments for transport type
    if len(sys.argv) > 1:
        if sys.argv[1] in ["--sse", "sse"]:
            transport = "sse"
        elif sys.argv[1] in ["--stdio", "stdio"]:
            transport = "stdio"
        elif sys.argv[1] in ["--help", "-h"]:
            print("Usage: python mcp_server.py [transport]")
            print("  transport: sse (default) or stdio")
            print(
                "  --sse: Start server with SSE transport for web/IDE integration (default)"
            )
            print("  --stdio: Start server with stdio transport for command line")
            return

    logger.info(f"Selected transport: {transport}")

    # Run startup handler first
    try:
        asyncio.run(startup_handler())
    except Exception as e:
        logger.error(f"[ERROR] Failed to initialize server: {e}", exc_info=True)
        print(f"[ERROR] Failed to initialize server: {e}")
        return

    # Start the MCP server with selected transport
    logger.info(f"Starting OpenAPI to MCP Server with {transport} transport...")
    print(f"Starting OpenAPI to MCP Server with {transport} transport...")

    if transport == "sse":
        run_sse_server()
    else:
        logger.info("[STDIO] Using stdio transport for command line integration")
        print("[STDIO] Using stdio transport for command line integration")
        print("[TIP] To use SSE instead: python mcp_server.py --sse")

        try:
            mcp.run(transport="stdio")
        except Exception as e:
            logger.error(f"[ERROR] stdio transport failed: {e}", exc_info=True)
            print(f"[ERROR] stdio transport failed: {e}")
            raise


if __name__ == "__main__":
    main()