cursor_tools.py

"""
GitLab Tools for Cursor IDE
Wrapper functions for easy integration with Cursor.
Compatible with polymcp and GitLab MCP Server.
"""

import os
import asyncio
from typing import Optional, Dict, Any, List
from dotenv import load_dotenv

# Load environment variables
load_dotenv()

# Import the MCP server
from gitlab_mcp_server import create_gitlab_mcp_server_inprocess

# Global server instance
_server = None

def _get_server():
    """Get or create the server instance."""
    global _server
    if _server is None:
        _server = create_gitlab_mcp_server_inprocess(verbose=False)
    return _server

def _run_async(coro):
    """Run async function in sync context."""
    try:
        loop = asyncio.get_event_loop()
        if loop.is_running():
            import concurrent.futures
            with concurrent.futures.ThreadPoolExecutor() as executor:
                future = executor.submit(asyncio.run, coro)
                return future.result()
        else:
            return loop.run_until_complete(coro)
    except RuntimeError:
        return asyncio.run(coro)

def _get_project_id(project_id: Optional[str] = None) -> str:
    """Get project ID from argument or environment."""
    pid = project_id or os.getenv("GITLAB_PROJECT_ID", "")
    if not pid:
        raise ValueError("project_id is required. Set GITLAB_PROJECT_ID in .env or pass it as argument.")
    return pid


# Merge Request Functions

def list_open_merge_requests(project_id: str = None) -> Dict[str, Any]:
    """
    List all open merge requests for a project.
    
    Args:
        project_id: GitLab project ID or path. Uses GITLAB_PROJECT_ID env var if not provided.
    
    Returns:
        Dictionary with merge request list and count.
    
    Example:
        >>> mrs = list_open_merge_requests("mygroup/myproject")
        >>> print(f"Found {mrs['count']} open MRs")
    """
    server = _get_server()
    return _run_async(server.invoke(
        "list_merge_requests",
        {"project_id": _get_project_id(project_id), "state": "opened"}
    ))

def list_all_merge_requests(
    project_id: str = None,
    state: str = "all",
    author: str = None
) -> Dict[str, Any]:
    """
    List merge requests with filters.
    
    Args:
        project_id: GitLab project ID or path.
        state: Filter by state (opened, closed, merged, all).
        author: Filter by author username.
    
    Returns:
        Dictionary with filtered merge request list.
    """
    server = _get_server()
    params = {
        "project_id": _get_project_id(project_id),
        "state": state
    }
    if author:
        params["author_username"] = author
    
    return _run_async(server.invoke("list_merge_requests", params))

def get_merge_request(
    mr_iid: int,
    project_id: str = None,
    include_changes: bool = False,
    include_discussions: bool = False
) -> Dict[str, Any]:
    """
    Get detailed information about a specific merge request.
    
    Args:
        mr_iid: Merge request internal ID (the number shown in GitLab).
        project_id: GitLab project ID or path.
        include_changes: Include file diffs.
        include_discussions: Include comments and discussions.
    
    Returns:
        Dictionary with merge request details.
    
    Example:
        >>> mr = get_merge_request(42, include_changes=True)
        >>> print(mr['merge_request']['title'])
    """
    server = _get_server()
    return _run_async(server.invoke(
        "get_merge_request_details",
        {
            "project_id": _get_project_id(project_id),
            "mr_iid": mr_iid,
            "include_changes": include_changes,
            "include_discussions": include_discussions
        }
    ))

def approve_merge_request(
    mr_iid: int,
    project_id: str = None,
    comment: str = None
) -> Dict[str, Any]:
    """
    Approve a merge request.
    
    Args:
        mr_iid: Merge request internal ID.
        project_id: GitLab project ID or path.
        comment: Optional approval comment.
    
    Returns:
        Approval status.
    """
    server = _get_server()
    params = {
        "project_id": _get_project_id(project_id),
        "mr_iid": mr_iid
    }
    if comment:
        params["approval_comment"] = comment
    
    return _run_async(server.invoke("approve_merge_request", params))

