mcpapp/src/tools/conversion.py

"""
Conversion utilities for MCP tools.

This module contains functions to convert between different tool formats
for various LLM providers (OpenAI, Anthropic, etc.).
"""

import logging
from typing import Any

logger = logging.getLogger(__name__)


def convert_to_openai_tools(mcp_tools: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """
    Convert MCP tools to OpenAI tool definitions.

    Args:
        mcp_tools: List of MCP tools (each with server_name, name, description, inputSchema).

    Returns:
        List of OpenAI tool definitions.
    """
    openai_tools = []
    logger.debug(f"Converting {len(mcp_tools)} MCP tools to OpenAI format.")

    for tool in mcp_tools:
        server_name = tool.get("server_name")
        tool_name = tool.get("name")
        description = tool.get("description")
        input_schema = tool.get("inputSchema")

        if not server_name or not tool_name or not description or not input_schema:
            logger.warning(f"Skipping invalid MCP tool definition during OpenAI conversion: {tool}")
            continue

        # Prefix tool name with server name for routing
        prefixed_tool_name = f"{server_name}__{tool_name}"

        # Initialize the OpenAI tool structure
        openai_tool = {
            "type": "function",
            "function": {
                "name": prefixed_tool_name,
                "description": description,
                "parameters": input_schema,  # OpenAI uses JSON Schema directly
            },
        }
        # Basic validation/cleaning of schema if needed could go here
        if not isinstance(input_schema, dict) or input_schema.get("type") != "object":
            logger.warning(f"Input schema for tool '{prefixed_tool_name}' is not a valid JSON object schema. OpenAI might reject this.")
            # Ensure basic structure if missing
            if not isinstance(input_schema, dict):
                input_schema = {}
            if "type" not in input_schema:
                input_schema["type"] = "object"
            if "properties" not in input_schema:
                input_schema["properties"] = {}
            openai_tool["function"]["parameters"] = input_schema

        openai_tools.append(openai_tool)
        logger.debug(f"Converted MCP tool to OpenAI: {prefixed_tool_name}")

    return openai_tools


def convert_to_anthropic_tools(mcp_tools: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """
    Convert MCP tools to Anthropic tool definitions.

    Args:
        mcp_tools: List of MCP tools (each with server_name, name, description, inputSchema).

    Returns:
        List of Anthropic tool definitions.
    """
    logger.debug(f"Converting {len(mcp_tools)} MCP tools to Anthropic format")
    anthropic_tools = []

    for tool in mcp_tools:
        server_name = tool.get("server_name")
        tool_name = tool.get("name")
        description = tool.get("description")
        input_schema = tool.get("inputSchema")

        if not server_name or not tool_name or not description or not input_schema:
            logger.warning(f"Skipping invalid MCP tool definition during Anthropic conversion: {tool}")
            continue

        # Prefix tool name with server name for routing
        prefixed_tool_name = f"{server_name}__{tool_name}"

        # Initialize the Anthropic tool structure
        # Anthropic's format is quite close to JSON Schema
        anthropic_tool = {"name": prefixed_tool_name, "description": description, "input_schema": input_schema}

        # Basic validation/cleaning of schema if needed
        if not isinstance(input_schema, dict) or input_schema.get("type") != "object":
            logger.warning(f"Input schema for tool '{prefixed_tool_name}' is not a valid JSON object schema. Anthropic might reject this.")
            # Ensure basic structure if missing
            if not isinstance(input_schema, dict):
                input_schema = {}
            if "type" not in input_schema:
                input_schema["type"] = "object"
            if "properties" not in input_schema:
                input_schema["properties"] = {}
            anthropic_tool["input_schema"] = input_schema

        anthropic_tools.append(anthropic_tool)
        logger.debug(f"Converted MCP tool to Anthropic: {prefixed_tool_name}")

    return anthropic_tools


def convert_to_google_tools(mcp_tools: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """
    Convert MCP tools to Google Gemini format (dictionary structure).

    Args:
        mcp_tools: List of MCP tools (each with server_name, name, description, inputSchema).

    Returns:
        List containing one dictionary with 'function_declarations'.
    """
    logger.debug(f"Converting {len(mcp_tools)} MCP tools to Google Gemini format")

    function_declarations = []

    for tool in mcp_tools:
        server_name = tool.get("server_name")
        tool_name = tool.get("name")
        description = tool.get("description")
        input_schema = tool.get("inputSchema")

        if not server_name or not tool_name or not description or not input_schema:
            logger.warning(f"Skipping invalid MCP tool definition during Google conversion: {tool}")
            continue

        # Prefix tool name with server name for routing
        prefixed_tool_name = f"{server_name}__{tool_name}"

        # Basic validation/cleaning of schema
        if not isinstance(input_schema, dict) or input_schema.get("type") != "object":
            logger.warning(f"Input schema for tool '{prefixed_tool_name}' is not a valid JSON object schema. Google might reject this.")
            # Ensure basic structure if missing
            if not isinstance(input_schema, dict):
                input_schema = {}
            if "type" not in input_schema:
                input_schema["type"] = "object"
            if "properties" not in input_schema:
                input_schema["properties"] = {}
            # Google requires properties for object type, add dummy if empty
            if not input_schema["properties"]:
                logger.warning(f"Empty properties for tool '{prefixed_tool_name}', adding dummy property for Google.")
                input_schema["properties"] = {"_dummy_param": {"type": "STRING", "description": "Placeholder"}}

        # Create function declaration for Google's format
        function_declaration = {
            "name": prefixed_tool_name,
            "description": description,
            "parameters": input_schema,  # Google uses JSON Schema directly
        }

        function_declarations.append(function_declaration)
        logger.debug(f"Converted MCP tool to Google FunctionDeclaration: {prefixed_tool_name}")

    # Google API expects a list containing one Tool object dict
    google_tools_wrapper = [{"function_declarations": function_declarations}] if function_declarations else []

    logger.debug(f"Final Google tools structure: {google_tools_wrapper}")
    return google_tools_wrapper


# Note: The _handle_schema_construct helper from the reference code is not strictly
# needed if we assume the inputSchema is already valid JSON Schema.
# If complex schemas (anyOf, etc.) need specific handling beyond standard JSON Schema,
# that logic could be added here or within the provider implementations.