gitingest/src/mcp_server/main.py

"""Main module for the MCP server application."""

from __future__ import annotations

import asyncio
import logging
import os
from typing import Any

from mcp.server.fastmcp import FastMCP

from gitingest.entrypoint import ingest_async
from gitingest.utils.logging_config import get_logger

# Initialize logger for this module
logger = get_logger(__name__)

# Create the FastMCP server instance
mcp = FastMCP("gitingest")

@mcp.tool()
async def ingest_repository(
    source: str,
    max_file_size: int = 10485760,
    include_patterns: list[str] | None = None,
    exclude_patterns: list[str] | None = None,
    branch: str | None = None,
    include_gitignored: bool = False,
    include_submodules: bool = False,
    token: str | None = None,
) -> str:
    """Ingest a Git repository or local directory and return a structured digest for LLMs.
    
    Args:
        source: Git repository URL or local directory path
        max_file_size: Maximum file size to process in bytes (default: 10MB)
        include_patterns: Shell-style patterns to include files
        exclude_patterns: Shell-style patterns to exclude files  
        branch: Git branch to clone and ingest
        include_gitignored: Include files matched by .gitignore
        include_submodules: Include repository's submodules
        token: GitHub personal access token for private repositories
    """
    try:
        logger.info("Starting MCP ingestion", extra={"source": source})

        # Convert patterns to sets if provided
        include_patterns_set = set(include_patterns) if include_patterns else None
        exclude_patterns_set = set(exclude_patterns) if exclude_patterns else None

        # Call the ingestion function
        summary, tree, content = await ingest_async(
            source=source,
            max_file_size=max_file_size,
            include_patterns=include_patterns_set,
            exclude_patterns=exclude_patterns_set,
            branch=branch,
            include_gitignored=include_gitignored,
            include_submodules=include_submodules,
            token=token,
            output=None  # Don't write to file, return content instead
        )

        # Create a structured response
        response_content = f"""# Repository Analysis: {source}

## Summary
{summary}

## File Structure
```
{tree}
```

## Content
{content}

---
*Generated by Gitingest MCP Server*
"""

        return response_content

    except Exception as e:
        logger.error(f"Error during ingestion: {e}", exc_info=True)
        return f"Error ingesting repository: {str(e)}"



async def start_mcp_server_tcp(host: str = "0.0.0.0", port: int = 8001):
    """Start the MCP server with HTTP transport using SSE."""
    logger.info(f"Starting Gitingest MCP server with HTTP/SSE transport on {host}:{port}")
    
    import uvicorn
    from fastapi import FastAPI, Request, HTTPException
    from fastapi.responses import StreamingResponse, JSONResponse
    from fastapi.middleware.cors import CORSMiddleware
    import json
    import asyncio
    from typing import AsyncGenerator
    
    tcp_app = FastAPI(title="Gitingest MCP Server", description="MCP server over HTTP/SSE")
    
    # Add CORS middleware for remote access
    tcp_app.add_middleware(
        CORSMiddleware,
        allow_origins=["*"],  # In production, specify allowed origins
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
    )
    
    @tcp_app.get("/health")
    async def health_check():
        """Health check endpoint."""
        return {"status": "healthy", "transport": "http", "version": "1.0"}
    
    @tcp_app.post("/message")
    async def handle_message(message: dict):
        """Handle MCP messages via HTTP POST."""
        try:
            logger.info(f"Received MCP message: {message}")
            
            # Handle different MCP message types
            if message.get("method") == "initialize":
                return JSONResponse({
                    "jsonrpc": "2.0",
                    "id": message.get("id"),
                    "result": {
                        "protocolVersion": "2024-11-05",
                        "capabilities": {
                            "tools": {}
                        },
                        "serverInfo": {
                            "name": "gitingest",
                            "version": "1.0.0"
                        }
                    }
                })
            
            elif message.get("method") == "tools/list":
                return JSONResponse({
                    "jsonrpc": "2.0", 
                    "id": message.get("id"),
                    "result": {
                        "tools": [{
                            "name": "ingest_repository",
                            "description": "Ingest a Git repository or local directory and return a structured digest for LLMs",
                            "inputSchema": {
                                "type": "object",
                                "properties": {
                                    "source": {
                                        "type": "string",
                                        "description": "Git repository URL or local directory path"
                                    },
                                    "max_file_size": {
                                        "type": "integer", 
                                        "description": "Maximum file size to process in bytes",
                                        "default": 10485760
                                    }
                                },
                                "required": ["source"]
                            }
                        }]
                    }
                })
            
            elif message.get("method") == "tools/call":
                tool_name = message.get("params", {}).get("name")
                arguments = message.get("params", {}).get("arguments", {})
                
                if tool_name == "ingest_repository":
                    try:
                        result = await ingest_repository(**arguments)
                        return JSONResponse({
                            "jsonrpc": "2.0",
                            "id": message.get("id"), 
                            "result": {
                                "content": [{"type": "text", "text": result}]
                            }
                        })
                    except Exception as e:
                        return JSONResponse({
                            "jsonrpc": "2.0",
                            "id": message.get("id"),
                            "error": {
                                "code": -32603,
                                "message": f"Tool execution failed: {str(e)}"
                            }
                        })
                
                else:
                    return JSONResponse({
                        "jsonrpc": "2.0",
                        "id": message.get("id"),
                        "error": {
                            "code": -32601,
                            "message": f"Unknown tool: {tool_name}"
                        }
                    })
            
            else:
                return JSONResponse({
                    "jsonrpc": "2.0", 
                    "id": message.get("id"),
                    "error": {
                        "code": -32601,
                        "message": f"Unknown method: {message.get('method')}"
                    }
                })
                
        except Exception as e:
            logger.error(f"Error handling MCP message: {e}", exc_info=True)
            return JSONResponse({
                "jsonrpc": "2.0",
                "id": message.get("id") if "message" in locals() else None,
                "error": {
                    "code": -32603,
                    "message": f"Internal error: {str(e)}"
                }
            })
    
    # Start the HTTP server
    config = uvicorn.Config(
        tcp_app,
        host=host,
        port=port,
        log_config=None,  # Use our logging config
        access_log=False
    )
    server = uvicorn.Server(config)
    await server.serve()