def merge_merge_request(
    mr_iid: int,
    project_id: str = None,
    squash: bool = False,
    remove_source_branch: bool = True,
    message: str = None
) -> Dict[str, Any]:
    """
    Merge a merge request.
    
    Args:
        mr_iid: Merge request internal ID.
        project_id: GitLab project ID or path.
        squash: Squash commits when merging.
        remove_source_branch: Delete source branch after merge.
        message: Custom merge commit message.
    
    Returns:
        Merge status.
    """
    server = _get_server()
    params = {
        "project_id": _get_project_id(project_id),
        "mr_iid": mr_iid,
        "squash": squash,
        "remove_source_branch": remove_source_branch
    }
    if message:
        params["merge_commit_message"] = message
    
    return _run_async(server.invoke("merge_merge_request", params))


# Pipeline and Job Functions

def list_pipeline_jobs(project_id: str = None, pipeline_id: int = None) -> Dict[str, Any]:
    """
    List jobs in a pipeline.
    
    Args:
        project_id: GitLab project ID or path.
        pipeline_id: Specific pipeline ID. Uses latest if not provided.
    
    Returns:
        Dictionary with job list grouped by status.
    
    Example:
        >>> jobs = list_pipeline_jobs()
        >>> print(f"Failed: {jobs['summary']['failed']}, Success: {jobs['summary']['success']}")
    """
    server = _get_server()
    params = {"project_id": _get_project_id(project_id)}
    if pipeline_id:
        params["pipeline_id"] = pipeline_id
    
    return _run_async(server.invoke("list_pipeline_jobs", params))

def analyze_pipeline_failures(project_id: str = None, pipeline_id: int = None) -> Dict[str, Any]:
    """
    Analyze failed jobs in a pipeline and get fix suggestions.
    
    Args:
        project_id: GitLab project ID or path.
        pipeline_id: Specific pipeline ID. Uses latest if not provided.
    
    Returns:
        Dictionary with failure analysis and fix suggestions.
    
    Example:
        >>> analysis = analyze_pipeline_failures()
        >>> for job in analysis['analyses']:
        ...     print(f"{job['job_name']}: {job['suggestions']}")
    """
    server = _get_server()
    params = {
        "project_id": _get_project_id(project_id),
        "suggest_fixes": True
    }
    if pipeline_id:
        params["pipeline_id"] = pipeline_id
    
    return _run_async(server.invoke("analyze_failed_jobs", params))

def get_job_log(job_id: int, project_id: str = None, lines: int = None) -> Dict[str, Any]:
    """
    Get the log output of a specific job.
    
    Args:
        job_id: The job ID.
        project_id: GitLab project ID or path.
        lines: Number of lines to return. Returns all if not specified.
    
    Returns:
        Dictionary with job info and log content.
    """
    server = _get_server()
    params = {
        "project_id": _get_project_id(project_id),
        "job_id": job_id
    }
    if lines:
        params["lines"] = lines
    
    return _run_async(server.invoke("get_job_log", params))

def retry_job(job_id: int, project_id: str = None) -> Dict[str, Any]:
    """
    Retry a failed job.
    
    Args:
        job_id: The job ID to retry.
        project_id: GitLab project ID or path.
    
    Returns:
        Status of the retry operation with new job details.
    """
    server = _get_server()
    return _run_async(server.invoke(
        "retry_failed_job",
        {
            "project_id": _get_project_id(project_id),
            "job_id": job_id
        }
    ))

def trigger_pipeline(
    project_id: str = None,
    ref: str = "main",
    variables: Dict[str, str] = None
) -> Dict[str, Any]:
    """
    Trigger a new pipeline run.
    
    Args:
        project_id: GitLab project ID or path.
        ref: Branch, tag, or commit to run pipeline on.
        variables: Pipeline variables to set.
    
    Returns:
        Details of the triggered pipeline.
    """
    server = _get_server()
    params = {
        "project_id": _get_project_id(project_id),
        "ref": ref
    }
    if variables:
        params["variables"] = variables
    
    return _run_async(server.invoke("trigger_pipeline", params))


# ADR Functions

