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)}"})