#!/usr/bin/env python3
"""
Bitbucket MCP Server - Comprehensive Bitbucket Cloud API Integration
Provides tools for managing pull requests, repositories, branches, commits,
pipelines, workspaces, issues, webhooks, snippets, and more.
"""
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
# Import all tool modules to register them with mcp
import bitbucket_mcp.tools # noqa: E402, F401
def main():
"""Entry point for the MCP server."""
mcp.run()
if __name__ == "__main__":
main()
