diff --git a/cookbook/tools/bitbucket_tools.py b/cookbook/tools/bitbucket_tools.py new file mode 100644 index 000000000..03c77218e --- /dev/null +++ b/cookbook/tools/bitbucket_tools.py @@ -0,0 +1,38 @@ +# from ...libs.agno.agno.tools.bitbucket import BitbucketTools +from agno.agent import Agent +from agno.tools.bitbucket import BitbucketTools + +repo_slug = "Your Repository Slug" +workspace = "Your Workspace" + +agent = Agent( + instructions=[ + f"Use your tools to answer questions about the repo {repo_slug} in the {workspace} workspace", + "Do not create any issues or pull requests unless explicitly asked to do so", + ], + tools=[BitbucketTools()], + show_tool_calls=True, +) + +# Example usage: List all the open pull requests +# agent.print_response("List open pull requests", markdown=True) + +# Example usage: Get pull request details +# agent.print_response("Get details of #230", markdown=True) + +# Example usage: Get pull request changes +# agent.print_response("Show changes for #230", markdown=True) + +# Example usage: List open issues. Only works if the repository has issues enabled +# agent.print_response("What is the latest opened issue?", markdown=True) + +# Example usage: Get the repo details +# agent.print_response("Get details of the repository", markdown=True) + +# Example usage: List all the repositories +# agent.print_response("List 5 repositories for this workspace", markdown=True) + +# Example usage: Create a Repo. Needs Admin Repository access when using App Password +# agent.print_response( +# "Create a repo called agent-testing and add description hello", markdown=True +# ) diff --git a/libs/agno/agno/tools/bitbucket.py b/libs/agno/agno/tools/bitbucket.py new file mode 100644 index 000000000..a75cf3ceb --- /dev/null +++ b/libs/agno/agno/tools/bitbucket.py @@ -0,0 +1,353 @@ +import base64 +import json +import os +from typing import Any, Dict, Optional, Union + +import requests + +from agno.tools import Toolkit +from agno.utils.log import logger + + +class BitbucketTools(Toolkit): + """A class that provides tools for interacting with the Bitbucket API.""" + + def __init__( + self, + server_url: Optional[str] = "api.bitbucket.org", + username: Optional[str] = None, + password: Optional[str] = None, + token: Optional[str] = None, + api_version: Optional[str] = "2.0", + ): + """Initializes Bitbucket Tools. + + Args: + server_url (str, optional): The Bitbucket server URL. Defaults to "api.bitbucket.org". + username (str, optional): The username to authenticate with. If not provided, it will take the value of `BITBUCKET_USERNAME` env variable. + password (str, optional): The password to authenticate with. If not provided, it will take the value of `BITBUCKET_PASSWORD` env variable.. + token (str, optional): The token to authenticate with. If not provided, it will take the value of `BITBUCKET_TOKEN` env variable.. + api_version (str, optional): The version of the Bitbucket API to use. Defaults to "2.0". + + Raises: + ValueError: If username and password or token are not provided. + + Example: + ```python + bitbucket = BitbucketTools( + username="your-username", + password="your-password", + server_url="your-server-url", + api_version="2.0" + ) + ``` + """ + super().__init__(name="bitbucket") + + self.server_url = server_url or os.getenv("BITBUCKET_SERVER_URL") + self.username = username or os.getenv("BITBUCKET_USERNAME") + self.password = password or os.getenv("BITBUCKET_PASSWORD") + self.token = token or os.getenv("BITBUCKET_TOKEN") + self.auth_password = self.token or self.password + self.base_url = f"https://{self.server_url}/{api_version}" + + if not (self.username and self.auth_password): + logger.error("Username and password or token are required") + raise ValueError("Username and password or token are required") + + self.headers = {"Accept": "application/json", "Authorization": f"Basic {self._generate_access_token()}"} + + # Register methods + self.register(self.list_repositories) + self.register(self.get_repository) + self.register(self.create_repository) + self.register(self.list_repository_commits) + self.register(self.list_pull_requests) + self.register(self.get_pull_request) + self.register(self.get_pull_request_changes) + self.register(self.list_issues) + self.register(self.list_repository_pipelines) + + def _generate_access_token(self) -> str: + """Generate an access token for Bitbucket API using Basic Auth. + + Returns: + str: The access token. + """ + auth_str = f"{self.username}:{self.auth_password}" + auth_bytes = auth_str.encode("ascii") + auth_base64 = base64.b64encode(auth_bytes).decode("ascii") + return auth_base64 + + def _make_request( + self, + method: str, + endpoint: str, + params: Optional[Dict[str, Any]] = None, + data: Optional[Dict[str, Any]] = None, + ) -> Union[str, Dict[str, Any]]: + """Make a request to Bitbucket API. + + Args: + method (str): The HTTP method to use for the request. + endpoint (str): The API endpoint to make the request to. + params (Dict[str, Any], optional): Query parameters to include in the request. Defaults to None. + data (Dict[str, Any], optional): The payload to send with the request. Defaults to None. + + Returns: + Union[str, Dict[str, Any]]: The response from the API as a string or a dictionary. + """ + url = f"{self.base_url}{endpoint}" + response = requests.request(method, url, headers=self.headers, json=data, params=params) + response.raise_for_status() + encoding_type = response.headers.get("Content-Type", "application/json") + if encoding_type.startswith("application/json"): + return response.json() if response.text else {} + elif encoding_type == "text/plain": + return response.text + + logger.warning(f"Unsupported content type: {encoding_type}") + return {} + + def list_repositories(self, workspace: str, page: int = 1, pagelen: int = 10) -> str: + """ + List repository info for a given workspace. + API Docs: https://developer.atlassian.com/cloud/bitbucket/rest/api-group-repositories/#api-repositories-workspace-get + + Args: + workspace (str): The slug of the workspace where the repository exists. + page (int, optional): The page number to retrieve. Defaults to 1. + pagelen (int, optional): The number of repositories to retrieve per page. Defaults to 10. + + Returns: + str: A JSON string containing repository list. + """ + try: + params = {"page": page, "pagelen": pagelen} + repo = self._make_request("GET", f"/repositories/{workspace}", params=params) + return json.dumps(repo, indent=2) + except Exception as e: + logger.error(f"Error retrieving repository list for workspace {workspace}: {str(e)}") + return json.dumps({"error": str(e)}) + + def get_repository(self, workspace: str, repo_slug: str) -> str: + """ + Retrieves repository information. + API Docs: https://developer.atlassian.com/cloud/bitbucket/rest/api-group-repositories/#api-repositories-workspace-repo-slug-get + + Args: + workspace (str): The slug of the workspace where the repository exists. + repo_slug (str): The slug of the repository to retrieve information for. + + Returns: + str: A JSON string containing repository information. + """ + try: + repo = self._make_request("GET", f"/repositories/{workspace}/{repo_slug}") + return json.dumps(repo, indent=2) + except Exception as e: + logger.error(f"Error retrieving repository information for {repo_slug}: {str(e)}") + return json.dumps({"error": str(e)}) + + def create_repository( + self, + workspace: str, + repo_slug: str, + name: str, + project: Optional[str] = None, + is_private: bool = False, + description: Optional[str] = None, + language: Optional[str] = None, + has_issues: bool = False, + has_wiki: bool = False, + ) -> str: + """ + Creates a new repository in Bitbucket for the given workspace. + API Docs: https://developer.atlassian.com/cloud/bitbucket/rest/api-group-repositories/#api-repositories-workspace-repo-slug-post + + Args: + workspace (str): The slug of the workspace where the repository exists. + repo_slug (str): The slug of the new repository. + name (str): The name of the new repository. + project (str, optional): The key of the project to create the repository in. Defaults to None. If not provided, the repository will be created in the oldest project in the workspace. + is_private (bool, optional): Whether the repository is private. Defaults to False. + description (str, optional): A short description of the repository. Defaults to None. + language (str, optional): The primary language of the repository. Defaults to None. + has_issues (bool, optional): Whether the repository has issues enabled. Defaults to False. + has_wiki (bool, optional): Whether the repository has a wiki enabled. Defaults to False. + + Returns: + str: A JSON string containing repository information. + """ + try: + payload: Dict[str, Any] = { + "name": name, + "scm": "git", + "is_private": is_private, + "description": description, + "language": language, + "has_issues": has_issues, + "has_wiki": has_wiki, + } + if project: + payload["project"] = {"key": project} + repo = self._make_request("POST", f"/repositories/{workspace}/{repo_slug}", data=payload) + return json.dumps(repo, indent=2) + except Exception as e: + logger.error(f"Error creating repository {repo_slug} for {workspace}: {str(e)}") + return json.dumps({"error": str(e)}) + + def list_repository_commits( + self, workspace: str, repo_slug: str, ctx: Optional[str] = None, page: int = 1, pagelen: int = 10 + ) -> str: + """ + Retrieves all commits in a repository. + API Docs: https://developer.atlassian.com/cloud/bitbucket/rest/api-group-commits/#api-repositories-workspace-repo-slug-commits-get + + Note: The underlying API uses cursor based pagination, so refrain from using the page parameter. Multiple API calls need to be made to retrieve commits of next pages. + + Args: + workspace (str): The slug of the workspace where the repository exists. + repo_slug (str): The slug of the repository to retrieve commits for. + ctx (str, optional): The cursor to navigate between pages. Provided by Bitbucket API. Defaults to None. + page (int, optional): The page number to retrieve. Defaults to 1. + pagelen (int, optional): The number of commits to retrieve per page. Defaults to 10. + + Returns: + str: A JSON string containing all commits. + """ + try: + if ctx: + commits = self._make_request( + "GET", f"/repositories/{workspace}/{repo_slug}/commits?ctx={ctx}&page={page}&pagelen={pagelen}" + ) + else: + commits = self._make_request("GET", f"/repositories/{workspace}/{repo_slug}/commits?pagelen={pagelen}") + for i in range(2, page + 1): + next_url = commits["next"] # type: ignore + query_param = next_url.split("?")[1] + commits = self._make_request("GET", f"/repositories/{workspace}/{repo_slug}/commits?{query_param}") + return json.dumps(commits, indent=2) + except Exception as e: + logger.error(f"Error retrieving commits for {repo_slug}: {str(e)}") + return json.dumps({"error": str(e)}) + + def list_pull_requests( + self, workspace: str, repo_slug: str, state: str = "OPEN", page: int = 1, pagelen: int = 10 + ) -> str: + """ + Retrieves all pull requests for a repository. + API Docs: https://developer.atlassian.com/cloud/bitbucket/rest/api-group-pullrequests/#api-repositories-workspace-repo-slug-pullrequests-get + + Args: + workspace (str): The slug of the workspace where the repository exists. + repo_slug (str): The slug of the repository to retrieve pull requests for. + state (str, optional): The state of the pull requests to retrieve. Defaults to "OPEN". Possible values: "OPEN", "MERGED", "DECLINED", "SUPERSEDED". + page (int, optional): The page number to retrieve. Defaults to 1. + pagelen (int, optional): The number of pull requests to retrieve per page. Defaults to 10. + + Returns: + str: A JSON string containing all pull requests. + """ + try: + VALID_STATES = ["OPEN", "MERGED", "DECLINED", "SUPERSEDED"] + if state not in VALID_STATES: + raise ValueError(f"Invalid state: {state}. Valid states are: {', '.join(VALID_STATES)}") + + params = {"state": state, "page": page, "pagelen": pagelen} + pull_requests = self._make_request( + "GET", f"/repositories/{workspace}/{repo_slug}/pullrequests", params=params + ) + return json.dumps(pull_requests, indent=2) + except Exception as e: + logger.error(f"Error retrieving pull requests for {repo_slug}: {str(e)}") + return json.dumps({"error": str(e)}) + + def get_pull_request(self, workspace: str, repo_slug: str, pull_request_id: int) -> str: + """ + Retrieves a pull request for a repository. + API Docs: https://developer.atlassian.com/cloud/bitbucket/rest/api-group-pullrequests/#api-repositories-workspace-repo-slug-pullrequests-pull-request-id-get + + Args: + workspace (str): The slug of the workspace where the repository exists. + repo_slug (str): The slug of the repository to retrieve pull requests for. + pull_request_id (int): The ID of the pull request to retrieve. + + Returns: + str: A JSON string containing the pull request. + """ + try: + pull_requests = self._make_request( + "GET", f"/repositories/{workspace}/{repo_slug}/pullrequests/{pull_request_id}" + ) + return json.dumps(pull_requests, indent=2) + except Exception as e: + logger.error(f"Error retrieving pull requests for {repo_slug}: {str(e)}") + return json.dumps({"error": str(e)}) + + def get_pull_request_changes(self, workspace: str, repo_slug: str, pull_request_id: int) -> str: + """ + Retrieves changes for a pull request in a repository. + API Docs: https://developer.atlassian.com/cloud/bitbucket/rest/api-group-pullrequests/#api-repositories-workspace-repo-slug-pullrequests-pull-request-id-diff-get + + Args: + workspace (str): The slug of the workspace where the repository exists. + repo_slug (str): The slug of the repository to retrieve pull requests for. + pull_request_id (int): The ID of the pull request to retrieve. + + Returns: + str: A markdown string containing the pull request diff. + """ + try: + diff = self._make_request( + "GET", f"/repositories/{workspace}/{repo_slug}/pullrequests/{pull_request_id}/diff" + ) + return f"```\n{diff}\n```" + except Exception as e: + logger.error(f"Error retrieving changes for pull request {pull_request_id} in {repo_slug}: {str(e)}") + return json.dumps({"error": str(e)}) + + def list_issues(self, workspace: str, repo_slug: str, page: int = 1, pagelen: int = 10) -> str: + """ + Retrieves all issues for a repository. + API Docs: https://developer.atlassian.com/cloud/bitbucket/rest/api-group-issue-tracker/#api-repositories-workspace-repo-slug-issues-get + + Args: + workspace (str): The slug of the workspace where the repository exists. + repo_slug (str): The slug of the repository to retrieve issues for. + page (int, optional): The page number to retrieve. Defaults to 1. + pagelen (int, optional): The number of issues to retrieve per page. Defaults to 10. + + Returns: + str: A JSON string containing all issues. + """ + try: + params = {"page": page, "pagelen": pagelen} + issues = self._make_request("GET", f"/repositories/{workspace}/{repo_slug}/issues", params=params) + return json.dumps(issues, indent=2) + except Exception as e: + logger.error(f"Error retrieving issues for {repo_slug}: {str(e)}") + return json.dumps({"error": str(e)}) + + def list_repository_pipelines(self, workspace: str, repo_slug: str, page: int = 1, pagelen: int = 10) -> str: + """ + Retrieves all pipelines for a repository. + API Docs: https://developer.atlassian.com/cloud/bitbucket/rest/api-group-pipelines/#api-repositories-workspace-repo-slug-pipelines-get + + Args: + workspace (str): The slug of the workspace where the repository exists. + repo_slug (str): The slug of the repository to retrieve pipelines for. + page (int, optional): The page number to retrieve. Defaults to 1. + pagelen (int, optional): The number of pipelines to retrieve per page. Defaults to 10. + + Returns: + str: A JSON string containing all pipelines. + """ + try: + pipelines = self._make_request( + "GET", f"/repositories/{workspace}/{repo_slug}/pipelines?page={page}&pagelen={pagelen}" + ) + return json.dumps(pipelines, indent=2) + except Exception as e: + logger.error(f"Error retrieving pipelines for {repo_slug}: {str(e)}") + return json.dumps({"error": str(e)})