"""Workspace and project 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 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 list_workspaces() -> dict:
"""
List all workspaces the authenticated user has access to.
Returns:
List of workspaces with their details
"""
try:
auth = get_auth()
with httpx.Client() as client:
response = client.get(
f"{BITBUCKET_API}/workspaces",
auth=auth,
params={"pagelen": 100},
timeout=30.0
)
if response.status_code == 200:
data = response.json()
workspaces = []
for ws in data.get("values", []):
workspaces.append({
"uuid": ws.get("uuid"),
"slug": ws.get("slug"),
"name": ws.get("name"),
"is_private": ws.get("is_private"),
"created_on": ws.get("created_on"),
"url": ws.get("links", {}).get("html", {}).get("href")
})
return {
"success": True,
"workspaces": workspaces,
"total": data.get("size", len(workspaces))
}
elif response.status_code == 401:
return {"success": False, "error": "Authentication failed. Please reconfigure with setup_bitbucket."}
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_workspace_details(
workspace: Optional[str] = None
) -> dict:
"""
Get detailed information about a workspace.
Args:
workspace: Bitbucket workspace slug (optional if configured)
Returns:
Workspace details
"""
try:
ws = get_workspace(workspace)
auth = get_auth()
with httpx.Client() as client:
response = client.get(
f"{BITBUCKET_API}/workspaces/{ws}",
auth=auth,
timeout=30.0
)
if response.status_code == 200:
data = response.json()
return {
"success": True,
"uuid": data.get("uuid"),
"slug": data.get("slug"),
"name": data.get("name"),
"is_private": data.get("is_private"),
"created_on": data.get("created_on"),
"url": data.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"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 list_workspace_permissions(
workspace: Optional[str] = None,
page: int = 1,
pagelen: int = 50
) -> dict:
"""
List permissions for a workspace.
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 permissions
"""
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}/permissions",
auth=auth,
params=params,
timeout=30.0
)
if response.status_code == 200:
data = response.json()
permissions = []
for perm in data.get("values", []):
user = perm.get("user", {})
permissions.append({
"user_display_name": user.get("display_name"),
"user_uuid": user.get("uuid"),
"permission": perm.get("permission")
})
return {
"success": True,
"permissions": permissions,
"total": data.get("size", len(permissions)),
"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 list_projects(
workspace: Optional[str] = None,
page: int = 1,
pagelen: int = 25
) -> dict:
"""
List projects in a workspace.
Args:
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 projects with their details
"""
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}/projects",
auth=auth,
params=params,
timeout=30.0
)
if response.status_code == 200:
data = response.json()
projects = []
for project in data.get("values", []):
projects.append({
"key": project.get("key"),
"name": project.get("name"),
"description": project.get("description"),
"is_private": project.get("is_private"),
"created_on": project.get("created_on"),
"updated_on": project.get("updated_on"),
"url": project.get("links", {}).get("html", {}).get("href")
})
return {
"success": True,
"projects": projects,
"total": data.get("size", len(projects)),
"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_project(
project_key: str,
workspace: Optional[str] = None
) -> dict:
"""
Get details of a specific project.
Args:
project_key: Project key
workspace: Bitbucket workspace (optional if configured)
Returns:
Project details
"""
try:
ws = get_workspace(workspace)
auth = get_auth()
with httpx.Client() as client:
response = client.get(
f"{BITBUCKET_API}/workspaces/{ws}/projects/{project_key}",
auth=auth,
timeout=30.0
)
if response.status_code == 200:
project = response.json()
return {
"success": True,
"key": project.get("key"),
"name": project.get("name"),
"description": project.get("description"),
"is_private": project.get("is_private"),
"created_on": project.get("created_on"),
"updated_on": project.get("updated_on"),
"owner": project.get("owner", {}).get("display_name"),
"url": project.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"Project '{project_key}' not found in workspace '{ws}'."}
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_project(
name: str,
key: str,
description: Optional[str] = None,
is_private: bool = True,
workspace: Optional[str] = None
) -> dict:
"""
Create a new project in a workspace.
Args:
name: Project name
key: Project key (unique identifier, uppercase letters and numbers)
description: Project description (optional)
is_private: Whether the project is private (default: True)
workspace: Bitbucket workspace (optional if configured)
Returns:
Created project details or error message
"""
try:
ws = get_workspace(workspace)
auth = get_auth()
payload = {
"name": name,
"key": key,
"is_private": is_private
}
if description:
payload["description"] = description
with httpx.Client() as client:
response = client.post(
f"{BITBUCKET_API}/workspaces/{ws}/projects",
auth=auth,
json=payload,
timeout=30.0
)
if response.status_code in (200, 201):
project = response.json()
return {
"success": True,
"key": project.get("key"),
"name": project.get("name"),
"description": project.get("description"),
"is_private": project.get("is_private"),
"url": project.get("links", {}).get("html", {}).get("href")
}
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 == 403:
return {"success": False, "error": "Permission denied. You may not have permission to create projects."}
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 search_code(
query: str,
workspace: Optional[str] = None,
page: int = 1,
pagelen: int = 25
) -> dict:
"""
Search for code across repositories in a workspace.
Args:
query: Search query string
workspace: Bitbucket workspace (optional if configured)
page: Page number for pagination (default: 1)
pagelen: Number of results per page, max 100 (default: 25)
Returns:
Code search results with file paths and matching content
"""
try:
ws = get_workspace(workspace)
auth = get_auth()
params = {
"search_query": query,
"page": page,
"pagelen": min(pagelen, 100)
}
with httpx.Client() as client:
response = client.get(
f"{BITBUCKET_API}/workspaces/{ws}/search/code",
auth=auth,
params=params,
timeout=60.0
)
if response.status_code == 200:
data = response.json()
results = []
for result in data.get("values", []):
file_info = result.get("file", {})
content_matches = []
for match in result.get("content_matches", []):
for line in match.get("lines", []):
content_matches.append({
"line": line.get("line"),
"content": line.get("segments", [{}])[0].get("text", "") if line.get("segments") else ""
})
results.append({
"file_path": file_info.get("path"),
"repository": file_info.get("links", {}).get("self", {}).get("href", "").split("/src/")[0].split("/")[-1] if file_info.get("links") else None,
"content_matches": content_matches[:5]
})
return {
"success": True,
"results": results,
"total": data.get("size", len(results)),
"page": data.get("page", page),
"pagelen": data.get("pagelen", pagelen),
"query": query
}
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)}"}
