"""Commit and build status 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 get_file_contents(
repo_slug: str,
file_path: str,
ref: str = "main",
workspace: Optional[str] = None
) -> dict:
"""
Get the contents of a file from a repository at a specific branch/commit.
Args:
repo_slug: Repository slug (name)
file_path: Path to the file in the repository
ref: Branch name, tag, or commit hash (default: main)
workspace: Bitbucket workspace (optional if configured)
Returns:
File contents as text
"""
try:
ws = get_workspace(workspace)
auth = get_auth()
with httpx.Client() as client:
response = client.get(
f"{BITBUCKET_API}/repositories/{ws}/{repo_slug}/src/{ref}/{file_path}",
auth=auth,
timeout=30.0
)
if response.status_code == 200:
content = response.text
max_size = 100 * 1024
truncated = False
if len(content) > max_size:
content = content[:max_size]
truncated = True
return {
"success": True,
"content": content,
"file_path": file_path,
"ref": ref,
"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:
return {"success": False, "error": f"File '{file_path}' not found at ref '{ref}' 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 list_commits(
repo_slug: str,
branch: Optional[str] = None,
include: Optional[str] = None,
exclude: Optional[str] = None,
workspace: Optional[str] = None,
page: int = 1,
pagelen: int = 25
) -> dict:
"""
List commits in a repository.
Args:
repo_slug: Repository slug (name)
branch: Branch name to list commits from (optional)
include: Commit hash to include (optional)
exclude: Commit hash to exclude (optional)
workspace: Bitbucket workspace (optional if configured)
page: Page number for pagination (default: 1)
pagelen: Number of results per page, max 100 (default: 25)
Returns:
List of commits with their details
"""
try:
ws = get_workspace(workspace)
auth = get_auth()
params = {
"page": page,
"pagelen": min(pagelen, 100)
}
if include:
params["include"] = include
if exclude:
params["exclude"] = exclude
url = f"{BITBUCKET_API}/repositories/{ws}/{repo_slug}/commits"
if branch:
url = f"{url}/{branch}"
with httpx.Client() as client:
response = client.get(
url,
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"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_commit(
repo_slug: str,
commit_hash: str,
workspace: Optional[str] = None
) -> dict:
"""
Get details of a specific commit.
Args:
repo_slug: Repository slug (name)
commit_hash: The commit hash
workspace: Bitbucket workspace (optional if configured)
Returns:
Commit details including message, author, date, and parents
"""
try:
ws = get_workspace(workspace)
auth = get_auth()
with httpx.Client() as client:
response = client.get(
f"{BITBUCKET_API}/repositories/{ws}/{repo_slug}/commit/{commit_hash}",
auth=auth,
timeout=30.0
)
if response.status_code == 200:
commit = response.json()
parents = [p.get("hash", "")[:7] for p in commit.get("parents", [])]
return {
"success": True,
"hash": commit.get("hash"),
"hash_short": commit.get("hash", "")[:7],
"message": commit.get("message"),
"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"),
"parents": parents,
"url": commit.get("links", {}).get("html", {}).get("href")
}
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"Commit '{commit_hash}' 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_commit_diff(
repo_slug: str,
spec: str,
workspace: Optional[str] = None
) -> dict:
"""
Get the diff for a commit or commit range.
Args:
repo_slug: Repository slug (name)
spec: Commit hash or range (e.g., "abc123" or "abc123..def456")
workspace: Bitbucket workspace (optional if configured)
Returns:
The diff content as text
"""
try:
ws = get_workspace(workspace)
auth = get_auth()
with httpx.Client(follow_redirects=True) as client:
response = client.get(
f"{BITBUCKET_API}/repositories/{ws}/{repo_slug}/diff/{spec}",
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,
"spec": spec,
"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:
return {"success": False, "error": f"Spec '{spec}' 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_commit_diffstat(
repo_slug: str,
spec: str,
workspace: Optional[str] = None
) -> dict:
"""
Get the diffstat (summary of changes) for a commit or commit range.
Args:
repo_slug: Repository slug (name)
spec: Commit hash or range (e.g., "abc123" or "abc123..def456")
workspace: Bitbucket workspace (optional if configured)
Returns:
List of files changed with line counts
"""
try:
ws = get_workspace(workspace)
auth = get_auth()
files = []
next_url = f"{BITBUCKET_API}/repositories/{ws}/{repo_slug}/diffstat/{spec}"
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"Spec '{spec}' 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 list_commit_statuses(
repo_slug: str,
commit_hash: str,
workspace: Optional[str] = None,
page: int = 1,
pagelen: int = 25
) -> dict:
"""
List build statuses for a specific commit.
Args:
repo_slug: Repository slug (name)
commit_hash: The commit hash
workspace: Bitbucket workspace (optional if configured)
page: Page number for pagination (default: 1)
pagelen: Number of results per page, max 100 (default: 25)
Returns:
List of build statuses for the commit
"""
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}/commit/{commit_hash}/statuses",
auth=auth,
params=params,
timeout=30.0
)
if response.status_code == 200:
data = response.json()
statuses = []
for status in data.get("values", []):
statuses.append({
"state": status.get("state"),
"key": status.get("key"),
"name": status.get("name"),
"description": status.get("description"),
"url": status.get("url"),
"created_on": status.get("created_on"),
"updated_on": status.get("updated_on")
})
return {
"success": True,
"statuses": statuses,
"total": data.get("size", len(statuses)),
"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"Commit '{commit_hash}' 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 create_build_status(
repo_slug: str,
commit_hash: str,
state: str,
key: str,
url: Optional[str] = None,
description: Optional[str] = None,
name: Optional[str] = None,
workspace: Optional[str] = None
) -> dict:
"""
Create a build status for a commit.
Args:
repo_slug: Repository slug (name)
commit_hash: The commit hash
state: Build state - "SUCCESSFUL", "FAILED", "INPROGRESS", or "STOPPED"
key: Unique key for this build status
url: URL to the build results (optional)
description: Description of the build status (optional)
name: Display name for the build (optional)
workspace: Bitbucket workspace (optional if configured)
Returns:
Created build status details or error message
"""
try:
ws = get_workspace(workspace)
auth = get_auth()
payload = {
"state": state.upper(),
"key": key,
}
if url:
payload["url"] = url
if description:
payload["description"] = description
if name:
payload["name"] = name
with httpx.Client() as client:
response = client.post(
f"{BITBUCKET_API}/repositories/{ws}/{repo_slug}/commit/{commit_hash}/statuses/build",
auth=auth,
json=payload,
timeout=30.0
)
if response.status_code in (200, 201):
data = response.json()
return {
"success": True,
"state": data.get("state"),
"key": data.get("key"),
"name": data.get("name"),
"description": data.get("description"),
"url": data.get("url"),
"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"Commit '{commit_hash}' 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 list_commit_comments(
repo_slug: str,
commit_hash: str,
workspace: Optional[str] = None,
page: int = 1,
pagelen: int = 25
) -> dict:
"""
List comments on a specific commit.
Args:
repo_slug: Repository slug (name)
commit_hash: The commit hash
workspace: Bitbucket workspace (optional if configured)
page: Page number for pagination (default: 1)
pagelen: Number of results per page, max 100 (default: 25)
Returns:
List of comments on the commit
"""
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}/commit/{commit_hash}/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")
}
comments.append({
"id": comment.get("id"),
"content": comment.get("content", {}).get("raw", ""),
"author": comment.get("user", {}).get("display_name"),
"created_on": comment.get("created_on"),
"updated_on": comment.get("updated_on"),
"inline": inline_info
})
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"Commit '{commit_hash}' 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)}"}
