"""User and account 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_current_user() -> dict:
"""
Get the currently authenticated user's profile information.
Returns:
Current user details including display name, UUID, and account info
"""
try:
auth = get_auth()
with httpx.Client() as client:
response = client.get(
f"{BITBUCKET_API}/user",
auth=auth,
timeout=30.0
)
if response.status_code == 200:
user = response.json()
return {
"success": True,
"uuid": user.get("uuid"),
"username": user.get("username"),
"display_name": user.get("display_name"),
"nickname": user.get("nickname"),
"account_id": user.get("account_id"),
"created_on": user.get("created_on"),
"url": user.get("links", {}).get("html", {}).get("href")
}
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_user_profile(
username: str
) -> dict:
"""
Get a user's public profile information.
Args:
username: Bitbucket username or UUID
Returns:
User profile details
"""
try:
auth = get_auth()
with httpx.Client() as client:
response = client.get(
f"{BITBUCKET_API}/users/{username}",
auth=auth,
timeout=30.0
)
if response.status_code == 200:
user = response.json()
return {
"success": True,
"uuid": user.get("uuid"),
"username": user.get("username"),
"display_name": user.get("display_name"),
"nickname": user.get("nickname"),
"account_id": user.get("account_id"),
"created_on": user.get("created_on"),
"url": user.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"User '{username}' 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_user_permissions(
workspace: Optional[str] = None,
page: int = 1,
pagelen: int = 25
) -> dict:
"""
Get the current user's repository permissions.
Args:
workspace: Filter by workspace (optional)
page: Page number for pagination (default: 1)
pagelen: Number of results per page, max 100 (default: 25)
Returns:
List of repositories and the user's permission level for each
"""
try:
auth = get_auth()
params = {
"page": page,
"pagelen": min(pagelen, 100)
}
if workspace:
params["q"] = f'repository.full_name ~ "{workspace}/"'
with httpx.Client() as client:
response = client.get(
f"{BITBUCKET_API}/user/permissions/repositories",
auth=auth,
params=params,
timeout=30.0
)
if response.status_code == 200:
data = response.json()
permissions = []
for perm in data.get("values", []):
repo = perm.get("repository", {})
permissions.append({
"repository": repo.get("full_name"),
"repository_slug": repo.get("slug"),
"permission": perm.get("permission"),
"url": repo.get("links", {}).get("html", {}).get("href")
})
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."}
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)}"}
