"""Snippet 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_snippets(
workspace: Optional[str] = None,
page: int = 1,
pagelen: int = 25
) -> dict:
"""
List snippets 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 snippets 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}/snippets/{ws}",
auth=auth,
params=params,
timeout=30.0
)
if response.status_code == 200:
data = response.json()
snippets = []
for snippet in data.get("values", []):
snippets.append({
"id": snippet.get("id"),
"title": snippet.get("title"),
"is_private": snippet.get("is_private"),
"created_on": snippet.get("created_on"),
"updated_on": snippet.get("updated_on"),
"owner": snippet.get("owner", {}).get("display_name"),
"url": snippet.get("links", {}).get("html", {}).get("href")
})
return {
"success": True,
"snippets": snippets,
"total": data.get("size", len(snippets)),
"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 create_snippet(
title: str,
content: str,
filename: str,
is_private: bool = False,
workspace: Optional[str] = None
) -> dict:
"""
Create a new snippet.
Args:
title: Snippet title
content: Snippet file content
filename: Filename for the snippet content
is_private: Whether the snippet is private (default: False)
workspace: Bitbucket workspace (optional if configured)
Returns:
Created snippet details or error message
"""
try:
ws = get_workspace(workspace)
auth = get_auth()
# Use multipart form data for snippet creation
files = {
"file": (filename, content),
}
data = {
"title": title,
"is_private": str(is_private).lower(),
}
with httpx.Client() as client:
response = client.post(
f"{BITBUCKET_API}/snippets/{ws}",
auth=auth,
data=data,
files=files,
timeout=30.0
)
if response.status_code in (200, 201):
snippet = response.json()
return {
"success": True,
"id": snippet.get("id"),
"title": snippet.get("title"),
"is_private": snippet.get("is_private"),
"url": snippet.get("links", {}).get("html", {}).get("href"),
"message": "Snippet created successfully."
}
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_snippet(
snippet_id: str,
workspace: Optional[str] = None
) -> dict:
"""
Get details of a specific snippet.
Args:
snippet_id: Snippet ID
workspace: Bitbucket workspace (optional if configured)
Returns:
Snippet details including files and metadata
"""
try:
ws = get_workspace(workspace)
auth = get_auth()
with httpx.Client() as client:
response = client.get(
f"{BITBUCKET_API}/snippets/{ws}/{snippet_id}",
auth=auth,
timeout=30.0
)
if response.status_code == 200:
snippet = response.json()
files = []
for filename, file_info in snippet.get("files", {}).items():
files.append({
"filename": filename,
"size": file_info.get("size"),
"url": file_info.get("links", {}).get("self", {}).get("href")
})
return {
"success": True,
"id": snippet.get("id"),
"title": snippet.get("title"),
"is_private": snippet.get("is_private"),
"owner": snippet.get("owner", {}).get("display_name"),
"created_on": snippet.get("created_on"),
"updated_on": snippet.get("updated_on"),
"files": files,
"url": snippet.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"Snippet '{snippet_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 delete_snippet(
snippet_id: str,
workspace: Optional[str] = None
) -> dict:
"""
Delete a snippet.
Args:
snippet_id: Snippet 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.delete(
f"{BITBUCKET_API}/snippets/{ws}/{snippet_id}",
auth=auth,
timeout=30.0
)
if response.status_code == 204:
return {
"success": True,
"message": f"Snippet '{snippet_id}' 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 can only delete your own snippets."}
elif response.status_code == 404:
return {"success": False, "error": f"Snippet '{snippet_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)}"}
