"""Issue tracker 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_issues(
repo_slug: str,
query: Optional[str] = None,
state: Optional[str] = None,
priority: Optional[str] = None,
workspace: Optional[str] = None,
page: int = 1,
pagelen: int = 25
) -> dict:
"""
List issues in a repository's issue tracker.
Args:
repo_slug: Repository slug (name)
query: Search query to filter issues (optional)
state: Filter by state - "new", "open", "resolved", "on hold", "invalid", "duplicate", "wontfix", "closed" (optional)
priority: Filter by priority - "trivial", "minor", "major", "critical", "blocker" (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 issues with their details
"""
try:
ws = get_workspace(workspace)
auth = get_auth()
params = {
"page": page,
"pagelen": min(pagelen, 100)
}
q_parts = []
if query:
q_parts.append(f'title ~ "{query}"')
if state:
q_parts.append(f'state = "{state}"')
if priority:
q_parts.append(f'priority = "{priority}"')
if q_parts:
params["q"] = " AND ".join(q_parts)
with httpx.Client() as client:
response = client.get(
f"{BITBUCKET_API}/repositories/{ws}/{repo_slug}/issues",
auth=auth,
params=params,
timeout=30.0
)
if response.status_code == 200:
data = response.json()
issues = []
for issue in data.get("values", []):
issues.append({
"id": issue.get("id"),
"title": issue.get("title"),
"state": issue.get("state"),
"priority": issue.get("priority"),
"kind": issue.get("kind"),
"assignee": issue.get("assignee", {}).get("display_name") if issue.get("assignee") else None,
"reporter": issue.get("reporter", {}).get("display_name") if issue.get("reporter") else None,
"created_on": issue.get("created_on"),
"updated_on": issue.get("updated_on"),
"votes": issue.get("votes"),
"url": issue.get("links", {}).get("html", {}).get("href")
})
return {
"success": True,
"issues": issues,
"total": data.get("size", len(issues)),
"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 or issue tracker not enabled."}
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_issue(
repo_slug: str,
title: str,
content: Optional[str] = None,
kind: str = "bug",
priority: str = "major",
assignee: Optional[str] = None,
workspace: Optional[str] = None
) -> dict:
"""
Create a new issue in a repository's issue tracker.
Args:
repo_slug: Repository slug (name)
title: Issue title
content: Issue description in markdown (optional)
kind: Issue kind - "bug", "enhancement", "proposal", or "task" (default: bug)
priority: Issue priority - "trivial", "minor", "major", "critical", or "blocker" (default: major)
assignee: Account ID of the assignee (optional)
workspace: Bitbucket workspace (optional if configured)
Returns:
Created issue details or error message
"""
try:
ws = get_workspace(workspace)
auth = get_auth()
payload = {
"title": title,
"kind": kind,
"priority": priority,
}
if content:
payload["content"] = {"raw": content}
if assignee:
payload["assignee"] = {"account_id": assignee}
with httpx.Client() as client:
response = client.post(
f"{BITBUCKET_API}/repositories/{ws}/{repo_slug}/issues",
auth=auth,
json=payload,
timeout=30.0
)
if response.status_code in (200, 201):
issue = response.json()
return {
"success": True,
"id": issue.get("id"),
"title": issue.get("title"),
"state": issue.get("state"),
"priority": issue.get("priority"),
"kind": issue.get("kind"),
"url": issue.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 == 403:
return {"success": False, "error": "Permission denied."}
elif response.status_code == 404:
return {"success": False, "error": f"Repository '{ws}/{repo_slug}' not found or issue tracker not enabled."}
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_issue(
repo_slug: str,
issue_id: int,
workspace: Optional[str] = None
) -> dict:
"""
Get details of a specific issue.
Args:
repo_slug: Repository slug (name)
issue_id: Issue ID
workspace: Bitbucket workspace (optional if configured)
Returns:
Issue details including description, assignee, and metadata
"""
try:
ws = get_workspace(workspace)
auth = get_auth()
with httpx.Client() as client:
response = client.get(
f"{BITBUCKET_API}/repositories/{ws}/{repo_slug}/issues/{issue_id}",
auth=auth,
timeout=30.0
)
if response.status_code == 200:
issue = response.json()
return {
"success": True,
"id": issue.get("id"),
"title": issue.get("title"),
"content": issue.get("content", {}).get("raw", ""),
"state": issue.get("state"),
"priority": issue.get("priority"),
"kind": issue.get("kind"),
"assignee": issue.get("assignee", {}).get("display_name") if issue.get("assignee") else None,
"reporter": issue.get("reporter", {}).get("display_name") if issue.get("reporter") else None,
"created_on": issue.get("created_on"),
"updated_on": issue.get("updated_on"),
"votes": issue.get("votes"),
"component": issue.get("component", {}).get("name") if issue.get("component") else None,
"milestone": issue.get("milestone", {}).get("name") if issue.get("milestone") else None,
"version": issue.get("version", {}).get("name") if issue.get("version") else None,
"url": issue.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"Issue #{issue_id} 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 update_issue(
repo_slug: str,
issue_id: int,
title: Optional[str] = None,
content: Optional[str] = None,
state: Optional[str] = None,
priority: Optional[str] = None,
assignee: Optional[str] = None,
kind: Optional[str] = None,
workspace: Optional[str] = None
) -> dict:
"""
Update an existing issue.
Args:
repo_slug: Repository slug (name)
issue_id: Issue ID
title: New issue title (optional)
content: New issue description in markdown (optional)
state: New state - "new", "open", "resolved", "on hold", "invalid", "duplicate", "wontfix", "closed" (optional)
priority: New priority - "trivial", "minor", "major", "critical", "blocker" (optional)
assignee: Account ID of new assignee, or empty string to unassign (optional)
kind: New kind - "bug", "enhancement", "proposal", "task" (optional)
workspace: Bitbucket workspace (optional if configured)
Returns:
Updated issue details or error message
"""
try:
ws = get_workspace(workspace)
auth = get_auth()
payload = {}
if title is not None:
payload["title"] = title
if content is not None:
payload["content"] = {"raw": content}
if state is not None:
payload["state"] = state
if priority is not None:
payload["priority"] = priority
if kind is not None:
payload["kind"] = kind
if assignee is not None:
if assignee == "":
payload["assignee"] = None
else:
payload["assignee"] = {"account_id": assignee}
if not payload:
return {"success": False, "error": "No update fields provided."}
with httpx.Client() as client:
response = client.put(
f"{BITBUCKET_API}/repositories/{ws}/{repo_slug}/issues/{issue_id}",
auth=auth,
json=payload,
timeout=30.0
)
if response.status_code == 200:
issue = response.json()
return {
"success": True,
"id": issue.get("id"),
"title": issue.get("title"),
"state": issue.get("state"),
"priority": issue.get("priority"),
"kind": issue.get("kind"),
"assignee": issue.get("assignee", {}).get("display_name") if issue.get("assignee") else None,
"url": issue.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 == 403:
return {"success": False, "error": "Permission denied."}
elif response.status_code == 404:
return {"success": False, "error": f"Issue #{issue_id} 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_issue_comments(
repo_slug: str,
issue_id: int,
workspace: Optional[str] = None,
page: int = 1,
pagelen: int = 25
) -> dict:
"""
List comments on an issue.
Args:
repo_slug: Repository slug (name)
issue_id: Issue ID
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 issue
"""
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}/issues/{issue_id}/comments",
auth=auth,
params=params,
timeout=30.0
)
if response.status_code == 200:
data = response.json()
comments = []
for comment in data.get("values", []):
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")
})
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"Issue #{issue_id} 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 add_issue_comment(
repo_slug: str,
issue_id: int,
content: str,
workspace: Optional[str] = None
) -> dict:
"""
Add a comment to an issue.
Args:
repo_slug: Repository slug (name)
issue_id: Issue ID
content: Comment text (supports markdown)
workspace: Bitbucket workspace (optional if configured)
Returns:
Created comment details or error message
"""
try:
ws = get_workspace(workspace)
auth = get_auth()
payload = {
"content": {
"raw": content
}
}
with httpx.Client() as client:
response = client.post(
f"{BITBUCKET_API}/repositories/{ws}/{repo_slug}/issues/{issue_id}/comments",
auth=auth,
json=payload,
timeout=30.0
)
if response.status_code in (200, 201):
data = response.json()
return {
"success": True,
"id": data.get("id"),
"content": data.get("content", {}).get("raw"),
"author": data.get("user", {}).get("display_name"),
"created_on": data.get("created_on"),
"message": f"Comment added to issue #{issue_id}"
}
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"Issue #{issue_id} 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)}"}
