my-pal-mcp-server/server.py

"""
Gemini MCP Server - Main server implementation
"""

import asyncio
import logging
import os
import sys
from datetime import datetime
from typing import Any, Dict, List

import google.generativeai as genai
from mcp.server import Server
from mcp.server.models import InitializationOptions
from mcp.server.stdio import stdio_server
from mcp.types import TextContent, Tool

from config import (DEFAULT_MODEL, MAX_CONTEXT_TOKENS, MAX_OUTPUT_TOKENS, __author__, __updated__,
                    __version__)
from tools import AnalyzeTool, DebugIssueTool, ReviewCodeTool, ThinkDeeperTool

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Create the MCP server instance
server: Server = Server("gemini-server")

# Initialize tools
TOOLS = {
    "think_deeper": ThinkDeeperTool(),
    "review_code": ReviewCodeTool(),
    "debug_issue": DebugIssueTool(),
    "analyze": AnalyzeTool(),
}


def configure_gemini():
    """Configure Gemini API with the provided API key"""
    api_key = os.getenv("GEMINI_API_KEY")
    if not api_key:
        raise ValueError(
            "GEMINI_API_KEY environment variable is required. "
            "Please set it with your Gemini API key."
        )
    genai.configure(api_key=api_key)
    logger.info("Gemini API configured successfully")


@server.list_tools()
async def handle_list_tools() -> List[Tool]:
    """List all available tools with verbose descriptions"""
    tools = []

    for tool in TOOLS.values():
        tools.append(
            Tool(
                name=tool.name,
                description=tool.description,
                inputSchema=tool.get_input_schema(),
            )
        )

    # Add utility tools
    tools.extend(
        [
            Tool(
                name="chat",
                description=(
                    "GENERAL CHAT & COLLABORATIVE THINKING - Use Gemini as your thinking partner! "
                    "Perfect for: bouncing ideas during your own analysis, getting second opinions on your plans, "
                    "collaborative brainstorming, validating your checklists and approaches, exploring alternatives. "
                    "Also great for: explanations, comparisons, general development questions. "
                    "Triggers: 'ask gemini', 'brainstorm with gemini', 'get gemini's opinion', 'discuss with gemini', "
                    "'share my thinking with gemini', 'explain', 'what is', 'how do I'."
                ),
                inputSchema={
                    "type": "object",
                    "properties": {
                        "prompt": {
                            "type": "string",
                            "description": "Your question, topic, or current thinking to discuss with Gemini",
                        },
                        "context_files": {
                            "type": "array",
                            "items": {"type": "string"},
                            "description": "Optional files for context",
                        },
                        "temperature": {
                            "type": "number",
                            "description": "Response creativity (0-1, default 0.5)",
                            "minimum": 0,
                            "maximum": 1,
                        },
                    },
                    "required": ["prompt"],
                },
            ),
            Tool(
                name="list_models",
                description=(
                    "LIST AVAILABLE MODELS - Show all Gemini models you can use. "
                    "Lists model names, descriptions, and which one is the default."
                ),
                inputSchema={"type": "object", "properties": {}},
            ),
            Tool(
                name="get_version",
                description=(
                    "VERSION & CONFIGURATION - Get server version, configuration details, "
                    "and list of available tools. Useful for debugging and understanding capabilities."
                ),
                inputSchema={"type": "object", "properties": {}},
            ),
        ]
    )

    return tools


@server.call_tool()
async def handle_call_tool(
    name: str, arguments: Dict[str, Any]
) -> List[TextContent]:
    """Handle tool execution requests"""

    # Handle dynamic tools
    if name in TOOLS:
        tool = TOOLS[name]
        return await tool.execute(arguments)

    # Handle static tools
    elif name == "chat":
        return await handle_chat(arguments)

    elif name == "list_models":
        return await handle_list_models()

    elif name == "get_version":
        return await handle_get_version()

    else:
        return [TextContent(type="text", text=f"Unknown tool: {name}")]


async def handle_chat(arguments: Dict[str, Any]) -> List[TextContent]:
    """Handle general chat requests"""
    from config import TEMPERATURE_BALANCED
    from prompts import CHAT_PROMPT
    from utils import read_files

    prompt = arguments.get("prompt", "")
    context_files = arguments.get("context_files", [])
    temperature = arguments.get("temperature", TEMPERATURE_BALANCED)

    # Build the full prompt with system context
    user_content = prompt
    if context_files:
        file_content, _ = read_files(context_files)
        user_content = f"{prompt}\n\n=== CONTEXT FILES ===\n{file_content}\n=== END CONTEXT ==="
    
    # Combine system prompt with user content
    full_prompt = f"{CHAT_PROMPT}\n\n=== USER REQUEST ===\n{user_content}\n=== END REQUEST ===\n\nPlease provide a thoughtful, comprehensive response:"

    try:
        model = genai.GenerativeModel(
            model_name=DEFAULT_MODEL,
            generation_config={
                "temperature": temperature,
                "max_output_tokens": MAX_OUTPUT_TOKENS,
                "candidate_count": 1,
            },
        )

        response = model.generate_content(full_prompt)

        if response.candidates and response.candidates[0].content.parts:
            text = response.candidates[0].content.parts[0].text
        else:
            text = "Response blocked or incomplete"

        return [TextContent(type="text", text=text)]

    except Exception as e:
        return [TextContent(type="text", text=f"Error in chat: {str(e)}")]


async def handle_list_models() -> List[TextContent]:
    """List available Gemini models"""
    try:
        import json

        models = []

        for model_info in genai.list_models():
            if (
                hasattr(model_info, "supported_generation_methods")
                and "generateContent"
                in model_info.supported_generation_methods
            ):
                models.append(
                    {
                        "name": model_info.name,
                        "display_name": getattr(
                            model_info, "display_name", "Unknown"
                        ),
                        "description": getattr(
                            model_info, "description", "No description"
                        ),
                        "is_default": model_info.name.endswith(DEFAULT_MODEL),
                    }
                )

        return [TextContent(type="text", text=json.dumps(models, indent=2))]

    except Exception as e:
        return [
            TextContent(type="text", text=f"Error listing models: {str(e)}")
        ]


async def handle_get_version() -> List[TextContent]:
    """Get version and configuration information"""
    version_info = {
        "version": __version__,
        "updated": __updated__,
        "author": __author__,
        "default_model": DEFAULT_MODEL,
        "max_context_tokens": f"{MAX_CONTEXT_TOKENS:,}",
        "python_version": f"{sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}",
        "server_started": datetime.now().isoformat(),
        "available_tools": list(TOOLS.keys())
        + ["chat", "list_models", "get_version"],
    }

    text = f"""Gemini MCP Server v{__version__}
Updated: {__updated__}
Author: {__author__}

Configuration:
- Default Model: {DEFAULT_MODEL}
- Max Context: {MAX_CONTEXT_TOKENS:,} tokens
- Python: {version_info['python_version']}
- Started: {version_info['server_started']}

Available Tools:
{chr(10).join(f"  - {tool}" for tool in version_info['available_tools'])}

For updates, visit: https://github.com/BeehiveInnovations/gemini-mcp-server"""

    return [TextContent(type="text", text=text)]


async def main():
    """Main entry point for the server"""
    # Configure Gemini API
    configure_gemini()

    # Run the server using stdio transport
    async with stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            InitializationOptions(
                server_name="gemini",
                server_version=__version__,
                capabilities={"tools": {}},
            ),
        )


if __name__ == "__main__":
    asyncio.run(main())