| import json |
| import base64 |
| from connector_base import call_github_api |
|
|
| GITHUB_GET_ME_TOOL = { |
| "type": "function", |
| "function": { |
| "name": "get_me", |
| "description": "Get details of the authenticated GitHub user profile, login, username, and permissions.", |
| "parameters": { |
| "type": "object", |
| "properties": {} |
| } |
| } |
| } |
|
|
| GITHUB_CREATE_REPOSITORY_TOOL = { |
| "type": "function", |
| "function": { |
| "name": "create_repository", |
| "description": "Create a new GitHub repository for the authenticated user or an organization.", |
| "parameters": { |
| "type": "object", |
| "properties": { |
| "name": { |
| "type": "string", |
| "description": "The name of the new repository." |
| }, |
| "description": { |
| "type": "string", |
| "description": "Short description of the repository." |
| }, |
| "private": { |
| "type": "boolean", |
| "description": "Whether the repository should be private. Default is false (public).", |
| "default": False |
| }, |
| "auto_init": { |
| "type": "boolean", |
| "description": "Initialize repository with a README.md file. Default is true.", |
| "default": True |
| }, |
| "org": { |
| "type": "string", |
| "description": "Organization name if creating under an org instead of personal account." |
| } |
| }, |
| "required": ["name"] |
| } |
| } |
| } |
|
|
| GITHUB_DELETE_REPOSITORY_TOOL = { |
| "type": "function", |
| "function": { |
| "name": "delete_repository", |
| "description": "Delete a repository permanently.", |
| "parameters": { |
| "type": "object", |
| "properties": { |
| "owner": { |
| "type": "string", |
| "description": "Repository owner." |
| }, |
| "repo": { |
| "type": "string", |
| "description": "Repository name." |
| } |
| }, |
| "required": ["owner", "repo"] |
| } |
| } |
| } |
|
|
| GITHUB_FORK_REPOSITORY_TOOL = { |
| "type": "function", |
| "function": { |
| "name": "fork_repository", |
| "description": "Fork a repository to the authenticated user account or an organization.", |
| "parameters": { |
| "type": "object", |
| "properties": { |
| "owner": { |
| "type": "string", |
| "description": "Repository owner." |
| }, |
| "repo": { |
| "type": "string", |
| "description": "Repository name." |
| }, |
| "organization": { |
| "type": "string", |
| "description": "Optional organization login to fork into." |
| } |
| }, |
| "required": ["owner", "repo"] |
| } |
| } |
| } |
|
|
| GITHUB_GET_FILE_CONTENTS_TOOL = { |
| "type": "function", |
| "function": { |
| "name": "get_file_contents", |
| "description": "Get contents and metadata of a specific file or directory in a repository.", |
| "parameters": { |
| "type": "object", |
| "properties": { |
| "owner": { |
| "type": "string", |
| "description": "Repository owner (username or organization)." |
| }, |
| "repo": { |
| "type": "string", |
| "description": "Repository name." |
| }, |
| "path": { |
| "type": "string", |
| "description": "Path to the file or directory in the repository." |
| }, |
| "ref": { |
| "type": "string", |
| "description": "The name of the commit/branch/tag. Default: repository default branch." |
| } |
| }, |
| "required": ["owner", "repo", "path"] |
| } |
| } |
| } |
|
|
| GITHUB_CREATE_OR_UPDATE_FILE_TOOL = { |
| "type": "function", |
| "function": { |
| "name": "create_or_update_file", |
| "description": "Create or update a single file in a GitHub repository with commit details.", |
| "parameters": { |
| "type": "object", |
| "properties": { |
| "owner": { |
| "type": "string", |
| "description": "Repository owner." |
| }, |
| "repo": { |
| "type": "string", |
| "description": "Repository name." |
| }, |
| "path": { |
| "type": "string", |
| "description": "Path to the file in the repository." |
| }, |
| "content": { |
| "type": "string", |
| "description": "Text content of the file." |
| }, |
| "message": { |
| "type": "string", |
| "description": "Commit message." |
| }, |
| "branch": { |
| "type": "string", |
| "description": "Branch name to commit to." |
| }, |
| "sha": { |
| "type": "string", |
| "description": "The blob SHA of the file being replaced (required when updating an existing file)." |
| } |
| }, |
| "required": ["owner", "repo", "path", "content", "message", "branch"] |
| } |
| } |
| } |
|
|
| GITHUB_PUSH_FILES_TOOL = { |
| "type": "function", |
| "function": { |
| "name": "push_files", |
| "description": "Create or update multiple files in a single atomic Git commit. Highly recommended for creating complete projects or multi-file website codebases in one single step.", |
| "parameters": { |
| "type": "object", |
| "properties": { |
| "owner": { |
| "type": "string", |
| "description": "Repository owner." |
| }, |
| "repo": { |
| "type": "string", |
| "description": "Repository name." |
| }, |
| "branch": { |
| "type": "string", |
| "description": "Branch to commit files to. Default is repository default branch (e.g. main)." |
| }, |
| "message": { |
| "type": "string", |
| "description": "Commit message." |
| }, |
| "files": { |
| "type": "array", |
| "items": { |
| "type": "object", |
| "properties": { |
| "path": { |
| "type": "string", |
| "description": "Relative file path (e.g., 'index.html', 'css/style.css')." |
| }, |
| "content": { |
| "type": "string", |
| "description": "Raw text content of the file." |
| } |
| }, |
| "required": ["path", "content"] |
| }, |
| "description": "List of files with relative paths and contents to commit." |
| } |
| }, |
| "required": ["owner", "repo", "message", "files"] |
| } |
| } |
| } |
|
|
| GITHUB_DELETE_FILE_TOOL = { |
| "type": "function", |
| "function": { |
| "name": "delete_file", |
| "description": "Delete a file from a GitHub repository.", |
| "parameters": { |
| "type": "object", |
| "properties": { |
| "owner": { |
| "type": "string", |
| "description": "Repository owner." |
| }, |
| "repo": { |
| "type": "string", |
| "description": "Repository name." |
| }, |
| "path": { |
| "type": "string", |
| "description": "Path to the file to delete." |
| }, |
| "message": { |
| "type": "string", |
| "description": "Commit message." |
| }, |
| "sha": { |
| "type": "string", |
| "description": "The blob SHA of the file being deleted." |
| }, |
| "branch": { |
| "type": "string", |
| "description": "The branch name." |
| } |
| }, |
| "required": ["owner", "repo", "path", "message", "sha", "branch"] |
| } |
| } |
| } |
|
|
| GITHUB_GET_REPOSITORY_TREE_TOOL = { |
| "type": "function", |
| "function": { |
| "name": "get_repository_tree", |
| "description": "Get the file and directory hierarchy/tree of a repository.", |
| "parameters": { |
| "type": "object", |
| "properties": { |
| "owner": { |
| "type": "string", |
| "description": "Repository owner." |
| }, |
| "repo": { |
| "type": "string", |
| "description": "Repository name." |
| }, |
| "tree_sha": { |
| "type": "string", |
| "description": "Commit SHA or branch name to get tree for. Default: default branch." |
| }, |
| "recursive": { |
| "type": "boolean", |
| "description": "Set to true to get the entire tree recursively." |
| } |
| }, |
| "required": ["owner", "repo"] |
| } |
| } |
| } |
|
|
| GITHUB_LIST_ISSUES_TOOL = { |
| "type": "function", |
| "function": { |
| "name": "list_issues", |
| "description": "List repository issues with filtering and sorting support.", |
| "parameters": { |
| "type": "object", |
| "properties": { |
| "owner": { |
| "type": "string", |
| "description": "Repository owner." |
| }, |
| "repo": { |
| "type": "string", |
| "description": "Repository name." |
| }, |
| "state": { |
| "type": "string", |
| "enum": ["open", "closed", "all"], |
| "description": "Filter by issue state.", |
| "default": "open" |
| }, |
| "labels": { |
| "type": "string", |
| "description": "Comma-separated list of label names to filter by." |
| }, |
| "sort": { |
| "type": "string", |
| "enum": ["created", "updated", "comments"], |
| "default": "created" |
| }, |
| "direction": { |
| "type": "string", |
| "enum": ["asc", "desc"], |
| "default": "desc" |
| }, |
| "per_page": { |
| "type": "integer", |
| "description": "Results per page (max 100).", |
| "default": 30 |
| }, |
| "page": { |
| "type": "integer", |
| "description": "Page number.", |
| "default": 1 |
| } |
| }, |
| "required": ["owner", "repo"] |
| } |
| } |
| } |
|
|
| GITHUB_CREATE_ISSUE_TOOL = { |
| "type": "function", |
| "function": { |
| "name": "create_issue", |
| "description": "Create a new issue in a GitHub repository.", |
| "parameters": { |
| "type": "object", |
| "properties": { |
| "owner": { |
| "type": "string", |
| "description": "Repository owner." |
| }, |
| "repo": { |
| "type": "string", |
| "description": "Repository name." |
| }, |
| "title": { |
| "type": "string", |
| "description": "The title of the issue." |
| }, |
| "body": { |
| "type": "string", |
| "description": "The contents of the issue." |
| }, |
| "assignees": { |
| "type": "array", |
| "items": {"type": "string"}, |
| "description": "Usernames to assign to this issue." |
| }, |
| "labels": { |
| "type": "array", |
| "items": {"type": "string"}, |
| "description": "Labels to add to this issue." |
| } |
| }, |
| "required": ["owner", "repo", "title"] |
| } |
| } |
| } |
|
|
| GITHUB_UPDATE_ISSUE_TOOL = { |
| "type": "function", |
| "function": { |
| "name": "update_issue", |
| "description": "Update an existing issue title, body, state, labels, or assignees.", |
| "parameters": { |
| "type": "object", |
| "properties": { |
| "owner": { |
| "type": "string", |
| "description": "Repository owner." |
| }, |
| "repo": { |
| "type": "string", |
| "description": "Repository name." |
| }, |
| "issue_number": { |
| "type": "integer", |
| "description": "The issue number." |
| }, |
| "title": { |
| "type": "string", |
| "description": "New title." |
| }, |
| "body": { |
| "type": "string", |
| "description": "New body." |
| }, |
| "state": { |
| "type": "string", |
| "enum": ["open", "closed"], |
| "description": "Target state." |
| }, |
| "labels": { |
| "type": "array", |
| "items": {"type": "string"}, |
| "description": "New array of labels." |
| }, |
| "assignees": { |
| "type": "array", |
| "items": {"type": "string"}, |
| "description": "New array of assignees." |
| } |
| }, |
| "required": ["owner", "repo", "issue_number"] |
| } |
| } |
| } |
|
|
| GITHUB_ADD_ISSUE_COMMENT_TOOL = { |
| "type": "function", |
| "function": { |
| "name": "add_issue_comment", |
| "description": "Add a comment to an existing issue or pull request.", |
| "parameters": { |
| "type": "object", |
| "properties": { |
| "owner": { |
| "type": "string", |
| "description": "Repository owner." |
| }, |
| "repo": { |
| "type": "string", |
| "description": "Repository name." |
| }, |
| "issue_number": { |
| "type": "integer", |
| "description": "The issue or PR number." |
| }, |
| "body": { |
| "type": "string", |
| "description": "Comment text body." |
| } |
| }, |
| "required": ["owner", "repo", "issue_number", "body"] |
| } |
| } |
| } |
|
|
| GITHUB_LIST_PULL_REQUESTS_TOOL = { |
| "type": "function", |
| "function": { |
| "name": "list_pull_requests", |
| "description": "List pull requests in a repository with state filtering and sorting.", |
| "parameters": { |
| "type": "object", |
| "properties": { |
| "owner": { |
| "type": "string", |
| "description": "Repository owner." |
| }, |
| "repo": { |
| "type": "string", |
| "description": "Repository name." |
| }, |
| "state": { |
| "type": "string", |
| "enum": ["open", "closed", "all"], |
| "default": "open" |
| }, |
| "head": { |
| "type": "string", |
| "description": "Filter by head user or namespace and branch name (user:branch-name)." |
| }, |
| "base": { |
| "type": "string", |
| "description": "Filter by base branch name." |
| }, |
| "sort": { |
| "type": "string", |
| "enum": ["created", "updated", "popularity", "long-running"], |
| "default": "created" |
| }, |
| "direction": { |
| "type": "string", |
| "enum": ["asc", "desc"], |
| "default": "desc" |
| }, |
| "per_page": { |
| "type": "integer", |
| "default": 30 |
| }, |
| "page": { |
| "type": "integer", |
| "default": 1 |
| } |
| }, |
| "required": ["owner", "repo"] |
| } |
| } |
| } |
|
|
| GITHUB_CREATE_PULL_REQUEST_TOOL = { |
| "type": "function", |
| "function": { |
| "name": "create_pull_request", |
| "description": "Create a new pull request in a GitHub repository.", |
| "parameters": { |
| "type": "object", |
| "properties": { |
| "owner": { |
| "type": "string", |
| "description": "Repository owner." |
| }, |
| "repo": { |
| "type": "string", |
| "description": "Repository name." |
| }, |
| "title": { |
| "type": "string", |
| "description": "The title of the new pull request." |
| }, |
| "head": { |
| "type": "string", |
| "description": "The branch where changes are implemented." |
| }, |
| "base": { |
| "type": "string", |
| "description": "The branch you want the changes pulled into." |
| }, |
| "body": { |
| "type": "string", |
| "description": "The contents of the pull request description." |
| }, |
| "draft": { |
| "type": "boolean", |
| "description": "Indicates whether the pull request is a draft." |
| } |
| }, |
| "required": ["owner", "repo", "title", "head", "base"] |
| } |
| } |
| } |
|
|
| GITHUB_MERGE_PULL_REQUEST_TOOL = { |
| "type": "function", |
| "function": { |
| "name": "merge_pull_request", |
| "description": "Merge a pull request into its base branch.", |
| "parameters": { |
| "type": "object", |
| "properties": { |
| "owner": { |
| "type": "string", |
| "description": "Repository owner." |
| }, |
| "repo": { |
| "type": "string", |
| "description": "Repository name." |
| }, |
| "pull_number": { |
| "type": "integer", |
| "description": "The pull request number." |
| }, |
| "commit_title": { |
| "type": "string", |
| "description": "Title for the automatic commit message." |
| }, |
| "commit_message": { |
| "type": "string", |
| "description": "Extra detail for the automatic commit message." |
| }, |
| "merge_method": { |
| "type": "string", |
| "enum": ["merge", "squash", "rebase"], |
| "default": "merge" |
| } |
| }, |
| "required": ["owner", "repo", "pull_number"] |
| } |
| } |
| } |
|
|
| GITHUB_LIST_BRANCHES_TOOL = { |
| "type": "function", |
| "function": { |
| "name": "list_branches", |
| "description": "List branches in a repository.", |
| "parameters": { |
| "type": "object", |
| "properties": { |
| "owner": { |
| "type": "string", |
| "description": "Repository owner." |
| }, |
| "repo": { |
| "type": "string", |
| "description": "Repository name." |
| }, |
| "protected": { |
| "type": "boolean", |
| "description": "Filter by protected status." |
| }, |
| "per_page": { |
| "type": "integer", |
| "default": 30 |
| }, |
| "page": { |
| "type": "integer", |
| "default": 1 |
| } |
| }, |
| "required": ["owner", "repo"] |
| } |
| } |
| } |
|
|
| GITHUB_CREATE_BRANCH_TOOL = { |
| "type": "function", |
| "function": { |
| "name": "create_branch", |
| "description": "Create a new Git branch from an existing branch or commit reference.", |
| "parameters": { |
| "type": "object", |
| "properties": { |
| "owner": { |
| "type": "string", |
| "description": "Repository owner." |
| }, |
| "repo": { |
| "type": "string", |
| "description": "Repository name." |
| }, |
| "branch": { |
| "type": "string", |
| "description": "Name for the new branch." |
| }, |
| "from_branch": { |
| "type": "string", |
| "description": "Source branch name to branch off from. Default: default branch." |
| } |
| }, |
| "required": ["owner", "repo", "branch"] |
| } |
| } |
| } |
|
|
| GITHUB_LIST_COMMITS_TOOL = { |
| "type": "function", |
| "function": { |
| "name": "list_commits", |
| "description": "List commits on a repository.", |
| "parameters": { |
| "type": "object", |
| "properties": { |
| "owner": { |
| "type": "string", |
| "description": "Repository owner." |
| }, |
| "repo": { |
| "type": "string", |
| "description": "Repository name." |
| }, |
| "sha": { |
| "type": "string", |
| "description": "SHA or branch to start listing commits from." |
| }, |
| "path": { |
| "type": "string", |
| "description": "Only commits containing this file path will be returned." |
| }, |
| "author": { |
| "type": "string", |
| "description": "GitHub login or email address by which to filter." |
| }, |
| "per_page": { |
| "type": "integer", |
| "default": 30 |
| }, |
| "page": { |
| "type": "integer", |
| "default": 1 |
| } |
| }, |
| "required": ["owner", "repo"] |
| } |
| } |
| } |
|
|
| GITHUB_GET_COMMIT_TOOL = { |
| "type": "function", |
| "function": { |
| "name": "get_commit", |
| "description": "Get a specific commit by its SHA reference, including diff and stats.", |
| "parameters": { |
| "type": "object", |
| "properties": { |
| "owner": { |
| "type": "string", |
| "description": "Repository owner." |
| }, |
| "repo": { |
| "type": "string", |
| "description": "Repository name." |
| }, |
| "ref": { |
| "type": "string", |
| "description": "Commit SHA or ref." |
| } |
| }, |
| "required": ["owner", "repo", "ref"] |
| } |
| } |
| } |
|
|
| GITHUB_TRIGGER_WORKFLOW_TOOL = { |
| "type": "function", |
| "function": { |
| "name": "actions_run_trigger", |
| "description": "Trigger a GitHub Actions workflow execution dispatch.", |
| "parameters": { |
| "type": "object", |
| "properties": { |
| "owner": { |
| "type": "string", |
| "description": "Repository owner." |
| }, |
| "repo": { |
| "type": "string", |
| "description": "Repository name." |
| }, |
| "workflow_id": { |
| "type": "string", |
| "description": "The ID or filename of the workflow (e.g. 'deploy.yml' or 'main.yml')." |
| }, |
| "ref": { |
| "type": "string", |
| "description": "Git reference (branch or tag name) to trigger the workflow on. Default is 'main'.", |
| "default": "main" |
| }, |
| "inputs": { |
| "type": "object", |
| "description": "Input keys and values configured in the workflow dispatch trigger." |
| } |
| }, |
| "required": ["owner", "repo", "workflow_id"] |
| } |
| } |
| } |
|
|
| GITHUB_GET_PAGES_INFO_TOOL = { |
| "type": "function", |
| "function": { |
| "name": "get_pages_info", |
| "description": "Get or enable GitHub Pages configuration and retrieve the live website URL and deployment status.", |
| "parameters": { |
| "type": "object", |
| "properties": { |
| "owner": { |
| "type": "string", |
| "description": "Repository owner." |
| }, |
| "repo": { |
| "type": "string", |
| "description": "Repository name." |
| }, |
| "enable": { |
| "type": "boolean", |
| "description": "Enable GitHub Pages if not already active. Default is true.", |
| "default": True |
| }, |
| "branch": { |
| "type": "string", |
| "description": "Source branch for Pages build. Default is 'main'.", |
| "default": "main" |
| }, |
| "path": { |
| "type": "string", |
| "enum": ["/", "/docs"], |
| "description": "Source folder path. Default is '/'.", |
| "default": "/" |
| } |
| }, |
| "required": ["owner", "repo"] |
| } |
| } |
| } |
|
|
| GITHUB_SEARCH_REPOSITORIES_TOOL = { |
| "type": "function", |
| "function": { |
| "name": "search_repositories", |
| "description": "Search for GitHub repositories using query qualifiers (e.g., 'topic:python stars:>500').", |
| "parameters": { |
| "type": "object", |
| "properties": { |
| "query": { |
| "type": "string", |
| "description": "Search query keywords and qualifiers." |
| }, |
| "sort": { |
| "type": "string", |
| "enum": ["stars", "forks", "help-wanted-issues", "updated"] |
| }, |
| "order": { |
| "type": "string", |
| "enum": ["asc", "desc"], |
| "default": "desc" |
| }, |
| "per_page": { |
| "type": "integer", |
| "default": 30 |
| }, |
| "page": { |
| "type": "integer", |
| "default": 1 |
| } |
| }, |
| "required": ["query"] |
| } |
| } |
| } |
|
|
| GITHUB_SEARCH_CODE_TOOL = { |
| "type": "function", |
| "function": { |
| "name": "search_code", |
| "description": "Search for code across GitHub repositories (e.g., 'addClass in:file repo:owner/repo').", |
| "parameters": { |
| "type": "object", |
| "properties": { |
| "query": { |
| "type": "string", |
| "description": "Search query and qualifiers." |
| }, |
| "sort": { |
| "type": "string", |
| "enum": ["indexed"] |
| }, |
| "order": { |
| "type": "string", |
| "enum": ["asc", "desc"], |
| "default": "desc" |
| }, |
| "per_page": { |
| "type": "integer", |
| "default": 30 |
| }, |
| "page": { |
| "type": "integer", |
| "default": 1 |
| } |
| }, |
| "required": ["query"] |
| } |
| } |
| } |
|
|
| GITHUB_SEARCH_ISSUES_TOOL = { |
| "type": "function", |
| "function": { |
| "name": "search_issues", |
| "description": "Search for issues and pull requests across GitHub (e.g., 'is:open is:issue repo:owner/repo error').", |
| "parameters": { |
| "type": "object", |
| "properties": { |
| "query": { |
| "type": "string", |
| "description": "Search query and qualifiers." |
| }, |
| "sort": { |
| "type": "string", |
| "enum": ["comments", "reactions", "created", "updated"] |
| }, |
| "order": { |
| "type": "string", |
| "enum": ["asc", "desc"], |
| "default": "desc" |
| }, |
| "per_page": { |
| "type": "integer", |
| "default": 30 |
| }, |
| "page": { |
| "type": "integer", |
| "default": 1 |
| } |
| }, |
| "required": ["query"] |
| } |
| } |
| } |
|
|
| ALL_GITHUB_MCP_TOOLS = [ |
| GITHUB_GET_ME_TOOL, |
| GITHUB_CREATE_REPOSITORY_TOOL, |
| GITHUB_DELETE_REPOSITORY_TOOL, |
| GITHUB_FORK_REPOSITORY_TOOL, |
| GITHUB_GET_FILE_CONTENTS_TOOL, |
| GITHUB_CREATE_OR_UPDATE_FILE_TOOL, |
| GITHUB_PUSH_FILES_TOOL, |
| GITHUB_DELETE_FILE_TOOL, |
| GITHUB_GET_REPOSITORY_TREE_TOOL, |
| GITHUB_LIST_ISSUES_TOOL, |
| GITHUB_CREATE_ISSUE_TOOL, |
| GITHUB_UPDATE_ISSUE_TOOL, |
| GITHUB_ADD_ISSUE_COMMENT_TOOL, |
| GITHUB_LIST_PULL_REQUESTS_TOOL, |
| GITHUB_CREATE_PULL_REQUEST_TOOL, |
| GITHUB_MERGE_PULL_REQUEST_TOOL, |
| GITHUB_LIST_BRANCHES_TOOL, |
| GITHUB_CREATE_BRANCH_TOOL, |
| GITHUB_LIST_COMMITS_TOOL, |
| GITHUB_GET_COMMIT_TOOL, |
| GITHUB_TRIGGER_WORKFLOW_TOOL, |
| GITHUB_GET_PAGES_INFO_TOOL, |
| GITHUB_SEARCH_REPOSITORIES_TOOL, |
| GITHUB_SEARCH_CODE_TOOL, |
| GITHUB_SEARCH_ISSUES_TOOL |
| ] |
|
|
| def get_github_metadata(func_name: str, args: dict): |
| if func_name == "get_me": |
| return "github", "Get authenticated GitHub user" |
| elif func_name == "create_repository": |
| name = args.get("name", "") |
| return "github", f"Creating repository: {name}" if name else "Creating GitHub repository" |
| elif func_name == "delete_repository": |
| repo = args.get("repo", "") |
| return "github", f"Deleting repository: {repo}" if repo else "Deleting GitHub repository" |
| elif func_name == "fork_repository": |
| repo = args.get("repo", "") |
| return "github", f"Forking repository: {repo}" if repo else "Forking repository" |
| elif func_name == "get_file_contents": |
| path = args.get("path", "") |
| return "github", f"Reading file: {path}" if path else "Reading file contents" |
| elif func_name == "create_or_update_file": |
| path = args.get("path", "") |
| return "github", f"Writing file: {path}" if path else "Updating file" |
| elif func_name == "push_files": |
| files = args.get("files", []) |
| count = len(files) if isinstance(files, list) else 0 |
| return "github", f"Pushing {count} files in commit" if count > 0 else "Pushing files in commit" |
| elif func_name == "delete_file": |
| path = args.get("path", "") |
| return "github", f"Deleting file: {path}" if path else "Deleting file" |
| elif func_name == "get_repository_tree": |
| repo = args.get("repo", "") |
| return "github", f"Browsing repository tree: {repo}" if repo else "Browsing repository tree" |
| elif func_name == "list_issues": |
| repo = args.get("repo", "") |
| return "github", f"Listing issues in {repo}" if repo else "Listing issues" |
| elif func_name == "create_issue": |
| title = args.get("title", "") |
| return "github", f"Creating issue: {title}" if title else "Creating issue" |
| elif func_name == "update_issue": |
| num = args.get("issue_number", "") |
| return "github", f"Updating issue #{num}" if num else "Updating issue" |
| elif func_name == "add_issue_comment": |
| num = args.get("issue_number", "") |
| return "github", f"Commenting on #{num}" if num else "Adding issue comment" |
| elif func_name == "list_pull_requests": |
| repo = args.get("repo", "") |
| return "github", f"Listing PRs in {repo}" if repo else "Listing pull requests" |
| elif func_name == "create_pull_request": |
| title = args.get("title", "") |
| return "github", f"Creating PR: {title}" if title else "Creating pull request" |
| elif func_name == "merge_pull_request": |
| num = args.get("pull_number", "") |
| return "github", f"Merging PR #{num}" if num else "Merging pull request" |
| elif func_name == "list_branches": |
| repo = args.get("repo", "") |
| return "github", f"Listing branches in {repo}" if repo else "Listing branches" |
| elif func_name == "create_branch": |
| branch = args.get("branch", "") |
| return "github", f"Creating branch: {branch}" if branch else "Creating branch" |
| elif func_name == "list_commits": |
| repo = args.get("repo", "") |
| return "github", f"Listing commits in {repo}" if repo else "Listing commits" |
| elif func_name == "get_commit": |
| ref = args.get("ref", "") |
| short_ref = ref[:7] if ref else "" |
| return "github", f"Inspecting commit {short_ref}" if short_ref else "Getting commit" |
| elif func_name == "actions_run_trigger": |
| w_id = args.get("workflow_id", "") |
| return "github", f"Triggering workflow: {w_id}" if w_id else "Triggering workflow dispatch" |
| elif func_name == "get_pages_info": |
| repo = args.get("repo", "") |
| return "github", f"Configuring GitHub Pages: {repo}" if repo else "Getting GitHub Pages info" |
| elif func_name == "search_repositories": |
| q = args.get("query", "") |
| return "github", f"Searching repos: {q}" if q else "Searching repositories" |
| elif func_name == "search_code": |
| q = args.get("query", "") |
| return "github", f"Searching code: {q}" if q else "Searching code" |
| elif func_name == "search_issues": |
| q = args.get("query", "") |
| return "github", f"Searching issues: {q}" if q else "Searching issues" |
| return "github", func_name |
|
|
| async def execute_github_tool(user_id: str, fn_name: str, args: dict) -> str: |
| try: |
| repo_scoped_tools = { |
| "delete_repository", "fork_repository", "get_file_contents", |
| "create_or_update_file", "push_files", "delete_file", |
| "get_repository_tree", "list_issues", "create_issue", |
| "update_issue", "add_issue_comment", "list_pull_requests", |
| "create_pull_request", "merge_pull_request", "list_branches", |
| "create_branch", "list_commits", "get_commit", |
| "actions_run_trigger", "get_pages_info" |
| } |
|
|
| if fn_name in repo_scoped_tools: |
| owner = args.get("owner") |
| repo = args.get("repo") |
| if not owner or not repo or not str(owner).strip() or not str(repo).strip(): |
| return json.dumps({"error": f"Missing required 'owner' or 'repo' parameter for '{fn_name}'. Both owner and repo must be non-empty strings."}) |
|
|
| if fn_name == "get_me": |
| status, body = await call_github_api(user_id, "GET", "/user") |
| if status != 200: |
| return body |
| d = json.loads(body) |
| sanitized = { |
| "login": d.get("login"), |
| "name": d.get("name"), |
| "email": d.get("email"), |
| "bio": d.get("bio"), |
| "public_repos": d.get("public_repos"), |
| "total_private_repos": d.get("total_private_repos"), |
| "html_url": d.get("html_url") |
| } |
| return json.dumps(sanitized) |
|
|
| elif fn_name == "create_repository": |
| name = args.get("name") |
| org = args.get("org") |
| if not name or not str(name).strip(): |
| return json.dumps({"error": "Missing required parameter: 'name'."}) |
|
|
| payload = { |
| "name": name, |
| "description": args.get("description", ""), |
| "private": args.get("private", False), |
| "auto_init": args.get("auto_init", True) |
| } |
|
|
| url = f"/orgs/{org}/repos" if org else "/user/repos" |
| status, body = await call_github_api(user_id, "POST", url, json_body=payload) |
| if status not in [200, 201]: |
| return body |
|
|
| d = json.loads(body) |
| return json.dumps({ |
| "status": "created", |
| "name": d.get("name"), |
| "full_name": d.get("full_name"), |
| "owner": d.get("owner", {}).get("login"), |
| "private": d.get("private"), |
| "html_url": d.get("html_url"), |
| "default_branch": d.get("default_branch", "main") |
| }) |
|
|
| elif fn_name == "delete_repository": |
| owner = args.get("owner") |
| repo = args.get("repo") |
| status, body = await call_github_api(user_id, "DELETE", f"/repos/{owner}/{repo}") |
| if status != 204: |
| return body |
|
|
| return json.dumps({"status": "deleted", "repository": f"{owner}/{repo}"}) |
|
|
| elif fn_name == "fork_repository": |
| owner = args.get("owner") |
| repo = args.get("repo") |
| payload = {} |
| if args.get("organization"): |
| payload["organization"] = args.get("organization") |
|
|
| status, body = await call_github_api(user_id, "POST", f"/repos/{owner}/{repo}/forks", json_body=payload) |
| if status not in [200, 202]: |
| return body |
|
|
| d = json.loads(body) |
| return json.dumps({ |
| "status": "forked", |
| "full_name": d.get("full_name"), |
| "html_url": d.get("html_url") |
| }) |
|
|
| elif fn_name == "get_file_contents": |
| owner = args.get("owner") |
| repo = args.get("repo") |
| path = args.get("path", "").lstrip("/") |
| ref = args.get("ref") |
| params = {"ref": ref} if ref else {} |
|
|
| status, body = await call_github_api(user_id, "GET", f"/repos/{owner}/{repo}/contents/{path}", params=params) |
| if status != 200: |
| return body |
|
|
| data = json.loads(body) |
| if isinstance(data, list): |
| items = [] |
| for item in data: |
| items.append({ |
| "name": item.get("name"), |
| "path": item.get("path"), |
| "type": item.get("type"), |
| "size": item.get("size"), |
| "sha": item.get("sha") |
| }) |
| return json.dumps({"type": "directory", "path": path, "entries": items}) |
|
|
| if isinstance(data, dict): |
| content_str = data.get("content", "") |
| encoding = data.get("encoding", "") |
| decoded_text = "" |
| if encoding == "base64" and content_str: |
| try: |
| decoded_text = base64.b64decode(content_str.replace("\n", "")).decode("utf-8", errors="replace") |
| except Exception: |
| decoded_text = content_str |
|
|
| return json.dumps({ |
| "type": "file", |
| "name": data.get("name"), |
| "path": data.get("path"), |
| "sha": data.get("sha"), |
| "size": data.get("size"), |
| "content": decoded_text if decoded_text else content_str |
| }) |
|
|
| elif fn_name == "create_or_update_file": |
| owner = args.get("owner") |
| repo = args.get("repo") |
| path = args.get("path", "").lstrip("/") |
| content_raw = args.get("content", "") |
| message = args.get("message", "Commit file") |
| branch = args.get("branch", "main") |
| sha = args.get("sha") |
|
|
| if not path: |
| return json.dumps({"error": "Missing required parameter: 'path'."}) |
|
|
| encoded_content = base64.b64encode(content_raw.encode("utf-8")).decode("utf-8") |
| payload = { |
| "message": message, |
| "content": encoded_content, |
| "branch": branch |
| } |
| if sha: |
| payload["sha"] = sha |
|
|
| status, body = await call_github_api(user_id, "PUT", f"/repos/{owner}/{repo}/contents/{path}", json_body=payload) |
| if status not in [200, 201]: |
| return body |
|
|
| d = json.loads(body) |
| return json.dumps({ |
| "status": "success", |
| "commit": d.get("commit", {}).get("sha"), |
| "content": { |
| "name": d.get("content", {}).get("name"), |
| "path": d.get("content", {}).get("path"), |
| "sha": d.get("content", {}).get("sha"), |
| "html_url": d.get("content", {}).get("html_url") |
| } |
| }) |
|
|
| elif fn_name == "push_files": |
| owner = args.get("owner") |
| repo = args.get("repo") |
| files = args.get("files", []) |
| message = args.get("message", "Batch commit files") |
| branch = args.get("branch") |
|
|
| if not files or not isinstance(files, list): |
| return json.dumps({"error": "Missing or invalid 'files' parameter. Must provide a non-empty list of objects containing 'path' and 'content'."}) |
|
|
| if not branch: |
| s_repo, b_repo = await call_github_api(user_id, "GET", f"/repos/{owner}/{repo}") |
| if s_repo == 200: |
| try: |
| branch = json.loads(b_repo).get("default_branch") or "main" |
| except Exception: |
| branch = "main" |
| else: |
| return json.dumps({"error": f"Failed to fetch repository information for {owner}/{repo}: {b_repo}"}) |
|
|
| tree_payload = [] |
| for f in files: |
| if not isinstance(f, dict): |
| continue |
| clean_path = f.get("path", "").lstrip("/") |
| if not clean_path: |
| continue |
| tree_payload.append({ |
| "path": clean_path, |
| "mode": "100644", |
| "type": "blob", |
| "content": f.get("content", "") |
| }) |
|
|
| if not tree_payload: |
| return json.dumps({"error": "No valid files found with 'path' and 'content'."}) |
|
|
| s_ref, b_ref = await call_github_api(user_id, "GET", f"/repos/{owner}/{repo}/git/refs/heads/{branch}") |
| if s_ref != 200: |
| s_ref, b_ref = await call_github_api(user_id, "GET", f"/repos/{owner}/{repo}/git/ref/heads/{branch}") |
|
|
| if s_ref == 200: |
| try: |
| latest_commit_sha = json.loads(b_ref).get("object", {}).get("sha") |
| s_commit, b_commit = await call_github_api(user_id, "GET", f"/repos/{owner}/{repo}/git/commits/{latest_commit_sha}") |
| if s_commit != 200: |
| return json.dumps({"error": f"Failed to get base commit: {b_commit}"}) |
|
|
| base_tree_sha = json.loads(b_commit).get("tree", {}).get("sha") |
|
|
| s_tree, b_tree = await call_github_api( |
| user_id, "POST", |
| f"/repos/{owner}/{repo}/git/trees", |
| json_body={"base_tree": base_tree_sha, "tree": tree_payload} |
| ) |
| if s_tree not in [200, 201]: |
| return json.dumps({"error": f"Failed to create git tree: {b_tree}"}) |
|
|
| new_tree_sha = json.loads(b_tree).get("sha") |
|
|
| s_new_commit, b_new_commit = await call_github_api( |
| user_id, "POST", |
| f"/repos/{owner}/{repo}/git/commits", |
| json_body={ |
| "message": message, |
| "tree": new_tree_sha, |
| "parents": [latest_commit_sha] |
| } |
| ) |
| if s_new_commit not in [200, 201]: |
| return json.dumps({"error": f"Failed to create commit: {b_new_commit}"}) |
|
|
| new_commit_sha = json.loads(b_new_commit).get("sha") |
|
|
| s_up, b_up = await call_github_api( |
| user_id, "PATCH", |
| f"/repos/{owner}/{repo}/git/refs/heads/{branch}", |
| json_body={"sha": new_commit_sha} |
| ) |
| if s_up not in [200, 201]: |
| return json.dumps({"error": f"Failed to update branch ref: {b_up}"}) |
|
|
| return json.dumps({ |
| "status": "success", |
| "commit_sha": new_commit_sha, |
| "branch": branch, |
| "files_committed": [item["path"] for item in tree_payload] |
| }) |
| except Exception as e: |
| return json.dumps({"error": f"Error updating files: {str(e)}"}) |
| else: |
| first_file = tree_payload[0] |
| init_payload = { |
| "message": message, |
| "content": base64.b64encode(first_file.get("content", "").encode("utf-8")).decode("utf-8"), |
| "branch": branch |
| } |
| s_init, b_init = await call_github_api( |
| user_id, "PUT", |
| f"/repos/{owner}/{repo}/contents/{first_file['path']}", |
| json_body=init_payload |
| ) |
| if s_init not in [200, 201]: |
| return json.dumps({"error": f"Failed to initialize repository with file '{first_file['path']}': {b_init}"}) |
|
|
| if len(tree_payload) == 1: |
| return json.dumps({ |
| "status": "success", |
| "branch": branch, |
| "files_committed": [first_file["path"]] |
| }) |
|
|
| remaining_files = tree_payload[1:] |
| s_ref2, b_ref2 = await call_github_api(user_id, "GET", f"/repos/{owner}/{repo}/git/refs/heads/{branch}") |
| if s_ref2 != 200: |
| return json.dumps({"status": "partial_success", "branch": branch, "files_committed": [first_file["path"]]}) |
|
|
| latest_commit_sha = json.loads(b_ref2).get("object", {}).get("sha") |
| s_commit, b_commit = await call_github_api(user_id, "GET", f"/repos/{owner}/{repo}/git/commits/{latest_commit_sha}") |
| if s_commit != 200: |
| return json.dumps({"status": "partial_success", "branch": branch, "files_committed": [first_file["path"]]}) |
|
|
| base_tree_sha = json.loads(b_commit).get("tree", {}).get("sha") |
| s_tree, b_tree = await call_github_api( |
| user_id, "POST", |
| f"/repos/{owner}/{repo}/git/trees", |
| json_body={"base_tree": base_tree_sha, "tree": remaining_files} |
| ) |
| if s_tree not in [200, 201]: |
| return json.dumps({"status": "partial_success", "branch": branch, "files_committed": [first_file["path"]]}) |
|
|
| new_tree_sha = json.loads(b_tree).get("sha") |
| s_new_commit, b_new_commit = await call_github_api( |
| user_id, "POST", |
| f"/repos/{owner}/{repo}/git/commits", |
| json_body={ |
| "message": message, |
| "tree": new_tree_sha, |
| "parents": [latest_commit_sha] |
| } |
| ) |
| if s_new_commit not in [200, 201]: |
| return json.dumps({"status": "partial_success", "branch": branch, "files_committed": [first_file["path"]]}) |
|
|
| new_commit_sha = json.loads(b_new_commit).get("sha") |
| await call_github_api( |
| user_id, "PATCH", |
| f"/repos/{owner}/{repo}/git/refs/heads/{branch}", |
| json_body={"sha": new_commit_sha} |
| ) |
|
|
| return json.dumps({ |
| "status": "success", |
| "commit_sha": new_commit_sha, |
| "branch": branch, |
| "files_committed": [item["path"] for item in tree_payload] |
| }) |
|
|
| elif fn_name == "delete_file": |
| owner = args.get("owner") |
| repo = args.get("repo") |
| path = args.get("path", "").lstrip("/") |
| message = args.get("message", "Delete file") |
| sha = args.get("sha") |
| branch = args.get("branch", "main") |
|
|
| if not path or not sha: |
| return json.dumps({"error": "Missing required parameters: 'path' and 'sha' are required."}) |
|
|
| payload = { |
| "message": message, |
| "sha": sha, |
| "branch": branch |
| } |
|
|
| status, body = await call_github_api(user_id, "DELETE", f"/repos/{owner}/{repo}/contents/{path}", json_body=payload) |
| if status != 200: |
| return body |
|
|
| d = json.loads(body) |
| return json.dumps({ |
| "status": "deleted", |
| "commit": d.get("commit", {}).get("sha") |
| }) |
|
|
| elif fn_name == "get_repository_tree": |
| owner = args.get("owner") |
| repo = args.get("repo") |
| tree_sha = args.get("tree_sha") |
| recursive = args.get("recursive", False) |
|
|
| if not tree_sha: |
| s_r, b_r = await call_github_api(user_id, "GET", f"/repos/{owner}/{repo}") |
| if s_r == 200: |
| try: |
| tree_sha = json.loads(b_r).get("default_branch") or "main" |
| except Exception: |
| tree_sha = "main" |
| else: |
| tree_sha = "main" |
|
|
| params = {"recursive": "1"} if recursive else {} |
| status, body = await call_github_api(user_id, "GET", f"/repos/{owner}/{repo}/git/trees/{tree_sha}", params=params) |
| if status == 404: |
| return json.dumps({ |
| "sha": "", |
| "truncated": False, |
| "tree": [] |
| }) |
| if status != 200: |
| return body |
|
|
| d = json.loads(body) |
| raw_tree = d.get("tree", []) |
| sanitized_tree = [] |
| for item in raw_tree[:300]: |
| sanitized_tree.append({ |
| "path": item.get("path"), |
| "type": item.get("type"), |
| "size": item.get("size") |
| }) |
| return json.dumps({ |
| "sha": d.get("sha"), |
| "truncated": d.get("truncated", False) or len(raw_tree) > 300, |
| "tree": sanitized_tree |
| }) |
|
|
| elif fn_name == "list_issues": |
| owner = args.get("owner") |
| repo = args.get("repo") |
| params = { |
| "state": args.get("state", "open"), |
| "per_page": min(args.get("per_page", 30), 100), |
| "page": args.get("page", 1), |
| "sort": args.get("sort", "created"), |
| "direction": args.get("direction", "desc") |
| } |
| if args.get("labels"): |
| params["labels"] = args.get("labels") |
|
|
| status, body = await call_github_api(user_id, "GET", f"/repos/{owner}/{repo}/issues", params=params) |
| if status != 200: |
| return body |
|
|
| issues = [] |
| for item in json.loads(body): |
| if "pull_request" in item: |
| continue |
| issues.append({ |
| "number": item.get("number"), |
| "title": item.get("title"), |
| "state": item.get("state"), |
| "user": item.get("user", {}).get("login"), |
| "labels": [l.get("name") for l in item.get("labels", []) if isinstance(l, dict)], |
| "comments": item.get("comments"), |
| "created_at": item.get("created_at"), |
| "updated_at": item.get("updated_at"), |
| "html_url": item.get("html_url") |
| }) |
| return json.dumps(issues) |
|
|
| elif fn_name == "create_issue": |
| owner = args.get("owner") |
| repo = args.get("repo") |
| title = args.get("title") |
| if not title: |
| return json.dumps({"error": "Missing required parameter: 'title'."}) |
|
|
| payload = { |
| "title": title, |
| "body": args.get("body", "") |
| } |
| if args.get("assignees"): |
| payload["assignees"] = args.get("assignees") |
| if args.get("labels"): |
| payload["labels"] = args.get("labels") |
|
|
| status, body = await call_github_api(user_id, "POST", f"/repos/{owner}/{repo}/issues", json_body=payload) |
| if status != 201: |
| return body |
|
|
| d = json.loads(body) |
| return json.dumps({ |
| "number": d.get("number"), |
| "title": d.get("title"), |
| "state": d.get("state"), |
| "html_url": d.get("html_url") |
| }) |
|
|
| elif fn_name == "update_issue": |
| owner = args.get("owner") |
| repo = args.get("repo") |
| issue_number = args.get("issue_number") |
| if not issue_number: |
| return json.dumps({"error": "Missing required parameter: 'issue_number'."}) |
|
|
| payload = {} |
| for field in ["title", "body", "state", "labels", "assignees"]: |
| if field in args and args[field] is not None: |
| payload[field] = args[field] |
|
|
| status, body = await call_github_api(user_id, "PATCH", f"/repos/{owner}/{repo}/issues/{issue_number}", json_body=payload) |
| if status != 200: |
| return body |
|
|
| d = json.loads(body) |
| return json.dumps({ |
| "number": d.get("number"), |
| "title": d.get("title"), |
| "state": d.get("state"), |
| "html_url": d.get("html_url") |
| }) |
|
|
| elif fn_name == "add_issue_comment": |
| owner = args.get("owner") |
| repo = args.get("repo") |
| issue_number = args.get("issue_number") |
| if not issue_number: |
| return json.dumps({"error": "Missing required parameter: 'issue_number'."}) |
|
|
| payload = {"body": args.get("body", "")} |
|
|
| status, body = await call_github_api(user_id, "POST", f"/repos/{owner}/{repo}/issues/{issue_number}/comments", json_body=payload) |
| if status != 201: |
| return body |
|
|
| d = json.loads(body) |
| return json.dumps({ |
| "id": d.get("id"), |
| "user": d.get("user", {}).get("login"), |
| "body": d.get("body"), |
| "created_at": d.get("created_at"), |
| "html_url": d.get("html_url") |
| }) |
|
|
| elif fn_name == "list_pull_requests": |
| owner = args.get("owner") |
| repo = args.get("repo") |
| params = { |
| "state": args.get("state", "open"), |
| "per_page": min(args.get("per_page", 30), 100), |
| "page": args.get("page", 1), |
| "sort": args.get("sort", "created"), |
| "direction": args.get("direction", "desc") |
| } |
| if args.get("head"): |
| params["head"] = args.get("head") |
| if args.get("base"): |
| params["base"] = args.get("base") |
|
|
| status, body = await call_github_api(user_id, "GET", f"/repos/{owner}/{repo}/pulls", params=params) |
| if status != 200: |
| return body |
|
|
| prs = [] |
| for item in json.loads(body): |
| prs.append({ |
| "number": item.get("number"), |
| "title": item.get("title"), |
| "state": item.get("state"), |
| "draft": item.get("draft"), |
| "user": item.get("user", {}).get("login"), |
| "head": item.get("head", {}).get("ref"), |
| "base": item.get("base", {}).get("ref"), |
| "created_at": item.get("created_at"), |
| "updated_at": item.get("updated_at"), |
| "html_url": item.get("html_url") |
| }) |
| return json.dumps(prs) |
|
|
| elif fn_name == "create_pull_request": |
| owner = args.get("owner") |
| repo = args.get("repo") |
| title = args.get("title") |
| head = args.get("head") |
| base = args.get("base") |
|
|
| if not title or not head or not base: |
| return json.dumps({"error": "Missing required parameters: 'title', 'head', and 'base' are required."}) |
|
|
| payload = { |
| "title": title, |
| "head": head, |
| "base": base, |
| "body": args.get("body", ""), |
| "draft": args.get("draft", False) |
| } |
|
|
| status, body = await call_github_api(user_id, "POST", f"/repos/{owner}/{repo}/pulls", json_body=payload) |
| if status != 201: |
| return body |
|
|
| d = json.loads(body) |
| return json.dumps({ |
| "number": d.get("number"), |
| "title": d.get("title"), |
| "state": d.get("state"), |
| "head": d.get("head", {}).get("ref"), |
| "base": d.get("base", {}).get("ref"), |
| "html_url": d.get("html_url") |
| }) |
|
|
| elif fn_name == "merge_pull_request": |
| owner = args.get("owner") |
| repo = args.get("repo") |
| pull_number = args.get("pull_number") |
| if not pull_number: |
| return json.dumps({"error": "Missing required parameter: 'pull_number'."}) |
|
|
| payload = { |
| "merge_method": args.get("merge_method", "merge") |
| } |
| if args.get("commit_title"): |
| payload["commit_title"] = args.get("commit_title") |
| if args.get("commit_message"): |
| payload["commit_message"] = args.get("commit_message") |
|
|
| status, body = await call_github_api(user_id, "PUT", f"/repos/{owner}/{repo}/pulls/{pull_number}/merge", json_body=payload) |
| return body |
|
|
| elif fn_name == "list_branches": |
| owner = args.get("owner") |
| repo = args.get("repo") |
| params = { |
| "per_page": min(args.get("per_page", 30), 100), |
| "page": args.get("page", 1) |
| } |
| if "protected" in args: |
| params["protected"] = str(args["protected"]).lower() |
|
|
| status, body = await call_github_api(user_id, "GET", f"/repos/{owner}/{repo}/branches", params=params) |
| if status != 200: |
| return body |
|
|
| branches = [] |
| for b in json.loads(body): |
| branches.append({ |
| "name": b.get("name"), |
| "protected": b.get("protected", False), |
| "commit_sha": b.get("commit", {}).get("sha") |
| }) |
| return json.dumps(branches) |
|
|
| elif fn_name == "create_branch": |
| owner = args.get("owner") |
| repo = args.get("repo") |
| branch = args.get("branch") |
| from_branch = args.get("from_branch") |
|
|
| if not branch: |
| return json.dumps({"error": "Missing required parameter: 'branch'."}) |
|
|
| if not from_branch: |
| s_r, b_r = await call_github_api(user_id, "GET", f"/repos/{owner}/{repo}") |
| if s_r == 200: |
| try: |
| from_branch = json.loads(b_r).get("default_branch") or "main" |
| except Exception: |
| from_branch = "main" |
| else: |
| from_branch = "main" |
|
|
| s_ref, b_ref = await call_github_api(user_id, "GET", f"/repos/{owner}/{repo}/git/refs/heads/{from_branch}") |
| if s_ref != 200: |
| s_ref, b_ref = await call_github_api(user_id, "GET", f"/repos/{owner}/{repo}/git/ref/heads/{from_branch}") |
| if s_ref != 200: |
| return json.dumps({"error": f"Failed to get base branch reference: {b_ref}"}) |
|
|
| sha = json.loads(b_ref).get("object", {}).get("sha") |
| payload = { |
| "ref": f"refs/heads/{branch}", |
| "sha": sha |
| } |
|
|
| status, body = await call_github_api(user_id, "POST", f"/repos/{owner}/{repo}/git/refs", json_body=payload) |
| if status != 201: |
| return body |
|
|
| return json.dumps({ |
| "ref": json.loads(body).get("ref"), |
| "sha": sha, |
| "status": "created" |
| }) |
|
|
| elif fn_name == "list_commits": |
| owner = args.get("owner") |
| repo = args.get("repo") |
| params = { |
| "per_page": min(args.get("per_page", 30), 100), |
| "page": args.get("page", 1) |
| } |
| for f in ["sha", "path", "author"]: |
| if args.get(f): |
| params[f] = args[f] |
|
|
| status, body = await call_github_api(user_id, "GET", f"/repos/{owner}/{repo}/commits", params=params) |
| if status != 200: |
| return body |
|
|
| commits = [] |
| for c in json.loads(body): |
| commits.append({ |
| "sha": c.get("sha"), |
| "message": c.get("commit", {}).get("message"), |
| "author": c.get("commit", {}).get("author", {}).get("name"), |
| "date": c.get("commit", {}).get("author", {}).get("date"), |
| "html_url": c.get("html_url") |
| }) |
| return json.dumps(commits) |
|
|
| elif fn_name == "get_commit": |
| owner = args.get("owner") |
| repo = args.get("repo") |
| ref = args.get("ref") |
| if not ref: |
| return json.dumps({"error": "Missing required parameter: 'ref'."}) |
|
|
| status, body = await call_github_api(user_id, "GET", f"/repos/{owner}/{repo}/commits/{ref}") |
| if status != 200: |
| return body |
|
|
| d = json.loads(body) |
| files = [] |
| for f in d.get("files", [])[:50]: |
| files.append({ |
| "filename": f.get("filename"), |
| "status": f.get("status"), |
| "additions": f.get("additions"), |
| "deletions": f.get("deletions"), |
| "patch": f.get("patch", "")[:1000] |
| }) |
| return json.dumps({ |
| "sha": d.get("sha"), |
| "message": d.get("commit", {}).get("message"), |
| "author": d.get("commit", {}).get("author", {}).get("name"), |
| "stats": d.get("stats"), |
| "files": files |
| }) |
|
|
| elif fn_name == "actions_run_trigger": |
| owner = args.get("owner") |
| repo = args.get("repo") |
| workflow_id = args.get("workflow_id") |
| if not workflow_id: |
| return json.dumps({"error": "Missing required parameter: 'workflow_id'."}) |
|
|
| payload = { |
| "ref": args.get("ref", "main") |
| } |
| if args.get("inputs"): |
| payload["inputs"] = args.get("inputs") |
|
|
| status, body = await call_github_api(user_id, "POST", f"/repos/{owner}/{repo}/actions/workflows/{workflow_id}/dispatches", json_body=payload) |
| if status != 204: |
| return body |
|
|
| return json.dumps({ |
| "status": "triggered", |
| "workflow": workflow_id, |
| "ref": payload["ref"] |
| }) |
|
|
| elif fn_name == "get_pages_info": |
| owner = args.get("owner") |
| repo = args.get("repo") |
| enable = args.get("enable", True) |
| branch = args.get("branch", "main") |
| path = args.get("path", "/") |
|
|
| status, body = await call_github_api(user_id, "GET", f"/repos/{owner}/{repo}/pages") |
| if status == 200: |
| d = json.loads(body) |
| return json.dumps({ |
| "status": d.get("status"), |
| "html_url": d.get("html_url"), |
| "source": d.get("source") |
| }) |
| elif status == 404 and enable: |
| enable_payload = { |
| "source": { |
| "branch": branch, |
| "path": path |
| } |
| } |
| status_en, body_en = await call_github_api(user_id, "POST", f"/repos/{owner}/{repo}/pages", json_body=enable_payload) |
| if status_en in [200, 201]: |
| d = json.loads(body_en) |
| return json.dumps({ |
| "status": "enabled", |
| "html_url": d.get("html_url"), |
| "source": d.get("source") |
| }) |
| else: |
| return json.dumps({"error": f"Failed to enable GitHub Pages: {body_en}"}) |
|
|
| return body |
|
|
| elif fn_name == "search_repositories": |
| query = args.get("query") |
| if not query: |
| return json.dumps({"error": "Missing required parameter: 'query'."}) |
|
|
| params = { |
| "q": query, |
| "per_page": min(args.get("per_page", 30), 100), |
| "page": args.get("page", 1), |
| "order": args.get("order", "desc") |
| } |
| if args.get("sort"): |
| params["sort"] = args.get("sort") |
|
|
| status, body = await call_github_api(user_id, "GET", "/search/repositories", params=params) |
| if status != 200: |
| return body |
|
|
| d = json.loads(body) |
| items = [] |
| for item in d.get("items", []): |
| items.append({ |
| "full_name": item.get("full_name"), |
| "description": item.get("description"), |
| "stars": item.get("stargazers_count"), |
| "forks": item.get("forks_count"), |
| "language": item.get("language"), |
| "html_url": item.get("html_url") |
| }) |
| return json.dumps({ |
| "total_count": d.get("total_count"), |
| "items": items |
| }) |
|
|
| elif fn_name == "search_code": |
| query = args.get("query") |
| if not query: |
| return json.dumps({"error": "Missing required parameter: 'query'."}) |
|
|
| params = { |
| "q": query, |
| "per_page": min(args.get("per_page", 30), 100), |
| "page": args.get("page", 1), |
| "order": args.get("order", "desc") |
| } |
| if args.get("sort"): |
| params["sort"] = args.get("sort") |
|
|
| status, body = await call_github_api(user_id, "GET", "/search/code", params=params) |
| if status != 200: |
| return body |
|
|
| d = json.loads(body) |
| items = [] |
| for item in d.get("items", []): |
| items.append({ |
| "name": item.get("name"), |
| "path": item.get("path"), |
| "repository": item.get("repository", {}).get("full_name"), |
| "html_url": item.get("html_url") |
| }) |
| return json.dumps({ |
| "total_count": d.get("total_count"), |
| "items": items |
| }) |
|
|
| elif fn_name == "search_issues": |
| query = args.get("query") |
| if not query: |
| return json.dumps({"error": "Missing required parameter: 'query'."}) |
|
|
| params = { |
| "q": query, |
| "per_page": min(args.get("per_page", 30), 100), |
| "page": args.get("page", 1), |
| "order": args.get("order", "desc") |
| } |
| if args.get("sort"): |
| params["sort"] = args.get("sort") |
|
|
| status, body = await call_github_api(user_id, "GET", "/search/issues", params=params) |
| if status != 200: |
| return body |
|
|
| d = json.loads(body) |
| items = [] |
| for item in d.get("items", []): |
| items.append({ |
| "number": item.get("number"), |
| "title": item.get("title"), |
| "state": item.get("state"), |
| "repository_url": item.get("repository_url"), |
| "user": item.get("user", {}).get("login"), |
| "comments": item.get("comments"), |
| "created_at": item.get("created_at"), |
| "html_url": item.get("html_url") |
| }) |
| return json.dumps({ |
| "total_count": d.get("total_count"), |
| "items": items |
| }) |
|
|
| return json.dumps({"error": f"Unsupported GitHub MCP tool: {fn_name}"}) |
|
|
| except Exception as e: |
| return json.dumps({"error": f"Error executing {fn_name}: {str(e)}"}) |
|
|
