create_merge_request
Create a merge request in GitLab to propose code changes. Specify project ID, source and target branches, and a title to initiate the merge process.
Instructions
Create a merge request.
Args:
project_id: GitLab project ID
source_branch: Source branch
target_branch: Target branch
title: MR title
description: MR description (optional)
token: GitLab Personal Access Token (optional)
ctx: MCP context (automatically injected)Input Schema
| Name | Required | Description | Default |
|---|---|---|---|
| project_id | Yes | ||
| source_branch | Yes | ||
| target_branch | Yes | ||
| title | Yes | ||
| description | No | ||
| token | No | ||
| ctx | No |
Output Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- Handler function for the create_merge_request tool. It posts to GitLab API /projects/{project_id}/merge_requests with source_branch, target_branch, title, and description, then returns the created MR's IID, title, and URL.
@mcp.tool() async def create_merge_request(project_id: int, source_branch: str, target_branch: str, title: str, description: str = "", token: str = None, ctx=None) -> str: """Create a merge request. Args: project_id: GitLab project ID source_branch: Source branch target_branch: Target branch title: MR title description: MR description (optional) token: GitLab Personal Access Token (optional) ctx: MCP context (automatically injected) """ data = { "source_branch": source_branch, "target_branch": target_branch, "title": title, "description": description } result = await make_gitlab_request(f"/projects/{project_id}/merge_requests", "POST", data, ctx=ctx, token=token) if isinstance(result, dict) and "error" in result: return f"Error creating merge request: {result['error']}" return f"Merge request created: !{result['iid']} - {result['title']}\nURL: {result['web_url']}" - Input schema/type definitions for create_merge_request: project_id (int), source_branch (str), target_branch (str), title (str), description (str, default ''), token (optional str), ctx (auto-injected context).
@mcp.tool() async def create_merge_request(project_id: int, source_branch: str, target_branch: str, title: str, description: str = "", token: str = None, ctx=None) -> str: """Create a merge request. Args: project_id: GitLab project ID source_branch: Source branch target_branch: Target branch title: MR title description: MR description (optional) token: GitLab Personal Access Token (optional) ctx: MCP context (automatically injected) """ - gitlab_clone_mcp_server/server.py:438-438 (registration)The @mcp.tool() decorator registers create_merge_request as an MCP tool on the FastMCP server instance named 'mcp'.
@mcp.tool() - Helper function that handles the actual HTTP request to the GitLab API. It resolves the auth token (parameter > context headers > env var), builds the URL from GITLAB_URL env var, and makes the HTTP request with proper error handling.
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)}