From b6d285f14e5349f00b8113fc8e4bce1b3df4f53b Mon Sep 17 00:00:00 2001 From: Said Sef Date: Mon, 21 Jul 2025 17:57:34 +0000 Subject: [PATCH 1/2] feat: functions annottation to dict - mcp now supports structured outputs - https://github.com/modelcontextprotocol/python-sdk/pull/993 --- src/mcp_github/issues_pr_analyser.py | 200 +++++++++++++-------------- 1 file changed, 93 insertions(+), 107 deletions(-) diff --git a/src/mcp_github/issues_pr_analyser.py b/src/mcp_github/issues_pr_analyser.py index cf852a2..73300b1 100644 --- a/src/mcp_github/issues_pr_analyser.py +++ b/src/mcp_github/issues_pr_analyser.py @@ -78,7 +78,7 @@ def _register_tools(self): """Registers the MCP tools for GitHub PR and issue management.""" @self.mcp.tool() - async def get_github_pr_diff(repo_owner: str, repo_name: str, pr_number: int) -> str: + async def get_github_pr_diff(repo_owner: str, repo_name: str, pr_number: int) -> dict[str, Any]: """ Fetches the diff of a specific pull request from a GitHub repository. Use only get_pr_diff to fetch the diff of a specific pull request from a GitHub repository. @@ -87,59 +87,55 @@ async def get_github_pr_diff(repo_owner: str, repo_name: str, pr_number: int) -> repo_name (str): The name of the GitHub repository. pr_number (int): The pull request number to fetch the diff for. Returns: - str: The diff of the pull request as a string. Returns a 'No changes' string if no changes are found, - or an error message string if an exception occurs. - Error Handling: - Logs and returns the error message if an exception is raised during the fetch operation. - The exception traceback is also printed to stderr for debugging purposes. + dict[str, Any]: A dictionary containing the pull request diff. Returns a 'No changes' string if no changes are found, + or an error message string if an exception occurs. """ logging.info(f"Fetching PR #{pr_number} diff from {repo_owner}/{repo_name}") try: pr_diff = self.gi.get_pr_diff(repo_owner, repo_name, pr_number) if pr_diff is None: no_changes = "No changes returned from get_pr_diff" - logging.info(f"{no_changes}") - return f"{no_changes}" - logging.info(f"Successfully fetched PR diff") - return pr_diff + logging.info({"status": "error", "message": no_changes}) + return {"status": "error", "message": no_changes} + logging.info({"status": "success", "message": f"Successfully fetched PR #{pr_number} diff"}) + return {"status": "success", "message": f"{pr_diff}"} except Exception as e: - logging.error(f"Error fetching PR diff: {str(e)}") + logging.error({"status": "error", "message": str(e)}) traceback.print_exc(file=sys.stderr) - return str(e) + return {"status": "error", "message": str(e)} @self.mcp.tool() - async def get_github_pr_content(repo_owner: str, repo_name: str, pr_number: int) -> str: + async def get_github_pr_content(repo_owner: str, repo_name: str, pr_number: int) -> dict[str, Any]: """ Fetches the content of a specific pull request from a GitHub repository. - Use only get_pr_diff to fetch the diff of a specific pull request from a GitHub repository. + Use only get_pr_content to fetch the content of a specific pull request from a GitHub repository. Args: repo_owner (str): The owner of the GitHub repository. repo_name (str): The name of the GitHub repository. pr_number (int): The pull request number to fetch the content for. Returns: - str: The content of the pull request as a string. Returns a 'No changes' string if no changes are found, - or an error message string if an exception occurs. - Error Handling: - Logs and returns the error message if an exception is raised during the fetch operation. + dict[str, Any]: A dictionary containing the pull request content. Returns a 'No changes' string if no changes are found, + or an error message string if an exception occurs. """ logging.info(f"Fetching PR #{pr_number} from {repo_owner}/{repo_name}") try: pr_info = self.gi.get_pr_content(repo_owner, repo_name, pr_number) if pr_info is None: no_changes = "No changes returned from get_pr_content" - logging.info(f"{no_changes}") - return f"{no_changes}" - logging.info(f"Successfully fetched PR information") - return str(pr_info) + logging.info({"status": "error", "message": no_changes}) + return {"status": "error", "message": no_changes} + logging.info({"status": "success", "message": f"Successfully fetched PR #{pr_number} content"}) + return pr_info except Exception as e: - logging.error(f"Error fetching PR: {str(e)}") + logging.error({"status": "error", "message": str(e)}) traceback.print_exc(file=sys.stderr) - return str(e) + return {"status": "error", "message": str(e)} @self.mcp.tool() - async def update_github_pr_description(repo_owner: str, repo_name: str, pr_number: int, new_title: str, new_description: str) -> str: + async def update_github_pr_description(repo_owner: str, repo_name: str, pr_number: int, new_title: str, new_description: str) -> dict[str, Any]: """ - Updates the description of a GitHub pull request with a new title and description. + Updates the title and description of a specific pull request on GitHub. + Use only update_pr_description to update the title and description of a specific pull request on GitHub. Args: repo_owner (str): The owner of the GitHub repository. repo_name (str): The name of the GitHub repository. @@ -147,26 +143,24 @@ async def update_github_pr_description(repo_owner: str, repo_name: str, pr_numbe new_title (str): The new title for the pull request. new_description (str): The new description for the pull request. Returns: - str: A message indicating the result of the update operation. Returns a success message if the update is successful, or an error message if an exception occurs. - Error Handling: - Catches and logs any exceptions that occur during the update process. If an error is encountered, the error message is logged and returned. + dict[str, Any]: A dictionary containing the status and message of the update operation. + Returns a success message if the update is successful, or an error message if an exception occurs """ logging.info(f"Updating PR #{pr_number} description") try: self.gi.update_pr_description(repo_owner, repo_name, pr_number, new_title, new_description) - logging.info(f"Successfully updated PR #{pr_number} description") - return f"Successfully updated PR #{pr_number} description" + logging.info({"status": "success", "message": f"Successfully updated PR #{pr_number} description"}) + return {"status": "success", "message": f"Successfully updated PR #{pr_number} description"} except Exception as e: error_msg = f"Error updating PR description: {str(e)}" logging.error(error_msg) traceback.print_exc(file=sys.stderr) - return error_msg + return {"status": "error", "message": error_msg} @self.mcp.tool() - async def add_github_pr_inline_comment(repo_owner: str, repo_name: str, pr_number: int, path: str, line: int, comment_body: str) -> str: + async def add_github_pr_inline_comment(repo_owner: str, repo_name: str, pr_number: int, path: str, line: int, comment_body: str) -> dict[str, Any]: """ Adds an inline review comment to a specific line in a pull request on GitHub. - Only comment if there are issues with the code, otherwise do not comment. Args: repo_owner (str): The owner of the GitHub repository. repo_name (str): The name of the GitHub repository. @@ -175,206 +169,198 @@ async def add_github_pr_inline_comment(repo_owner: str, repo_name: str, pr_numbe line (int): The line number in the file where the comment should be added. comment_body (str): The content of the comment to be added. Returns: - str: A message indicating the result of the inline comment addition. Returns a success message if the comment is added successfully, or an error message if an exception occurs. - Error Handling: - Catches and logs any exceptions that occur during the inline comment addition process. If an error is encountered, the error message is logged and returned. + dict[str, Any]: A dictionary containing the status and message of the comment addition. + Returns a success message if the comment is added successfully, or an error message if an exception occurs. """ logging.info(f"Adding inline review comment to PR #{pr_number}") try: self.gi.add_inline_pr_comment(repo_owner, repo_name, pr_number, path, line, comment_body) - logging.info(f"Successfully added inline review comment to PR #{pr_number}") - return f"Successfully added inline review comment to PR #{pr_number}" + logging.info({"status": "success", "message": f"Successfully added inline review comment to PR #{pr_number}"}) + return {"status": "success", "message": f"Successfully added inline review comment to PR #{pr_number}"} except Exception as e: error_msg = f"Error adding inline review comment to PR: {str(e)}" logging.error(error_msg) traceback.print_exc(file=sys.stderr) - return error_msg + return {"status": "error", "message": error_msg} @self.mcp.tool() - async def add_github_pr_comment(repo_owner: str, repo_name: str, pr_number: int, comment: str) -> str: + async def add_github_pr_comment(repo_owner: str, repo_name: str, pr_number: int, comment: str) -> dict[str, Any]: """ - Adds a comment to a GitHub pull request. + Adds a comment to a specific pull request on GitHub. Args: repo_owner (str): The owner of the GitHub repository. repo_name (str): The name of the GitHub repository. pr_number (int): The pull request number to add the comment to. comment (str): The content of the comment to be added. Returns: - str: A message indicating the result of the comment addition. Returns a success message if the comment is added successfully, or an error message if an exception occurs. - Error Handling: - Catches and logs any exceptions that occur during the comment addition process. If an error is encountered, the error message is logged and returned. + dict[str, Any]: A dictionary containing the status and message of the comment addition. + Returns a success message if the comment is added successfully, or an error message if an exception occurs. """ logging.info(f"Adding comment to PR #{pr_number}") try: self.gi.add_pr_comments(repo_owner, repo_name, pr_number, comment) - logging.info(f"Successfully added comment to PR #{pr_number}") - return f"Successfully added comment to PR #{pr_number}" + logging.info({"status": "success", "message": f"Successfully added comment to PR #{pr_number}"}) + return {"status": "success", "message": f"Successfully added comment to PR #{pr_number}"} except Exception as e: error_msg = f"Error adding comment to PR: {str(e)}" logging.error(error_msg) traceback.print_exc(file=sys.stderr) - return error_msg + return {"status": "error", "message": error_msg} @self.mcp.tool() - async def list_github_issues_prs(repo_owner: str, issue: str) -> str: + async def list_github_issues_prs(repo_owner: str, issue: str) -> dict[str, Any]: """ - Lists all open pull requests or issues for a given GitHub repository owner. - GitHub /issues/ endpoint is used to list all open issues and pull requests. + Lists open issues or pull requests for a specified GitHub repository. Args: repo_owner (str): The owner of the GitHub repository. - issue (str): The type of items to list, either 'pr' for pull requests or 'issue' for issues. + issue (str): The type of items to list, either 'pr' for pull requests or 'issue' for issues. Defaults to 'pr'. Returns: - str: A message indicating the result of the listing. Returns a success message with the list of open pull requests or issues if successful, or an error message if an exception occurs. - Error Handling: - Catches and logs any exceptions that occur during the listing process. If an error is encountered, the error message is logged and returned. + dict[str, Any]: A dictionary containing the list of open issues or pull requests. + Returns an error message if an exception occurs during the listing process. """ - logging.info(f"Listing open {issue} for {repo_owner}") + logging.info({"status": "info", "message": f"Listing open {issue} for {repo_owner}"}) try: open_issues_prs = self.gi.list_open_issues_prs(repo_owner, issue) - return f"{open_issues_prs}" + return open_issues_prs except Exception as e: error_msg = f"Error listing {issue} for {repo_owner}: {str(e)}" logging.error(error_msg) traceback.print_exc(file=sys.stderr) - return error_msg + return {"status": "error", "message": error_msg} @self.mcp.tool() - async def create_github_issue(repo_owner: str, repo_name: str, title: str, body: str, labels: list[str]) -> str: + async def create_github_issue(repo_owner: str, repo_name: str, title: str, body: str, labels: list[str]) -> dict[str, Any]: """ - Creates a GitHub issue in the specified repository. - Then use add_github_pr_comment to link the issue to the PR with comment "Resolves #". + Creates a GitHub issue with the specified parameters. Args: repo_owner (str): The owner of the GitHub repository. repo_name (str): The name of the GitHub repository. - title (str): The title of the issue to be created, one of chore, fix, bug, feat, docs etc. + title (str): The title of the issue, one of chore, fix, bug, feat, docs etc. body (str): The body content of the issue. labels (list[str]): A list of labels to assign to the issue. Returns: - str: A message indicating the result of the issue creation. Returns a success message if the issue is created successfully, or an error message if an exception occurs. - Error Handling: - Logs and returns an error message if an exception is raised during the issue creation process. The exception traceback is also printed to stderr for debugging purposes. + dict[str, Any]: A dictionary containing the status and message of the issue creation. + Returns a success message if the issue is created successfully, or an error message if an exception occurs. """ logging.info(f"Creating GitHub issue: {title}") try: issue = self.gi.create_issue(repo_owner, repo_name, title, body, labels) - logging.info(f"GitHub issue '{title}' created successfully!") - return f"GitHub issue number #{issue} created successfully!" + logging.info({"status": "success", "message": f"GitHub issue #{issue} created successfully!"}) + return {"status": "success", "message": f"GitHub issue number #{issue} created successfully!"} except Exception as e: error_msg = f"Error creating GitHub issue: {str(e)}" logging.error(error_msg) traceback.print_exc(file=sys.stderr) - return error_msg + return {"status": "error", "message": error_msg} @self.mcp.tool() - async def update_github_issue(repo_owner: str, repo_name: str, issue_number: int, title: str, body: str, labels: list[str], state: str) -> str: + async def update_github_issue(repo_owner: str, repo_name: str, issue_number: int, title: str, body: str, labels: list[str], state: str) -> dict[str, Any]: """ Updates a GitHub issue with the specified parameters. Args: repo_owner (str): The owner of the GitHub repository. repo_name (str): The name of the GitHub repository. - issue_number (int): The number of the issue to update. - title (str): The new title for the issue, one of chore, fix, bug, feat, docs etc. - body (str): The new body content for the issue. + issue_number (int): The issue number to update. + title (str): The new title for the issue. + body (str): The new body content of the issue. labels (list[str]): A list of labels to assign to the issue. - state (str): The state to set for the issue (e.g., 'open', 'closed'). + state (str): The new state of the issue, either 'open' or 'closed'. Returns: - str: A message indicating whether the issue was updated successfully or an error occurred. - Error Handling: - Logs and returns an error message if an exception occurs during the update process. + dict[str, Any]: A dictionary containing the status and message of the issue update. + Returns a success message if the issue is updated successfully, or an error message if an exception occurs. """ logging.info(f"Updating GitHub issue #{issue_number}: {title}") try: issue = self.gi.update_issue(repo_owner, repo_name, issue_number, title, body, labels, state) - logging.info(f"GitHub issue #{issue_number} updated successfully!") - return f"GitHub issue number #{issue} updated successfully!" + logging.info({"status": "success", "message": f"GitHub issue #{issue} updated successfully!"}) + return {"status": "success", "message": f"GitHub issue number #{issue} updated successfully!"} except Exception as e: error_msg = f"Error updating GitHub issue: {str(e)}" logging.error(error_msg) traceback.print_exc(file=sys.stderr) - return error_msg + return {"status": "error", "message": error_msg} @self.mcp.tool() - async def create_github_tag(repo_owner: str, repo_name: str, tag_name: str, message: str) -> str: + async def create_github_tag(repo_owner: str, repo_name: str, tag_name: str, message: str) -> dict[str, Any]: """ Creates a GitHub tag for the specified repository. Args: repo_owner (str): The owner of the GitHub repository. repo_name (str): The name of the GitHub repository. - tag_name (str): The name of the tag to create. - message (str): The message associated with the tag. + tag_name (str): The name of the tag to be created. + message (str): The message or description for the tag. Returns: - str: A success message if the tag is created successfully, or an error message if an exception occurs. - Error Handling: - Logs and returns an error message if the tag creation fails due to an exception. + dict[str, Any]: A dictionary containing the status and message of the tag creation. + Returns a success message if the tag is created successfully, or an error message if an exception occurs. """ logging.info(f"Creating GitHub tag: {tag_name}") try: tag = self.gi.create_tag(repo_owner, repo_name, tag_name, message) - logging.info(f"GitHub tag '{tag_name}' created successfully!") - return f"GitHub tag '{tag}' created successfully!" + logging.info({"status": "success", "message": f"GitHub tag '{tag}' created successfully!"}) + return {"status": "success", "message": f"GitHub tag '{tag}' created successfully!"} except Exception as e: error_msg = f"Error creating GitHub tag: {str(e)}" logging.error(error_msg) traceback.print_exc(file=sys.stderr) - return error_msg + return {"status": "error", "message": error_msg} @self.mcp.tool() - async def create_github_release(repo_owner: str, repo_name: str, tag_name: str, release_name: str, body: str) -> str: + async def create_github_release(repo_owner: str, repo_name: str, tag_name: str, release_name: str, body: str) -> dict[str, Any]: """ - Creates a GitHub release for the specified repository and tag. + Creates a GitHub release from a specified tag. Args: repo_owner (str): The owner of the GitHub repository. repo_name (str): The name of the GitHub repository. - tag_name (str): The tag name to create the release from. - release_name (str): The name of the release to be created. - body (str): The description or body content of the release. + tag_name (str): The name of the tag to create the release from. + release_name (str): The name of the release. + body (str): The body content of the release. Returns: - str: A success message if the release is created successfully, or an error message if an exception occurs. - Error Handling: - Logs and returns an error message if the release creation fails due to an exception. + dict[str, Any]: A dictionary containing the status and message of the release creation. + Returns a success message if the release is created successfully, or an error message if an exception occurs. """ logging.info(f"Creating GitHub release '{release_name}' from tag '{tag_name}'") try: release = self.gi.create_release(repo_owner, repo_name, tag_name, release_name, body) - logging.info(f"GitHub release '{release_name}' created successfully!") - return f"GitHub release '{release}' created successfully!" + logging.info({"status": "success", "message": f"GitHub release '{release}' created successfully!"}) + return {"status": "success", "message": f"GitHub release '{release}' created successfully!"} except Exception as e: error_msg = f"Error creating GitHub release: {str(e)}" logging.error(error_msg) traceback.print_exc(file=sys.stderr) - return error_msg + return {"status": "error", "message": error_msg} @self.mcp.tool() - async def get_ipv4_ipv6_info() -> Dict[str, Any]: + async def get_ipv4_ipv6_info() -> dict[str, Any]: """ Fetches IPv4 and IPv6 information. This function retrieves IPv4 and IPv6 address information using the `self.ip` interface. If either IPv4 or IPv6 information is unavailable, it logs the event and returns an empty dictionary. If both are available, the IPv6 address is added to the IPv4 info dictionary under the "ipv6" key. Returns: - Dict[str, Any]: A dictionary containing IPv4 information, with the IPv6 address included if available. + dict[str, Any]: A dictionary containing IPv4 information, with the IPv6 address included if available. Returns an empty dictionary if either IPv4 or IPv6 information is missing or if an error occurs. Error Handling: Logs any exceptions encountered during the fetching process and returns an empty dictionary. """ - logging.info(f"Fetching IPv4 and IPv6 information") + logging.info({"status": "info", "message": "Fetching IPv4 and IPv6 information"}) try: ipv4_info = self.ip.get_ipv4_info() ipv6_info = self.ip.get_ipv6_info() if ipv4_info is None: - logging.info("No changes returned from getv4_ip_info") + logging.info({"status": "error", "message": "No changes returned from getv4_ip_info"}) return {} if ipv6_info is None: - logging.info("No changes returned from getv6_ip_info") + logging.info({"status": "error", "message": "No changes returned from getv6_ip_info"}) return {} if ipv4_info and ipv6_info: ipv4_info["ipv6"] = ipv6_info["ip"] - logging.info(f"Successfully fetched IP information") + logging.info({"status": "success", "message": "Successfully fetched IPv4 and IPv6 information"}) return ipv4_info except Exception as e: - logging.error(f"Error fetching IP info: {str(e)}") + error_message = f"Error fetching IP info: {str(e)}" + logging.error({"status": "error", "message": error_message}) traceback.print_exc(file=sys.stderr) - return {} + return {"status": "error", "message": error_message} def run(self): """ From 8e69b0a0763e99cff049c8a9c550c5a874a09e2e Mon Sep 17 00:00:00 2001 From: Said Sef Date: Wed, 23 Jul 2025 01:00:24 +0100 Subject: [PATCH 2/2] chore: empty pr diff info message and not error --- src/mcp_github/issues_pr_analyser.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/mcp_github/issues_pr_analyser.py b/src/mcp_github/issues_pr_analyser.py index 73300b1..cf8c6e0 100644 --- a/src/mcp_github/issues_pr_analyser.py +++ b/src/mcp_github/issues_pr_analyser.py @@ -95,10 +95,10 @@ async def get_github_pr_diff(repo_owner: str, repo_name: str, pr_number: int) -> pr_diff = self.gi.get_pr_diff(repo_owner, repo_name, pr_number) if pr_diff is None: no_changes = "No changes returned from get_pr_diff" - logging.info({"status": "error", "message": no_changes}) - return {"status": "error", "message": no_changes} + logging.info({"status": "info", "message": no_changes}) + return {"status": "info", "message": no_changes} logging.info({"status": "success", "message": f"Successfully fetched PR #{pr_number} diff"}) - return {"status": "success", "message": f"{pr_diff}"} + return {"status": "success", "message": pr_diff} except Exception as e: logging.error({"status": "error", "message": str(e)}) traceback.print_exc(file=sys.stderr)