def create_architecture_decision(
    title: str,
    context: str,
    decision: str,
    consequences: str,
    status: str = "proposed",
    alternatives: str = None,
    participants: List[str] = None
) -> Dict[str, Any]:
    """
    Create an Architecture Decision Record document.
    
    Args:
        title: Title of the ADR.
        context: Context and problem statement.
        decision: The decision that was made.
        consequences: Positive and negative consequences.
        status: Decision status (proposed, accepted, deprecated, superseded).
        alternatives: Alternative solutions that were considered.
        participants: List of people involved in the decision.
    
    Returns:
        Dictionary with ADR content in markdown format.
    
    Example:
        >>> adr = create_architecture_decision(
        ...     title="Use PostgreSQL for primary database",
        ...     context="We need a reliable relational database",
        ...     decision="We will use PostgreSQL 15",
        ...     consequences="Good performance, need DBA expertise"
        ... )
        >>> print(adr['content'])
    """
    server = _get_server()
    params = {
        "title": title,
        "status": status,
        "context": context,
        "decision": decision,
        "consequences": consequences
    }
    if alternatives:
        params["alternatives"] = alternatives
    if participants:
        params["participants"] = participants
    
    return _run_async(server.invoke("create_adr_document", params))

def commit_adr(
    project_id: str,
    adr_content: str,
    adr_title: str,
    branch: str = "main",
    create_mr: bool = True
) -> Dict[str, Any]:
    """
    Commit an ADR document to a GitLab repository.
    
    Args:
        project_id: GitLab project ID or path.
        adr_content: The ADR markdown content.
        adr_title: Title for the ADR.
        branch: Target branch for the commit.
        create_mr: Create a merge request for the ADR.
    
    Returns:
        Commit and MR details.
    """
    server = _get_server()
    return _run_async(server.invoke(
        "commit_adr_to_gitlab",
        {
            "project_id": _get_project_id(project_id),
            "adr_content": adr_content,
            "adr_title": adr_title,
            "branch": branch,
            "create_mr": create_mr
        }
    ))


# Utility Functions

def get_server_stats() -> Dict[str, Any]:
    """
    Get server execution statistics.
    
    Returns:
        Dictionary with execution count, error count, and success rate.
    """
    server = _get_server()
    return server.get_stats()

def list_available_tools() -> List[str]:
    """
    List all available GitLab tools.
    
    Returns:
        List of tool names.
    """
    server = _get_server()
    result = _run_async(server.list_tools())
    return [tool['name'] for tool in result['tools']]

def show_help():
    """Print available commands and usage examples."""
    print("""
GitLab Tools for Cursor
=======================

Merge Request Commands:
  list_open_merge_requests()           - List open MRs
  list_all_merge_requests(state='all') - List MRs with filters
  get_merge_request(mr_iid)            - Get MR details
  approve_merge_request(mr_iid)        - Approve an MR
  merge_merge_request(mr_iid)          - Merge an MR

Pipeline Commands:
  list_pipeline_jobs()                 - List jobs in latest pipeline
  analyze_pipeline_failures()          - Analyze failed jobs with suggestions
  get_job_log(job_id)                  - Get job log output
  retry_job(job_id)                    - Retry a failed job
  trigger_pipeline(ref='main')         - Trigger new pipeline

ADR Commands:
  create_architecture_decision(...)    - Create ADR document
  commit_adr(...)                       - Commit ADR to GitLab

Utility Commands:
  list_available_tools()               - List all tools
  get_server_stats()                   - Get execution stats
  show_help()                          - Show this help

Configuration:
  Set GITLAB_TOKEN and GITLAB_PROJECT_ID in .env file.

Examples:
  >>> mrs = list_open_merge_requests()
  >>> analysis = analyze_pipeline_failures()
  >>> adr = create_architecture_decision(
  ...     title="Use Redis for caching",
  ...     context="Need faster response times",
  ...     decision="Implement Redis caching layer",
  ...     consequences="Added infrastructure complexity"
  ... )
""")


# Validate configuration on import
_token = os.getenv("GITLAB_TOKEN")
if not _token:
    print("[GitLab Tools] WARNING: GITLAB_TOKEN not set in .env file")
else:
    print("[GitLab Tools] Ready. Type show_help() for available commands")