get_merge_requests
Retrieve merge requests for a GitLab project by ID, with optional state filter and token. Quickly access MRs for review or management.
Instructions
Get merge requests for a GitLab project.
Args:
project_id: GitLab project ID
state: MR state (opened, closed, merged, all)
token: GitLab Personal Access Token (optional)
ctx: MCP context (automatically injected)Input Schema
| Name | Required | Description | Default |
|---|---|---|---|
| project_id | Yes | ||
| state | No | opened | |
| token | No | ||
| ctx | No |
Output Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- The get_merge_requests tool handler function. It calls the GitLab API /projects/{id}/merge_requests endpoint with an optional state filter (opened, closed, merged, all), limits results to 10 MRs, and returns formatted output with IID, title, state, and author.
@mcp.tool() async def get_merge_requests(project_id: int, state: str = "opened", token: str = None, ctx=None) -> str: """Get merge requests for a GitLab project. Args: project_id: GitLab project ID state: MR state (opened, closed, merged, all) token: GitLab Personal Access Token (optional) ctx: MCP context (automatically injected) """ endpoint = f"/projects/{project_id}/merge_requests?state={state}" data = await make_gitlab_request(endpoint, ctx=ctx, token=token) if isinstance(data, dict) and "error" in data: return f"Error: {data['error']}" if not data: return f"No {state} merge requests found." mrs = [] for mr in data[:10]: # Limit to 10 MRs mrs.append(f"!{mr['iid']}: {mr['title']} - {mr['state']} ({mr['author']['name']})") return "\n".join(mrs) - gitlab_clone_mcp_server/server.py:146-147 (registration)The get_merge_requests function is registered as an MCP tool via the @mcp.tool() decorator from FastMCP.
@mcp.tool() async def get_merge_requests(project_id: int, state: str = "opened", token: str = None, ctx=None) -> str: - The make_gitlab_request helper function handles all GitLab API calls. It manages token priority (explicit > context headers > env var), constructs the URL with the GitLab base URL, and performs HTTP requests with httpx.
async def make_gitlab_request(endpoint: str, method: str = "GET", data: dict = None, ctx=None, token: str = None) -> dict[str, Any] | None: """Make a request to GitLab API with proper error handling.""" # Priority: 1. Explicit token parameter, 2. Context headers, 3. Environment variable # If no explicit token provided, try to get from context if not token and ctx and hasattr(ctx, 'request_context') and ctx.request_context: # Try to get from request headers if hasattr(ctx.request_context, 'headers'): token = ctx.request_context.headers.get('GITLAB_TOKEN') # Fallback to environment variable if not token: token = os.getenv("GITLAB_TOKEN") if not token: return {"error": "GitLab token not provided. Please provide a token parameter, GITLAB_TOKEN in the request headers, or set the environment variable."} # Get GitLab URL (from context or environment) gitlab_url = os.getenv("GITLAB_URL", "https://gitlab.com") headers = { "PRIVATE-TOKEN": token, "Content-Type": "application/json" } url = f"{gitlab_url}/api/v4{endpoint}" async with httpx.AsyncClient() as client: try: if method == "GET": response = await client.get(url, headers=headers, timeout=30.0) elif method == "POST": response = await client.post(url, headers=headers, json=data, timeout=30.0) elif method == "PUT": response = await client.put(url, headers=headers, json=data, timeout=30.0) elif method == "DELETE": response = await client.delete(url, headers=headers, timeout=30.0) response.raise_for_status() return response.json() if response.content else {"success": True} except Exception as e: return {"error": str(e)}