"""Webhook management 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_webhooks(
repo_slug: str,
workspace: Optional[str] = None,
page: int = 1,
pagelen: int = 25
) -> dict:
"""
List webhooks configured for a repository.
Args:
repo_slug: Repository slug (name)
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 webhooks 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}/repositories/{ws}/{repo_slug}/hooks",
auth=auth,
params=params,
timeout=30.0
)
if response.status_code == 200:
data = response.json()
webhooks = []
for hook in data.get("values", []):
webhooks.append({
"uuid": hook.get("uuid"),
"url": hook.get("url"),
"description": hook.get("description"),
"active": hook.get("active"),
"events": hook.get("events", []),
"created_at": hook.get("created_at")
})
return {
"success": True,
"webhooks": webhooks,
"total": data.get("size", len(webhooks)),
"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 create_webhook(
repo_slug: str,
url: str,
events: list[str],
description: Optional[str] = None,
active: bool = True,
workspace: Optional[str] = None
) -> dict:
"""
Create a webhook for a repository.
Args:
repo_slug: Repository slug (name)
url: The URL to receive webhook events
events: List of events to subscribe to (e.g., ["repo:push", "pullrequest:created"])
description: Webhook description (optional)
active: Whether the webhook is active (default: True)
workspace: Bitbucket workspace (optional if configured)
Returns:
Created webhook details or error message
"""
try:
ws = get_workspace(workspace)
auth = get_auth()
payload = {
"url": url,
"events": events,
"active": active
}
if description:
payload["description"] = description
with httpx.Client() as client:
response = client.post(
f"{BITBUCKET_API}/repositories/{ws}/{repo_slug}/hooks",
auth=auth,
json=payload,
timeout=30.0
)
if response.status_code in (200, 201):
hook = response.json()
return {
"success": True,
"uuid": hook.get("uuid"),
"url": hook.get("url"),
"description": hook.get("description"),
"active": hook.get("active"),
"events": hook.get("events", []),
"message": "Webhook created successfully."
}
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 need admin access to manage webhooks."}
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 delete_webhook(
repo_slug: str,
webhook_uid: str,
workspace: Optional[str] = None
) -> dict:
"""
Delete a webhook from a repository.
Args:
repo_slug: Repository slug (name)
webhook_uid: Webhook UUID (with or without braces)
workspace: Bitbucket workspace (optional if configured)
Returns:
Confirmation or error message
"""
try:
ws = get_workspace(workspace)
auth = get_auth()
if not webhook_uid.startswith("{"):
webhook_uid = f"{{{webhook_uid}}}"
with httpx.Client() as client:
response = client.delete(
f"{BITBUCKET_API}/repositories/{ws}/{repo_slug}/hooks/{webhook_uid}",
auth=auth,
timeout=30.0
)
if response.status_code == 204:
return {
"success": True,
"message": f"Webhook '{webhook_uid}' deleted successfully."
}
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 need admin access to manage webhooks."}
elif response.status_code == 404:
return {"success": False, "error": f"Webhook '{webhook_uid}' 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_workspace_webhooks(
workspace: Optional[str] = None,
page: int = 1,
pagelen: int = 25
) -> dict:
"""
List webhooks configured 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: 25)
Returns:
List of workspace webhooks
"""
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}/hooks",
auth=auth,
params=params,
timeout=30.0
)
if response.status_code == 200:
data = response.json()
webhooks = []
for hook in data.get("values", []):
webhooks.append({
"uuid": hook.get("uuid"),
"url": hook.get("url"),
"description": hook.get("description"),
"active": hook.get("active"),
"events": hook.get("events", []),
"created_at": hook.get("created_at")
})
return {
"success": True,
"webhooks": webhooks,
"total": data.get("size", len(webhooks)),
"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)}"}
