Bitbucket MCP Server

#!/usr/bin/env python3 """ Bitbucket MCP Server - Manage Pull Requests on Bitbucket Cloud Tools: - setup_bitbucket: Configure Bitbucket credentials - get_config_status: Check configuration status - list_workspace_members: List workspace members (for finding reviewers) - get_default_reviewers: Get default reviewers for a repository - create_pull_request: Create a new PR (with default reviewers support) - list_pull_requests: List PRs for a repository - get_pull_request: View details of a specific PR - update_pull_request: Edit an existing PR (including reviewers) - approve_pull_request: Approve a PR - unapprove_pull_request: Remove approval from a PR - request_changes_pull_request: Request changes on a PR - add_pull_request_comment: Add a comment to a PR """ import json import os import stat from pathlib import Path from typing import Optional import httpx from fastmcp import FastMCP # Initialize FastMCP server mcp = FastMCP("bitbucket") # Config file location CONFIG_DIR = Path.home() / ".bitbucket-mcp" CONFIG_FILE = CONFIG_DIR / "config.json" # Bitbucket API base URL BITBUCKET_API = "https://api.bitbucket.org/2.0" def load_config() -> dict: """Load configuration from file.""" if CONFIG_FILE.exists(): with open(CONFIG_FILE, "r") as f: return json.load(f) return {} def save_config(config: dict) -> None: """Save configuration to file with secure permissions.""" CONFIG_DIR.mkdir(parents=True, exist_ok=True) with open(CONFIG_FILE, "w") as f: json.dump(config, f, indent=2) # Set file permissions to 600 (owner read/write only) os.chmod(CONFIG_FILE, stat.S_IRUSR | stat.S_IWUSR) def get_auth(token: Optional[str] = None, username: Optional[str] = None) -> httpx.BasicAuth: """Get Basic Auth for Bitbucket API (Atlassian API Token).""" config = load_config() auth_token = token or config.get("token") or os.environ.get("BITBUCKET_API_TOKEN") auth_user = username or config.get("username") or os.environ.get("BITBUCKET_USERNAME") if not auth_token or not auth_user: raise ValueError("No Bitbucket credentials configured. Please run setup_bitbucket first.") return httpx.BasicAuth(auth_user, auth_token) def get_workspace(workspace: Optional[str] = None) -> str: """Get workspace from parameter, config, or environment.""" config = load_config() ws = workspace or config.get("workspace") or os.environ.get("BITBUCKET_WORKSPACE") if not ws: raise ValueError("No workspace specified. Please provide workspace parameter or run setup_bitbucket.") return ws @mcp.tool() def setup_bitbucket(workspace: str, username: str, api_token: str) -> str: """ Configure Bitbucket credentials for the MCP server. Args: workspace: Bitbucket workspace slug (e.g., "myworkspace") username: Your Atlassian account email (e.g., "user@example.com") api_token: Atlassian API Token from https://id.atlassian.com/manage-profile/security/api-tokens Returns: Success or error message """ try: # Validate credentials by making a test API call with Basic Auth auth = httpx.BasicAuth(username, api_token) with httpx.Client() as client: # First try to get user info to validate token user_response = client.get( f"{BITBUCKET_API}/user", auth=auth, timeout=30.0 ) if user_response.status_code == 401: return "Error: Invalid credentials. Please check your email and API token." # Then validate workspace access ws_response = client.get( f"{BITBUCKET_API}/workspaces/{workspace}", auth=auth, timeout=30.0 ) if ws_response.status_code == 404: return f"Error: Workspace '{workspace}' not found or you don't have access to it." elif ws_response.status_code == 403: return f"Error: No permission to access workspace '{workspace}'." elif ws_response.status_code != 200: return f"Error: Failed to validate workspace. Status: {ws_response.status_code} - {ws_response.text}" # Save configuration config = { "workspace": workspace, "username": username, "token": api_token } save_config(config) return f"Successfully configured Bitbucket for workspace '{workspace}'. Configuration saved to {CONFIG_FILE}" except httpx.RequestError as e: return f"Error: Failed to connect to Bitbucket API: {str(e)}" except Exception as e: return f"Error: {str(e)}" @mcp.tool() def get_config_status() -> dict: """ Check if Bitbucket is configured and return current status. Returns: Configuration status with workspace info """ config = load_config() env_token = os.environ.get("BITBUCKET_API_TOKEN") env_workspace = os.environ.get("BITBUCKET_WORKSPACE") env_username = os.environ.get("BITBUCKET_USERNAME") has_token = bool(config.get("token") or env_token) has_username = bool(config.get("username") or env_username) configured = has_token and has_username workspace = config.get("workspace") or env_workspace username = config.get("username") or env_username return { "configured": configured, "workspace": workspace, "username": username, "config_file": str(CONFIG_FILE), "source": "config_file" if config.get("token") else ("environment" if env_token else None) } @mcp.tool() def list_workspace_members( workspace: Optional[str] = None, page: int = 1, pagelen: int = 50 ) -> dict: """ List members of a Bitbucket workspace. Use this to find users who can be added as reviewers. Args: workspace: Bitbucket workspace (optional if configured) page: Page number for pagination (default: 1) pagelen: Number of results per page, max 100 (default: 50) Returns: List of workspace members with their UUIDs and display names """ try: ws = get_workspace(workspace) auth = get_auth() params = { "page": page, "pagelen": min(pagelen, 100) } with httpx.Client() as client: response = client.get( f"{BITBUCKET_API}/workspaces/{ws}/members", auth=auth, params=params, timeout=30.0 ) if response.status_code == 200: data = response.json() members = [] for member in data.get("values", []): user = member.get("user", {}) members.append({ "uuid": user.get("uuid"), "account_id": user.get("account_id"), "display_name": user.get("display_name"), "nickname": user.get("nickname") }) return { "success": True, "members": members, "total": data.get("size", len(members)), "page": data.get("page", page), "pagelen": data.get("pagelen", pagelen) } elif response.status_code == 401: return {"success": False, "error": "Authentication failed. Please reconfigure with setup_bitbucket."} elif response.status_code == 404: return {"success": False, "error": f"Workspace '{ws}' not found."} else: return {"success": False, "error": f"API error: {response.status_code} - {response.text}"} except ValueError as e: return {"success": False, "error": str(e)} except httpx.RequestError as e: return {"success": False, "error": f"Connection error: {str(e)}"} except Exception as e: return {"success": False, "error": f"Unexpected error: {str(e)}"} @mcp.tool() def get_default_reviewers( repo_slug: str, workspace: Optional[str] = None ) -> dict: """ Get the default reviewers configured for a repository. Args: repo_slug: Repository slug (name) workspace: Bitbucket workspace (optional if configured) Returns: List of default reviewers with their UUIDs and display names """ try: ws = get_workspace(workspace) auth = get_auth() with httpx.Client() as client: response = client.get( f"{BITBUCKET_API}/repositories/{ws}/{repo_slug}/default-reviewers", auth=auth, timeout=30.0 ) if response.status_code == 200: data = response.json() reviewers = [] for reviewer in data.get("values", []): reviewers.append({ "uuid": reviewer.get("uuid"), "account_id": reviewer.get("account_id"), "display_name": reviewer.get("display_name"), "nickname": reviewer.get("nickname") }) return { "success": True, "default_reviewers": reviewers, "count": len(reviewers) } elif response.status_code == 401: return {"success": False, "error": "Authentication failed. Please reconfigure with setup_bitbucket."} elif response.status_code == 404: return {"success": False, "error": f"Repository '{ws}/{repo_slug}' not found."} else: return {"success": False, "error": f"API error: {response.status_code} - {response.text}"} except ValueError as e: return {"success": False, "error": str(e)} except httpx.RequestError as e: return {"success": False, "error": f"Connection error: {str(e)}"} except Exception as e: return {"success": False, "error": f"Unexpected error: {str(e)}"} @mcp.tool() def create_pull_request( repo_slug: str, title: str, source_branch: str, destination_branch: str = "main", description: str = "", reviewers: Optional[list[str]] = None, use_default_reviewers: bool = True, workspace: Optional[str] = None ) -> dict: """ Create a Pull Request on Bitbucket Cloud. Args: repo_slug: Repository slug (name) title: PR title source_branch: Branch to merge from destination_branch: Branch to merge into (default: main) description: PR description (optional) reviewers: List of reviewer UUIDs or account_ids (optional). These are added in addition to default reviewers. use_default_reviewers: Whether to include default reviewers (default: True) workspace: Bitbucket workspace (optional if configured) Returns: PR details including URL and ID, or error message """ try: ws = get_workspace(workspace) auth = get_auth() payload = { "title": title, "source": { "branch": { "name": source_branch } }, "destination": { "branch": { "name": destination_branch } } } if description: payload["description"] = description # Collect reviewers reviewer_list = [] # Add default reviewers if requested if use_default_reviewers: with httpx.Client() as client: default_response = client.get( f"{BITBUCKET_API}/repositories/{ws}/{repo_slug}/default-reviewers", auth=auth, timeout=30.0 ) if default_response.status_code == 200: default_data = default_response.json() for reviewer in default_data.get("values", []): reviewer_list.append({"uuid": reviewer.get("uuid")}) # Add explicitly specified reviewers if reviewers: for r in reviewers: if r.startswith("{"): reviewer_list.append({"uuid": r}) else: reviewer_list.append({"account_id": r}) # Remove duplicates by uuid seen_uuids = set() unique_reviewers = [] for r in reviewer_list: uuid = r.get("uuid") or r.get("account_id") if uuid and uuid not in seen_uuids: seen_uuids.add(uuid) unique_reviewers.append(r) if unique_reviewers: payload["reviewers"] = unique_reviewers with httpx.Client() as client: response = client.post( f"{BITBUCKET_API}/repositories/{ws}/{repo_slug}/pullrequests", auth=auth, json=payload, timeout=30.0 ) if response.status_code == 201: data = response.json() pr_reviewers = [ {"display_name": r.get("display_name"), "uuid": r.get("uuid")} for r in data.get("reviewers", []) ] return { "success": True, "id": data.get("id"), "title": data.get("title"), "url": data.get("links", {}).get("html", {}).get("href"), "state": data.get("state"), "source_branch": source_branch, "destination_branch": destination_branch, "reviewers": pr_reviewers } elif response.status_code == 400: error_data = response.json() error_msg = error_data.get("error", {}).get("message", "Bad request") return {"success": False, "error": error_msg} elif response.status_code == 401: return {"success": False, "error": "Authentication failed. Please reconfigure with setup_bitbucket."} elif response.status_code == 404: return {"success": False, "error": f"Repository '{ws}/{repo_slug}' not found."} else: return {"success": False, "error": f"API error: {response.status_code} - {response.text}"} except ValueError as e: return {"success": False, "error": str(e)} except httpx.RequestError as e: return {"success": False, "error": f"Connection error: {str(e)}"} except Exception as e: return {"success": False, "error": f"Unexpected error: {str(e)}"} @mcp.tool() def list_pull_requests( repo_slug: str, state: str = "OPEN", workspace: Optional[str] = None, page: int = 1, pagelen: int = 25 ) -> dict: """ List Pull Requests for a repository on Bitbucket Cloud. Args: repo_slug: Repository slug (name) state: PR state filter - OPEN, MERGED, DECLINED, SUPERSEDED, or ALL (default: OPEN) workspace: Bitbucket workspace (optional if configured) page: Page number for pagination (default: 1) pagelen: Number of results per page, max 50 (default: 25) Returns: List of PRs with their details """ try: ws = get_workspace(workspace) auth = get_auth() params = { "page": page, "pagelen": min(pagelen, 50) } if state.upper() != "ALL": params["state"] = state.upper() with httpx.Client() as client: response = client.get( f"{BITBUCKET_API}/repositories/{ws}/{repo_slug}/pullrequests", auth=auth, params=params, timeout=30.0 ) if response.status_code == 200: data = response.json() prs = [] for pr in data.get("values", []): prs.append({ "id": pr.get("id"), "title": pr.get("title"), "state": pr.get("state"), "author": pr.get("author", {}).get("display_name"), "source_branch": pr.get("source", {}).get("branch", {}).get("name"), "destination_branch": pr.get("destination", {}).get("branch", {}).get("name"), "url": pr.get("links", {}).get("html", {}).get("href"), "created_on": pr.get("created_on"), "updated_on": pr.get("updated_on") }) return { "success": True, "pull_requests": prs, "total": data.get("size", len(prs)), "page": data.get("page", page), "pagelen": data.get("pagelen", pagelen) } elif response.status_code == 401: return {"success": False, "error": "Authentication failed. Please reconfigure with setup_bitbucket."} elif response.status_code == 404: return {"success": False, "error": f"Repository '{ws}/{repo_slug}' not found."} else: return {"success": False, "error": f"API error: {response.status_code} - {response.text}"} except ValueError as e: return {"success": False, "error": str(e)} except httpx.RequestError as e: return {"success": False, "error": f"Connection error: {str(e)}"} except Exception as e: return {"success": False, "error": f"Unexpected error: {str(e)}"} @mcp.tool() def get_pull_request( repo_slug: str, pr_id: int, workspace: Optional[str] = None ) -> dict: """ Get details of a specific Pull Request on Bitbucket Cloud. Args: repo_slug: Repository slug (name) pr_id: Pull Request ID workspace: Bitbucket workspace (optional if configured) Returns: Detailed PR information including description, reviewers, and comments count """ try: ws = get_workspace(workspace) auth = get_auth() with httpx.Client() as client: response = client.get( f"{BITBUCKET_API}/repositories/{ws}/{repo_slug}/pullrequests/{pr_id}", auth=auth, timeout=30.0 ) if response.status_code == 200: pr = response.json() reviewers = [] for reviewer in pr.get("reviewers", []): reviewers.append({ "display_name": reviewer.get("display_name"), "uuid": reviewer.get("uuid"), "approved": reviewer.get("approved", False) }) participants = [] for participant in pr.get("participants", []): participants.append({ "display_name": participant.get("user", {}).get("display_name"), "role": participant.get("role"), "approved": participant.get("approved", False), "state": participant.get("state") }) return { "success": True, "id": pr.get("id"), "title": pr.get("title"), "description": pr.get("description", ""), "state": pr.get("state"), "author": pr.get("author", {}).get("display_name"), "source_branch": pr.get("source", {}).get("branch", {}).get("name"), "destination_branch": pr.get("destination", {}).get("branch", {}).get("name"), "url": pr.get("links", {}).get("html", {}).get("href"), "created_on": pr.get("created_on"), "updated_on": pr.get("updated_on"), "close_source_branch": pr.get("close_source_branch", False), "comment_count": pr.get("comment_count", 0), "task_count": pr.get("task_count", 0), "reviewers": reviewers, "participants": participants, "merge_commit": pr.get("merge_commit", {}).get("hash") if pr.get("merge_commit") else None } elif response.status_code == 401: return {"success": False, "error": "Authentication failed. Please reconfigure with setup_bitbucket."} elif response.status_code == 404: return {"success": False, "error": f"Pull request #{pr_id} not found in '{ws}/{repo_slug}'."} else: return {"success": False, "error": f"API error: {response.status_code} - {response.text}"} except ValueError as e: return {"success": False, "error": str(e)} except httpx.RequestError as e: return {"success": False, "error": f"Connection error: {str(e)}"} except Exception as e: return {"success": False, "error": f"Unexpected error: {str(e)}"} @mcp.tool() def update_pull_request( repo_slug: str, pr_id: int, title: Optional[str] = None, description: Optional[str] = None, destination_branch: Optional[str] = None, reviewers: Optional[list[str]] = None, close_source_branch: Optional[bool] = None, workspace: Optional[str] = None ) -> dict: """ Update/edit an existing Pull Request on Bitbucket Cloud. Args: repo_slug: Repository slug (name) pr_id: Pull Request ID to update title: New PR title (optional) description: New PR description (optional) destination_branch: New destination branch (optional) reviewers: List of reviewer UUIDs or account_ids to set as reviewers (optional). Pass empty list [] to remove all reviewers. close_source_branch: Whether to close source branch on merge (optional) workspace: Bitbucket workspace (optional if configured) Returns: Updated PR details or error message """ try: ws = get_workspace(workspace) auth = get_auth() # Build payload with only provided fields payload = {} if title is not None: payload["title"] = title if description is not None: payload["description"] = description if destination_branch is not None: payload["destination"] = {"branch": {"name": destination_branch}} if reviewers is not None: payload["reviewers"] = [{"uuid": r} if r.startswith("{") else {"account_id": r} for r in reviewers] if close_source_branch is not None: payload["close_source_branch"] = close_source_branch if not payload: return {"success": False, "error": "No update fields provided. Specify at least one of: title, description, destination_branch, reviewers, close_source_branch"} with httpx.Client() as client: response = client.put( f"{BITBUCKET_API}/repositories/{ws}/{repo_slug}/pullrequests/{pr_id}", auth=auth, json=payload, timeout=30.0 ) if response.status_code == 200: pr = response.json() reviewer_list = [ {"display_name": r.get("display_name"), "uuid": r.get("uuid")} for r in pr.get("reviewers", []) ] return { "success": True, "id": pr.get("id"), "title": pr.get("title"), "description": pr.get("description", ""), "state": pr.get("state"), "source_branch": pr.get("source", {}).get("branch", {}).get("name"), "destination_branch": pr.get("destination", {}).get("branch", {}).get("name"), "url": pr.get("links", {}).get("html", {}).get("href"), "close_source_branch": pr.get("close_source_branch", False), "reviewers": reviewer_list, "updated_on": pr.get("updated_on") } elif response.status_code == 400: error_data = response.json() error_msg = error_data.get("error", {}).get("message", "Bad request") return {"success": False, "error": error_msg} elif response.status_code == 401: return {"success": False, "error": "Authentication failed. Please reconfigure with setup_bitbucket."} elif response.status_code == 404: return {"success": False, "error": f"Pull request #{pr_id} not found in '{ws}/{repo_slug}'."} elif response.status_code == 403: return {"success": False, "error": "Permission denied. You may not have write access to this PR."} else: return {"success": False, "error": f"API error: {response.status_code} - {response.text}"} except ValueError as e: return {"success": False, "error": str(e)} except httpx.RequestError as e: return {"success": False, "error": f"Connection error: {str(e)}"} except Exception as e: return {"success": False, "error": f"Unexpected error: {str(e)}"} @mcp.tool() def approve_pull_request( repo_slug: str, pr_id: int, workspace: Optional[str] = None ) -> dict: """ Approve a Pull Request on Bitbucket Cloud. Args: repo_slug: Repository slug (name) pr_id: Pull Request ID to approve workspace: Bitbucket workspace (optional if configured) Returns: Approval confirmation or error message """ try: ws = get_workspace(workspace) auth = get_auth() with httpx.Client() as client: response = client.post( f"{BITBUCKET_API}/repositories/{ws}/{repo_slug}/pullrequests/{pr_id}/approve", auth=auth, timeout=30.0 ) if response.status_code in (200, 201): data = response.json() return { "success": True, "message": f"Pull request #{pr_id} approved successfully", "approved": data.get("approved", True), "user": data.get("user", {}).get("display_name") } elif response.status_code == 401: return {"success": False, "error": "Authentication failed. Please reconfigure with setup_bitbucket."} elif response.status_code == 404: return {"success": False, "error": f"Pull request #{pr_id} not found in '{ws}/{repo_slug}'."} elif response.status_code == 409: return {"success": False, "error": "You have already approved this pull request."} else: return {"success": False, "error": f"API error: {response.status_code} - {response.text}"} except ValueError as e: return {"success": False, "error": str(e)} except httpx.RequestError as e: return {"success": False, "error": f"Connection error: {str(e)}"} except Exception as e: return {"success": False, "error": f"Unexpected error: {str(e)}"} @mcp.tool() def unapprove_pull_request( repo_slug: str, pr_id: int, workspace: Optional[str] = None ) -> dict: """ Remove approval from a Pull Request on Bitbucket Cloud. Args: repo_slug: Repository slug (name) pr_id: Pull Request ID to unapprove workspace: Bitbucket workspace (optional if configured) Returns: Confirmation or error message """ try: ws = get_workspace(workspace) auth = get_auth() with httpx.Client() as client: response = client.delete( f"{BITBUCKET_API}/repositories/{ws}/{repo_slug}/pullrequests/{pr_id}/approve", auth=auth, timeout=30.0 ) if response.status_code in (200, 204): return { "success": True, "message": f"Approval removed from pull request #{pr_id}" } elif response.status_code == 401: return {"success": False, "error": "Authentication failed. Please reconfigure with setup_bitbucket."} elif response.status_code == 404: return {"success": False, "error": f"Pull request #{pr_id} not found or not previously approved."} else: return {"success": False, "error": f"API error: {response.status_code} - {response.text}"} except ValueError as e: return {"success": False, "error": str(e)} except httpx.RequestError as e: return {"success": False, "error": f"Connection error: {str(e)}"} except Exception as e: return {"success": False, "error": f"Unexpected error: {str(e)}"} @mcp.tool() def request_changes_pull_request( repo_slug: str, pr_id: int, workspace: Optional[str] = None ) -> dict: """ Request changes on a Pull Request on Bitbucket Cloud. Args: repo_slug: Repository slug (name) pr_id: Pull Request ID workspace: Bitbucket workspace (optional if configured) Returns: Confirmation or error message """ try: ws = get_workspace(workspace) auth = get_auth() with httpx.Client() as client: response = client.post( f"{BITBUCKET_API}/repositories/{ws}/{repo_slug}/pullrequests/{pr_id}/request-changes", auth=auth, timeout=30.0 ) if response.status_code in (200, 201): data = response.json() return { "success": True, "message": f"Changes requested on pull request #{pr_id}", "user": data.get("user", {}).get("display_name") } elif response.status_code == 401: return {"success": False, "error": "Authentication failed. Please reconfigure with setup_bitbucket."} elif response.status_code == 404: return {"success": False, "error": f"Pull request #{pr_id} not found in '{ws}/{repo_slug}'."} elif response.status_code == 409: return {"success": False, "error": "You have already requested changes on this pull request."} else: return {"success": False, "error": f"API error: {response.status_code} - {response.text}"} except ValueError as e: return {"success": False, "error": str(e)} except httpx.RequestError as e: return {"success": False, "error": f"Connection error: {str(e)}"} except Exception as e: return {"success": False, "error": f"Unexpected error: {str(e)}"} @mcp.tool() def add_pull_request_comment( repo_slug: str, pr_id: int, comment: str, workspace: Optional[str] = None ) -> dict: """ Add a comment to a Pull Request on Bitbucket Cloud. Args: repo_slug: Repository slug (name) pr_id: Pull Request ID comment: Comment text (supports markdown) workspace: Bitbucket workspace (optional if configured) Returns: Comment details or error message """ try: ws = get_workspace(workspace) auth = get_auth() payload = { "content": { "raw": comment } } with httpx.Client() as client: response = client.post( f"{BITBUCKET_API}/repositories/{ws}/{repo_slug}/pullrequests/{pr_id}/comments", auth=auth, json=payload, timeout=30.0 ) if response.status_code in (200, 201): data = response.json() return { "success": True, "id": data.get("id"), "message": f"Comment added to pull request #{pr_id}", "content": data.get("content", {}).get("raw"), "user": data.get("user", {}).get("display_name"), "created_on": data.get("created_on") } elif response.status_code == 401: return {"success": False, "error": "Authentication failed. Please reconfigure with setup_bitbucket."} elif response.status_code == 404: return {"success": False, "error": f"Pull request #{pr_id} not found in '{ws}/{repo_slug}'."} else: return {"success": False, "error": f"API error: {response.status_code} - {response.text}"} except ValueError as e: return {"success": False, "error": str(e)} except httpx.RequestError as e: return {"success": False, "error": f"Connection error: {str(e)}"} except Exception as e: return {"success": False, "error": f"Unexpected error: {str(e)}"} if __name__ == "__main__": mcp.run()