"""Pull Request tools for Bitbucket MCP Server."""
from typing import Optional
import httpx
from bitbucket_mcp.server import mcp, get_auth, get_workspace, BITBUCKET_API
@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
reviewer_list = []
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")})
if reviewers:
for r in reviewers:
if r.startswith("{"):
reviewer_list.append({"uuid": r})
else:
reviewer_list.append({"account_id": r})
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()
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)}"}
@mcp.tool()
def get_pull_request_comments(
repo_slug: str,
pr_id: int,
workspace: Optional[str] = None,
page: int = 1,
pagelen: int = 50
) -> dict:
"""
Get comments from a Pull Request on Bitbucket Cloud.
Args:
repo_slug: Repository slug (name)
pr_id: Pull Request ID
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 comments with their details including content, author, and timestamps
"""
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}/repositories/{ws}/{repo_slug}/pullrequests/{pr_id}/comments",
auth=auth,
params=params,
timeout=30.0
)
if response.status_code == 200:
data = response.json()
comments = []
for comment in data.get("values", []):
inline = comment.get("inline")
inline_info = None
if inline:
inline_info = {
"path": inline.get("path"),
"from_line": inline.get("from"),
"to_line": inline.get("to")
}
parent = comment.get("parent")
parent_id = parent.get("id") if parent else None
comments.append({
"id": comment.get("id"),
"content": comment.get("content", {}).get("raw", ""),
"content_html": comment.get("content", {}).get("html", ""),
"author": comment.get("user", {}).get("display_name"),
"author_uuid": comment.get("user", {}).get("uuid"),
"created_on": comment.get("created_on"),
"updated_on": comment.get("updated_on"),
"deleted": comment.get("deleted", False),
"inline": inline_info,
"parent_id": parent_id
})
return {
"success": True,
"comments": comments,
"total": data.get("size", len(comments)),
"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"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 get_pull_request_diffstat(
repo_slug: str,
pr_id: int,
workspace: Optional[str] = None
) -> dict:
"""
Get a summary of files changed in a Pull Request with line counts.
Args:
repo_slug: Repository slug (name)
pr_id: Pull Request ID
workspace: Bitbucket workspace (optional if configured)
Returns:
List of files with lines added/removed counts
"""
try:
ws = get_workspace(workspace)
auth = get_auth()
files = []
next_url = f"{BITBUCKET_API}/repositories/{ws}/{repo_slug}/pullrequests/{pr_id}/diffstat"
with httpx.Client(follow_redirects=True) as client:
while next_url:
response = client.get(
next_url,
auth=auth,
timeout=30.0
)
if response.status_code == 200:
data = response.json()
for file_stat in data.get("values", []):
old_path = file_stat.get("old", {}).get("path") if file_stat.get("old") else None
new_path = file_stat.get("new", {}).get("path") if file_stat.get("new") else None
files.append({
"path": new_path or old_path,
"old_path": old_path if old_path != new_path else None,
"status": file_stat.get("status"),
"lines_added": file_stat.get("lines_added", 0),
"lines_removed": file_stat.get("lines_removed", 0)
})
next_url = data.get("next")
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}"}
total_added = sum(f["lines_added"] for f in files)
total_removed = sum(f["lines_removed"] for f in files)
return {
"success": True,
"files": files,
"total_files": len(files),
"total_lines_added": total_added,
"total_lines_removed": total_removed
}
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_diff(
repo_slug: str,
pr_id: int,
file_path: Optional[str] = None,
context_lines: int = 3,
workspace: Optional[str] = None
) -> dict:
"""
Get the diff (code changes) for a Pull Request.
Args:
repo_slug: Repository slug (name)
pr_id: Pull Request ID
file_path: Optional specific file path to get diff for (recommended for large PRs)
context_lines: Number of context lines around changes (default: 3)
workspace: Bitbucket workspace (optional if configured)
Returns:
The diff content as text
"""
try:
ws = get_workspace(workspace)
auth = get_auth()
url = f"{BITBUCKET_API}/repositories/{ws}/{repo_slug}/pullrequests/{pr_id}/diff"
if file_path:
url = f"{url}/{file_path}"
with httpx.Client(follow_redirects=True) as client:
response = client.get(
url,
auth=auth,
headers={"Accept": "text/plain"},
timeout=60.0
)
if response.status_code == 200:
diff_content = response.text
max_size = 100 * 1024
truncated = False
if len(diff_content) > max_size:
diff_content = diff_content[:max_size]
truncated = True
return {
"success": True,
"diff": diff_content,
"file_path": file_path,
"truncated": truncated,
"size_bytes": len(response.text)
}
elif response.status_code == 401:
return {"success": False, "error": "Authentication failed. Please reconfigure with setup_bitbucket."}
elif response.status_code == 404:
if file_path:
return {"success": False, "error": f"File '{file_path}' not found in PR #{pr_id}."}
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 merge_pull_request(
repo_slug: str,
pr_id: int,
merge_strategy: str = "merge_commit",
close_source_branch: bool = False,
message: Optional[str] = None,
workspace: Optional[str] = None
) -> dict:
"""
Merge a Pull Request on Bitbucket Cloud.
Args:
repo_slug: Repository slug (name)
pr_id: Pull Request ID to merge
merge_strategy: Merge strategy - "merge_commit", "squash", or "fast_forward" (default: merge_commit)
close_source_branch: Whether to close the source branch after merge (default: False)
message: Custom merge commit message (optional)
workspace: Bitbucket workspace (optional if configured)
Returns:
Merge result with commit details or error message
"""
try:
ws = get_workspace(workspace)
auth = get_auth()
strategy_map = {
"merge_commit": "merge_commit",
"squash": "squash",
"fast_forward": "fast_forward"
}
if merge_strategy not in strategy_map:
return {"success": False, "error": "Invalid merge strategy. Use: merge_commit, squash, or fast_forward"}
payload = {
"type": strategy_map[merge_strategy],
"close_source_branch": close_source_branch
}
if message:
payload["message"] = message
with httpx.Client() as client:
response = client.post(
f"{BITBUCKET_API}/repositories/{ws}/{repo_slug}/pullrequests/{pr_id}/merge",
auth=auth,
json=payload,
timeout=60.0
)
if response.status_code == 200:
data = response.json()
return {
"success": True,
"message": f"Pull request #{pr_id} merged successfully",
"merge_commit": data.get("merge_commit", {}).get("hash"),
"state": data.get("state"),
"closed_source_branch": close_source_branch
}
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:
error_data = response.json()
error_msg = error_data.get("error", {}).get("message", "Merge conflict or PR cannot be merged")
return {"success": False, "error": error_msg}
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 decline_pull_request(
repo_slug: str,
pr_id: int,
reason: Optional[str] = None,
workspace: Optional[str] = None
) -> dict:
"""
Decline/close a Pull Request on Bitbucket Cloud without merging.
Args:
repo_slug: Repository slug (name)
pr_id: Pull Request ID to decline
reason: Optional reason for declining (will be added as a comment)
workspace: Bitbucket workspace (optional if configured)
Returns:
Confirmation or error message
"""
try:
ws = get_workspace(workspace)
auth = get_auth()
if reason:
comment_payload = {
"content": {
"raw": f"**PR Declined:** {reason}"
}
}
with httpx.Client() as client:
client.post(
f"{BITBUCKET_API}/repositories/{ws}/{repo_slug}/pullrequests/{pr_id}/comments",
auth=auth,
json=comment_payload,
timeout=30.0
)
with httpx.Client() as client:
response = client.post(
f"{BITBUCKET_API}/repositories/{ws}/{repo_slug}/pullrequests/{pr_id}/decline",
auth=auth,
timeout=30.0
)
if response.status_code == 200:
data = response.json()
return {
"success": True,
"message": f"Pull request #{pr_id} declined",
"state": data.get("state"),
"reason": reason
}
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": "PR is already merged or declined."}
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 reply_to_comment(
repo_slug: str,
pr_id: int,
comment_id: int,
content: str,
workspace: Optional[str] = None
) -> dict:
"""
Reply to a specific comment on a Pull Request.
Args:
repo_slug: Repository slug (name)
pr_id: Pull Request ID
comment_id: ID of the comment to reply to
content: Reply text (supports markdown)
workspace: Bitbucket workspace (optional if configured)
Returns:
Reply details or error message
"""
try:
ws = get_workspace(workspace)
auth = get_auth()
payload = {
"content": {
"raw": content
},
"parent": {
"id": comment_id
}
}
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"Reply added to comment #{comment_id}",
"content": data.get("content", {}).get("raw"),
"user": data.get("user", {}).get("display_name"),
"created_on": data.get("created_on"),
"parent_id": comment_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} or comment #{comment_id} 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 add_inline_comment(
repo_slug: str,
pr_id: int,
file_path: str,
line: int,
content: str,
workspace: Optional[str] = None
) -> dict:
"""
Add an inline comment on a specific line of code in a Pull Request.
Args:
repo_slug: Repository slug (name)
pr_id: Pull Request ID
file_path: Path to the file to comment on
line: Line number to comment on
content: 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": content
},
"inline": {
"path": file_path,
"to": line
}
}
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"Inline comment added to {file_path}:{line}",
"content": data.get("content", {}).get("raw"),
"user": data.get("user", {}).get("display_name"),
"created_on": data.get("created_on"),
"file_path": file_path,
"line": line
}
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} or file '{file_path}' 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 delete_comment(
repo_slug: str,
pr_id: int,
comment_id: int,
workspace: Optional[str] = None
) -> dict:
"""
Delete a comment from a Pull Request.
Args:
repo_slug: Repository slug (name)
pr_id: Pull Request ID
comment_id: ID of the comment to delete
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}/comments/{comment_id}",
auth=auth,
timeout=30.0
)
if response.status_code in (200, 204):
return {
"success": True,
"message": f"Comment #{comment_id} deleted from PR #{pr_id}"
}
elif response.status_code == 401:
return {"success": False, "error": "Authentication failed. Please reconfigure with setup_bitbucket."}
elif response.status_code == 403:
return {"success": False, "error": "Permission denied. You can only delete your own comments."}
elif response.status_code == 404:
return {"success": False, "error": f"Comment #{comment_id} not found in PR #{pr_id}."}
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_activity(
repo_slug: str,
pr_id: int,
workspace: Optional[str] = None,
page: int = 1,
pagelen: int = 50
) -> dict:
"""
Get the activity/timeline of a Pull Request including comments, approvals, and updates.
Args:
repo_slug: Repository slug (name)
pr_id: Pull Request ID
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 activity events on the PR
"""
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}/repositories/{ws}/{repo_slug}/pullrequests/{pr_id}/activity",
auth=auth,
params=params,
timeout=30.0
)
if response.status_code == 200:
data = response.json()
activities = []
for activity in data.get("values", []):
activity_item = {}
if "approval" in activity:
approval = activity["approval"]
activity_item = {
"type": "approval",
"user": approval.get("user", {}).get("display_name"),
"date": approval.get("date")
}
elif "update" in activity:
update = activity["update"]
activity_item = {
"type": "update",
"author": update.get("author", {}).get("display_name"),
"date": update.get("date"),
"state": update.get("state"),
"title": update.get("title"),
"description": update.get("description"),
"changes": update.get("changes")
}
elif "comment" in activity:
comment = activity["comment"]
activity_item = {
"type": "comment",
"id": comment.get("id"),
"user": comment.get("user", {}).get("display_name"),
"content": comment.get("content", {}).get("raw"),
"created_on": comment.get("created_on"),
"inline": comment.get("inline")
}
if activity_item:
activities.append(activity_item)
return {
"success": True,
"activities": activities,
"total": data.get("size", len(activities)),
"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"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 get_pull_request_commits(
repo_slug: str,
pr_id: int,
workspace: Optional[str] = None,
page: int = 1,
pagelen: int = 50
) -> dict:
"""
Get the list of commits in a Pull Request.
Args:
repo_slug: Repository slug (name)
pr_id: Pull Request ID
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 commits in the PR
"""
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}/repositories/{ws}/{repo_slug}/pullrequests/{pr_id}/commits",
auth=auth,
params=params,
timeout=30.0
)
if response.status_code == 200:
data = response.json()
commits = []
for commit in data.get("values", []):
commits.append({
"hash": commit.get("hash"),
"hash_short": commit.get("hash", "")[:7],
"message": commit.get("message"),
"message_summary": commit.get("message", "").split("\n")[0] if commit.get("message") else None,
"author": commit.get("author", {}).get("user", {}).get("display_name") if commit.get("author", {}).get("user") else commit.get("author", {}).get("raw"),
"date": commit.get("date"),
"url": commit.get("links", {}).get("html", {}).get("href")
})
return {
"success": True,
"commits": commits,
"total": data.get("size", len(commits)),
"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"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 add_reviewer(
repo_slug: str,
pr_id: int,
reviewer: str,
workspace: Optional[str] = None
) -> dict:
"""
Add a reviewer to a Pull Request without removing existing reviewers.
Args:
repo_slug: Repository slug (name)
pr_id: Pull Request ID
reviewer: UUID or account_id of the reviewer to add
workspace: Bitbucket workspace (optional if configured)
Returns:
Updated reviewer list or error message
"""
try:
ws = get_workspace(workspace)
auth = get_auth()
with httpx.Client() as client:
get_response = client.get(
f"{BITBUCKET_API}/repositories/{ws}/{repo_slug}/pullrequests/{pr_id}",
auth=auth,
timeout=30.0
)
if get_response.status_code != 200:
return {"success": False, "error": f"Failed to get PR: {get_response.status_code}"}
pr_data = get_response.json()
current_reviewers = pr_data.get("reviewers", [])
reviewer_list = [{"uuid": r.get("uuid")} for r in current_reviewers]
if reviewer.startswith("{"):
new_reviewer = {"uuid": reviewer}
else:
new_reviewer = {"account_id": reviewer}
existing_uuids = {r.get("uuid") for r in reviewer_list}
if reviewer in existing_uuids:
return {"success": False, "error": "User is already a reviewer on this PR."}
reviewer_list.append(new_reviewer)
with httpx.Client() as client:
response = client.put(
f"{BITBUCKET_API}/repositories/{ws}/{repo_slug}/pullrequests/{pr_id}",
auth=auth,
json={"reviewers": reviewer_list},
timeout=30.0
)
if response.status_code == 200:
pr = response.json()
reviewers = [
{"display_name": r.get("display_name"), "uuid": r.get("uuid")}
for r in pr.get("reviewers", [])
]
return {
"success": True,
"message": f"Reviewer added to PR #{pr_id}",
"reviewers": 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"Pull request #{pr_id} 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 remove_reviewer(
repo_slug: str,
pr_id: int,
reviewer: str,
workspace: Optional[str] = None
) -> dict:
"""
Remove a reviewer from a Pull Request.
Args:
repo_slug: Repository slug (name)
pr_id: Pull Request ID
reviewer: UUID or account_id of the reviewer to remove
workspace: Bitbucket workspace (optional if configured)
Returns:
Updated reviewer list or error message
"""
try:
ws = get_workspace(workspace)
auth = get_auth()
with httpx.Client() as client:
get_response = client.get(
f"{BITBUCKET_API}/repositories/{ws}/{repo_slug}/pullrequests/{pr_id}",
auth=auth,
timeout=30.0
)
if get_response.status_code != 200:
return {"success": False, "error": f"Failed to get PR: {get_response.status_code}"}
pr_data = get_response.json()
current_reviewers = pr_data.get("reviewers", [])
reviewer_list = []
found = False
for r in current_reviewers:
if r.get("uuid") == reviewer or r.get("account_id") == reviewer:
found = True
else:
reviewer_list.append({"uuid": r.get("uuid")})
if not found:
return {"success": False, "error": "User is not a reviewer on this PR."}
with httpx.Client() as client:
response = client.put(
f"{BITBUCKET_API}/repositories/{ws}/{repo_slug}/pullrequests/{pr_id}",
auth=auth,
json={"reviewers": reviewer_list},
timeout=30.0
)
if response.status_code == 200:
pr = response.json()
reviewers = [
{"display_name": r.get("display_name"), "uuid": r.get("uuid")}
for r in pr.get("reviewers", [])
]
return {
"success": True,
"message": f"Reviewer removed from PR #{pr_id}",
"reviewers": 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"Pull request #{pr_id} 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_merge_status(
repo_slug: str,
pr_id: int,
workspace: Optional[str] = None
) -> dict:
"""
Check if a Pull Request can be merged and get merge status details.
Args:
repo_slug: Repository slug (name)
pr_id: Pull Request ID
workspace: Bitbucket workspace (optional if configured)
Returns:
Merge status including conflicts, required approvals, and blockers
"""
try:
ws = get_workspace(workspace)
auth = get_auth()
with httpx.Client() as client:
pr_response = client.get(
f"{BITBUCKET_API}/repositories/{ws}/{repo_slug}/pullrequests/{pr_id}",
auth=auth,
timeout=30.0
)
if pr_response.status_code != 200:
if pr_response.status_code == 401:
return {"success": False, "error": "Authentication failed. Please reconfigure with setup_bitbucket."}
elif pr_response.status_code == 404:
return {"success": False, "error": f"Pull request #{pr_id} not found in '{ws}/{repo_slug}'."}
return {"success": False, "error": f"API error: {pr_response.status_code} - {pr_response.text}"}
pr_data = pr_response.json()
with httpx.Client() as client:
diff_response = client.get(
f"{BITBUCKET_API}/repositories/{ws}/{repo_slug}/pullrequests/{pr_id}/diffstat",
auth=auth,
timeout=30.0
)
has_conflicts = False
conflict_files = []
if diff_response.status_code == 200:
diff_data = diff_response.json()
for file_stat in diff_data.get("values", []):
if file_stat.get("status") == "merge conflict":
has_conflicts = True
conflict_files.append(file_stat.get("new", {}).get("path") or file_stat.get("old", {}).get("path"))
participants = pr_data.get("participants", [])
approvals = []
changes_requested = []
for p in participants:
user_name = p.get("user", {}).get("display_name")
if p.get("approved"):
approvals.append(user_name)
if p.get("state") == "changes_requested":
changes_requested.append(user_name)
state = pr_data.get("state")
is_open = state == "OPEN"
blockers = []
if not is_open:
blockers.append(f"PR is {state}, not OPEN")
if has_conflicts:
blockers.append("Has merge conflicts")
if changes_requested:
blockers.append(f"Changes requested by: {', '.join(changes_requested)}")
can_merge = is_open and not has_conflicts and not changes_requested
return {
"success": True,
"can_merge": can_merge,
"state": state,
"has_conflicts": has_conflicts,
"conflict_files": conflict_files if conflict_files else None,
"approvals": approvals,
"approval_count": len(approvals),
"changes_requested_by": changes_requested if changes_requested else None,
"blockers": blockers if blockers else None,
"source_branch": pr_data.get("source", {}).get("branch", {}).get("name"),
"destination_branch": pr_data.get("destination", {}).get("branch", {}).get("name")
}
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)}"}
