apunkt/MetaGPT
1
0
Fork 0
mirror of https://github.com/FoundationAgents/MetaGPT.git synced 2026-05-18 13:55:17 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge branch 'feature/git/pr_issue' into 'mgx_ops'

Browse source 
feat: +tool lib git clone, push

See merge request pub/MetaGPT!89
This commit is contained in:
林义章 2024-05-08 03:27:41 +00:00
commit d6fdb21207
6 changed files with 309 additions and 192 deletions
Download patch file Download diff file Expand all files Collapse all files

270
metagpt/tools/libs/git.py
View file

@ -3,56 +3,48 @@
from __future__ import annotations
from pathlib import Path
from typing import Optional
from typing import Optional, Union
from github.Issue import Issue
from github.PullRequest import PullRequest
from metagpt.tools.tool_registry import register_tool
from metagpt.utils.git_repository import GitRepository
from metagpt.utils.git_repository import GitBranch, GitRepository
@register_tool(tags=["git"])
async def git_clone(url: str, output_dir: str | Path = None) -> Path:
@register_tool(tags=["software development", "git", "Clone a git repository to local"])
async def git_clone(url: Union[str, Path], output_dir: Union[str, Path] = None) -> Path:
"""
Clones a Git repository from the given URL.
Args:
url (str): The URL of the Git repository to clone.
output_dir (str or Path, optional): The directory where the repository will be cloned.
If not provided, the repository will be cloned into the current working directory.
url (Union[str, Path]): The URL or local path of the Git repository to clone.
output_dir (Union[str, Path], optional): The directory where the repository should be cloned.
If None, the repository will be cloned into the current working directory. Defaults to None.
Returns:
Path: The path to the cloned repository.
Raises:
ValueError: If the specified Git root is invalid.
Example:
>>> # git clone to /TO/PATH
>>> url = 'https://github.com/geekan/MetaGPT.git'
>>> output_dir = "/TO/PATH"
>>> repo_dir = await git_clone(url=url, output_dir=output_dir)
>>> print(repo_dir)
/TO/PATH/MetaGPT
>>> # git clone to default directory.
>>> url = 'https://github.com/geekan/MetaGPT.git'
>>> repo_dir = await git_clone(url)
>>> print(repo_dir)
/WORK_SPACE/downloads/MetaGPT
>>> url = "https://github.com/iorisa/snake-game.git"
>>> local_path = await git_clone(url=url)
>>> print(local_path)
/local/path/to/snake-game
"""
repo = await GitRepository.clone_from(url, output_dir)
repo = await GitRepository.clone_from(url=url, output_dir=output_dir)
return repo.workdir
@register_tool(
tags=["software development", "git", "Checks out the specific commit/branch/tag of the local Git repository."]
)
async def git_checkout(repo_dir: str | Path, commit_id: str):
"""
Checks out a specific commit in a Git repository.
Args:
repo_dir (str or Path): The directory containing the Git repository.
commit_id (str): The ID of the commit to check out.
commit_id (str): The ID of the commit or the name of branch/tag to check out.
Raises:
ValueError: If the specified Git root is invalid.
@ -69,145 +61,171 @@ async def git_checkout(repo_dir: str | Path, commit_id: str):
await repo.checkout(commit_id)
@register_tool(tags=["git"])
async def create_pull_request(
@register_tool(tags=["software development", "git", "Commit the changes and push to remote git repository."])
async def git_push(
local_path: Union[str, Path],
access_token: str,
comments: str = "Commit",
new_branch: str = "",
) -> GitBranch:
"""
Pushes changes from a local Git repository to its remote counterpart.
Args:
local_path (Union[str, Path]): The path to the local Git repository.
access_token (str): The access token for authentication. Use `get_env` to get access token.
comments (str, optional): The commit message to use. Defaults to "Commit".
new_branch (str, optional): The name of the new branch to create and push changes to.
If not provided, changes will be pushed to the current branch. Defaults to "".
Returns:
GitBranch: The branch to which the changes were pushed.
Raises:
ValueError: If the provided local_path does not point to a valid Git repository.
Example:
>>> url = "https://github.com/iorisa/snake-game.git"
>>> local_path = await git_clone(url=url)
>>> from metagpt.tools.libs import get_env
>>> access_token = await get_env(key="access_token", app_name="github") # Read access token from enviroment variables.
>>> comments = "Archive"
>>> new_branch = "feature/new"
>>> branch = await git_push(local_path=local_path, access_token=access_token, comments=comments, new_branch=new_branch)
>>> base = branch.base
>>> head = branch.head
>>> repo_name = branch.repo_name
>>> print(f"base branch:'{base}', head branch:'{head}', repo_name:'{repo_name}'")
base branch:'master', head branch:'feature/new', repo_name:'iorisa/snake-game'
"""
if not GitRepository.is_git_dir(local_path):
raise ValueError("Invalid local git repository")
repo = GitRepository(local_path=local_path, auto_init=False)
branch = await repo.push(new_branch=new_branch, comments=comments, access_token=access_token)
return branch
@register_tool(tags=["software development", "git", "create a git pull/merge request"])
async def git_create_pull(
base: str,
head: str,
base_repo_name: str,
access_token: str,
head_repo_name: Optional[str] = None,
title: Optional[str] = None,
body: Optional[str] = None,
issue: Optional[Issue] = None,
) -> PullRequest:
"""
Creates a pull request in a Git repository.
Creates a pull request on a Git repository.
Args:
access_token (str): The access token for authentication.
base (str): The name of the base branch of the pull request (e.g., 'main', 'master').
head (str): The name of the head branch of the pull request (e.g., 'feature-branch').
base (str): The base branch of the pull request.
head (str): The head branch of the pull request.
base_repo_name (str): The full repository name (user/repo) where the pull request will be created.
access_token (str): The access token for authentication. Use `get_env` to get access token.
head_repo_name (Optional[str], optional): The full repository name (user/repo) where the pull request will merge from. Defaults to None.
title (Optional[str]): The title of the pull request.
body (Optional[str]): The body of the pull request.
title (Optional[str], optional): The title of the pull request. Defaults to None.
body (Optional[str], optional): The body of the pull request. Defaults to None.
issue (Optional[Issue], optional): The related issue of the pull request. Defaults to None.
Example:
>>> # push and create pull
>>> url = "https://github.com/iorisa/snake-game.git"
>>> local_path = await git_clone(url=url)
>>> from metagpt.tools.libs import get_env
>>> access_token = await get_env(key="access_token", app_name="github")
>>> comments = "Archive"
>>> new_branch = "feature/new"
>>> branch = await git_push(local_path=local_path, access_token=access_token, comments=comments, new_branch=new_branch)
>>> base = branch.base
>>> head = branch.head
>>> repo_name = branch.repo_name
>>> print(f"base branch:'{base}', head branch:'{head}', repo_name:'{repo_name}'")
base branch:'master', head branch:'feature/new', repo_name:'iorisa/snake-game'
>>> title = "feat: modify http lib",
>>> body = "Change HTTP library used to send requests"
>>> pr = await git_create_pull(
>>> base_repo_name=repo_name,
>>> base=base,
>>> head=head,
>>> title=title,
>>> body=body,
>>> access_token=access_token,
>>> )
>>> print(pr)
PullRequest("feat: modify http lib")
>>> # create pull request
>>> base_repo_name = "geekan/MetaGPT"
>>> head_repo_name = "ioris/MetaGPT"
>>> base = "master"
>>> head = "feature/http"
>>> title = "feat: modify http lib",
>>> body = "Change HTTP library used to send requests"
>>> from metagpt.tools.libs import get_env
>>> access_token = await get_env(key="access_token", app_name="github")
>>> pr = await git_create_pull(
>>> base_repo_name=base_repo_name,
>>> head_repo_name=head_repo_name,
>>> base=base,
>>> head=head,
>>> title=title,
>>> body=body,
>>> access_token=access_token,
>>> )
>>> print(pr)
PullRequest("feat: modify http lib")
Returns:
PullRequest: The created pull request object.
Raises:
ValueError: If `access_token` is invalid. Visit: "https://github.com/settings/tokens"
Any exceptions that might occur during the pull request creation process.
Note:
This function is intended to be used in an asynchronous context (with `await`).
Example:
>>> # Merge Request
>>> repo_name = "user/repo" # "user/repo" for example: "https://github.com/user/repo.git"
>>> base = "master" # branch that merge to
>>> head = "feature/new_feature" # branch that merge from
>>> title = "Implement new feature"
>>> body = "This pull request adds functionality X, Y, and Z."
>>> pull_request = await create_pull_request(
repo_name=repo_name,
base=base,
head=head,
title=title,
body=body,
access_token=get_env("git_access_token")
)
>>> print(pull_request)
PullRequest(title="Implement new feature", number=26)
>>> # Pull Request
>>> base_repo_name = "user1/repo1" # for example: "user1/repo1" from "https://github.com/user1/repo1.git"
>>> head_repo_name = "user2/repo2" # for example: "user2/repo2" from "https://github.com/user2/repo2.git"
>>> base = "master" # branch that merge to
>>> head = "feature/new_feature" # branch that merge from
>>> title = "Implement new feature"
>>> body = "This pull request adds functionality X, Y, and Z."
>>> pull_request = await create_pull_request(
base_repo_name=base_repo_name,
head_repo_name=head_repo_name,
base=base,
head=head,
title=title,
body=body,
access_token=get_env("git_access_token")
)
>>> print(pull_request)
PullRequest(title="Implement new feature", number=26)
PullRequest: The created pull request.
"""
return await GitRepository.create_pull(
base_repo_name=base_repo_name,
head_repo_name=head_repo_name,
base=base,
head=head,
base_repo_name=base_repo_name,
head_repo_name=head_repo_name,
title=title,
body=body,
issue=issue,
access_token=access_token,
)
@register_tool(tags=["git"])
async def create_issue(
access_token: str,
@register_tool(tags=["software development", "create a git issue"])
async def git_create_issue(
repo_name: str,
title: str,
access_token: str,
body: Optional[str] = None,
assignee: Optional[str] = None,
labels: Optional[list[str]] = None,
) -> Issue:
"""
Creates an issue in the specified repository.
Creates an issue on a Git repository.
Args:
access_token (str): The access token for authentication.
Visit `https://github.com/settings/tokens` to obtain a personal access token.
For more authentication options, visit: `https://pygithub.readthedocs.io/en/latest/examples/Authentication.html`
repo_name (str): The full repository name (user/repo) where the issue will be created.
repo_name (str): The name of the repository.
title (str): The title of the issue.
access_token (str): The access token for authentication. Use `get_env` to get access token.
body (Optional[str], optional): The body of the issue. Defaults to None.
assignee (Optional[str], optional): The username of the assignee for the issue. Defaults to None.
labels (Optional[list[str]], optional): A list of label names to associate with the issue. Defaults to None.
Returns:
Issue: The created issue object.
Example:
>>> # Create an issue with title and body
>>> repo_name = "username/repository"
>>> title = "Bug Report"
>>> body = "I found a bug in the application."
>>> issue = await create_issue(
repo_name=repo_name,
title=title,
body=body,
access_token=get_env("git_access_token")
)
>>> repo_name = "geekan/MetaGPT"
>>> title = "This is a new issue"
>>> from metagpt.tools.libs import get_env
>>> access_token = await get_env(key="access_token", app_name="github")
>>> body = "This is the issue body."
>>> issue = await git_create_issue(
>>> repo_name=repo_name,
>>> title=title,
>>> access_token=access_token,
>>> body=body,
>>> )
>>> print(issue)
Issue(title="Bug Report", number=26)
Issue("This is a new issue")
>>> # Create an issue with title, body, assignee, and labels
>>> repo_name = "username/repository"
>>> title = "Bug Report"
>>> body = "I found a bug in the application."
>>> assignee = "john_doe"
>>> labels = ["enhancement", "help wanted"]
>>> issue = await create_issue(
repo_name=repo_name,
title=title,
body=body,
assignee=assigee,
labels=labels,
access_token=get_env("git_access_token")
)
>>> print(issue)
Issue(title="Bug Report", number=26)
Returns:
Issue: The created issue.
"""
return await GitRepository.create_issue(
repo_name=repo_name, title=title, body=body, assignee=assignee, labels=labels, access_token=access_token
)
return await GitRepository.create_issue(repo_name=repo_name, title=title, body=body, access_token=access_token)

11
metagpt/tools/libs/shell.py
View file

@ -49,12 +49,5 @@ async def shell_execute(
"""
cwd = str(cwd) if cwd else None
shell = True if isinstance(command, str) else False
process = subprocess.Popen(command, cwd=cwd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env, shell=shell)
try:
# Wait for the process to complete, with a timeout
stdout, stderr = process.communicate(timeout=timeout)
return stdout.decode("utf-8"), stderr.decode("utf-8"), process.returncode
except subprocess.TimeoutExpired:
process.kill() # Kill the process if it times out
stdout, stderr = process.communicate()
raise ValueError(f"{stdout.decode('utf-8')}\n{stderr.decode('utf-8')}")
result = subprocess.run(command, cwd=cwd, capture_output=True, text=True, env=env, timeout=timeout, shell=shell)
return result.stdout, result.stderr, result.returncode

2
metagpt/tools/libs/software_development.py
View file

@ -6,10 +6,8 @@ from pathlib import Path
from metagpt.const import ASSISTANT_ALIAS
from metagpt.logs import ToolLogItem, log_tool_output
from metagpt.tools.tool_registry import register_tool
@register_tool(tags=["software development", "import git repo"])
async def import_git_repo(url: str) -> Path:
"""
Imports a project from a Git website and formats it to MetaGPT project format to enable incremental appending requirements.

107
metagpt/utils/git_repository.py
View file

@ -13,11 +13,12 @@ import shutil
import uuid
from enum import Enum
from pathlib import Path
from subprocess import TimeoutExpired
from typing import Dict, List, Optional
from git.repo import Repo
from git.repo.fun import is_git_dir
from github import Auth, Github
from github import Auth, BadCredentialsException, Github
from github.GithubObject import NotSet
from github.Issue import Issue
from github.Label import Label
@ -25,6 +26,7 @@ from github.Milestone import Milestone
from github.NamedUser import NamedUser
from github.PullRequest import PullRequest
from gitignore_parser import parse_gitignore
from pydantic import BaseModel
from tenacity import retry, stop_after_attempt, wait_random_exponential
from metagpt.logs import logger
@ -49,6 +51,12 @@ class RateLimitError(Exception):
super().__init__(self.message)
class GitBranch(BaseModel):
head: str
base: str
repo_name: str
class GitRepository:
"""A class representing a Git repository.
@ -177,6 +185,52 @@ class GitRepository:
return None
return Path(self._repository.working_dir)
@property
def current_branch(self) -> str:
"""
Returns the name of the current active branch.
Returns:
str: The name of the current active branch.
"""
return self._repository.active_branch.name
@property
def remote_url(self) -> str:
try:
return self._repository.remotes.origin.url
except AttributeError:
return ""
@property
def repo_name(self) -> str:
if self.remote_url:
# This assumes a standard HTTPS or SSH format URL
# HTTPS format example: https://github.com/username/repo_name.git
# SSH format example: git@github.com:username/repo_name.git
if self.remote_url.startswith("https://"):
return self.remote_url.split("/", maxsplit=3)[-1].replace(".git", "")
elif self.remote_url.startswith("git@"):
return self.remote_url.split(":")[-1].replace(".git", "")
return ""
def new_branch(self, branch_name: str) -> str:
"""
Creates a new branch with the given name.
Args:
branch_name (str): The name of the new branch to create.
Returns:
str: The name of the newly created branch.
If the provided branch_name is empty, returns the name of the current active branch.
"""
if not branch_name:
return self.current_branch
new_branch = self._repository.create_head(branch_name)
new_branch.checkout()
return new_branch.name
def archive(self, comments="Archive"):
"""Archive the current state of the Git repository.
@ -186,6 +240,57 @@ class GitRepository:
self.add_change(self.changed_files)
self.commit(comments)
async def push(
self, new_branch: str, comments="Archive", access_token: Optional[str] = None, auth: Optional[Auth] = None
) -> GitBranch:
"""
Pushes changes to the remote repository.
Args:
new_branch (str): The name of the new branch to be pushed.
comments (str, optional): Comments to be associated with the push. Defaults to "Archive".
access_token (str, optional): Access token for authentication. Defaults to None. Visit `https://pygithub.readthedocs.io/en/latest/examples/Authentication.html`, `https://github.com/PyGithub/PyGithub/blob/main/doc/examples/Authentication.rst`.
auth (Auth, optional): Optional authentication object. Defaults to None.
Returns:
GitBranch: The pushed branch object.
Raises:
ValueError: If neither `auth` nor `access_token` is provided.
BadCredentialsException: If authentication fails due to bad credentials or timeout.
Note:
This function assumes that `self.current_branch`, `self.new_branch()`, `self.archive()`,
`ctx.config.proxy`, `ctx.config`, `self.remote_url`, `shell_execute()`, and `logger` are
defined and accessible within the scope of this function.
"""
if not auth and not access_token:
raise ValueError('`access_token` is invalid. Visit: "https://github.com/settings/tokens"')
from metagpt.context import Context
base = self.current_branch
head = base if not new_branch else self.new_branch(new_branch)
self.archive(comments)
ctx = Context()
env = ctx.new_environ()
proxy = ["-c", f"http.proxy={ctx.config.proxy}"] if ctx.config.proxy else []
token = access_token or auth.token
remote_url = f"https://{token}@" + self.remote_url.removeprefix("https://")
command = ["git"] + proxy + ["push", remote_url]
logger.info(" ".join(command).replace(token, "<TOKEN>"))
try:
stdout, stderr, return_code = await shell_execute(
command=command, cwd=str(self.workdir), env=env, timeout=15
)
except TimeoutExpired as e:
info = str(e).replace(token, "<TOKEN>")
raise BadCredentialsException(status=401, message=info)
info = f"{stdout}\n{stderr}\nexit: {return_code}\n"
info = info.replace(token, "<TOKEN>")
logger.info(info)
return GitBranch(base=base, head=head, repo_name=self.repo_name)
def new_file_repository(self, relative_path: Path | str = ".") -> FileRepository:
"""Create a new instance of FileRepository associated with this Git repository.

61
tests/metagpt/tools/libs/test_git.py
View file

@ -3,12 +3,17 @@
from __future__ import annotations
import os
import uuid
import pytest
from github import Auth, Github
from pydantic import BaseModel
from metagpt.context import Context
from metagpt.roles.di.data_interpreter import DataInterpreter
from metagpt.schema import UserMessage
from metagpt.tools.libs.git import git_checkout, git_clone
from metagpt.utils.common import awrite
from metagpt.utils.git_repository import GitRepository
@ -17,7 +22,7 @@ class SWEBenchItem(BaseModel):
repo: str
def get_env(key):
async def get_env(key: str, app_name: str = ""):
return os.environ.get(key)
@ -37,8 +42,9 @@ async def test_git(url: str, commit_id: str):
@pytest.mark.skip
def test_login():
auth = Auth.Login(get_env("GITHUB_USER"), get_env("GITHUB_PWD"))
@pytest.mark.asyncio
async def test_login():
auth = Auth.Login(await get_env("GITHUB_USER"), await get_env("GITHUB_PWD"))
g = Github(auth=auth)
repo = g.get_repo("geekan/MetaGPT")
topics = repo.get_topics()
@ -55,7 +61,7 @@ async def test_new_issue():
repo_name="iorisa/MetaGPT",
title="This is a new issue",
body="This is the issue body",
access_token=get_env("GITHUB_PERSONAL_ACCESS_TOKEN"),
access_token=await get_env(key="access_token", app_name="github"),
)
print(issue)
assert issue.number
@ -74,20 +80,21 @@ async def test_new_pr():
>>> - [x] Send 'POST' request with/without body
"""
pr = await GitRepository.create_pull(
repo_name="iorisa/MetaGPT",
base_repo_name="iorisa/MetaGPT",
base="send18",
head="fixbug/gbk",
title="Test pr",
body=body,
access_token=get_env("GITHUB_PERSONAL_ACCESS_TOKEN"),
access_token=await get_env(key="access_token", app_name="github"),
)
print(pr)
assert pr
@pytest.mark.skip
def test_auth():
access_token = get_env("GITHUB_PERSONAL_ACCESS_TOKEN")
@pytest.mark.asyncio
async def test_auth():
access_token = await get_env(key="access_token", app_name="github")
auth = Auth.Token(access_token)
g = Github(auth=auth)
u = g.get_user()
@ -98,5 +105,43 @@ def test_auth():
pass
@pytest.mark.skip
@pytest.mark.asyncio
async def test_github(context):
repo = await GitRepository.clone_from(url="https://github.com/iorisa/snake-game.git")
content = uuid.uuid4().hex
await awrite(filename=repo.workdir / "README.md", data=content)
branch = await repo.push(
new_branch=f"feature/{content[0:8]}", access_token=await get_env(key="access_token", app_name="github")
)
pr = await GitRepository.create_pull(
base=branch.base,
head=branch.head,
base_repo_name=branch.repo_name,
title=f"new pull {content[0:8]}",
access_token=await get_env(key="access_token", app_name="github"),
)
assert pr
@pytest.mark.asyncio
@pytest.mark.parametrize(
"content",
[
# "create a new issue to github repo 'iorisa/snake-game' :'The snake did not grow longer after eating'",
"Resolve the issue #1 'Snake not growing longer after eating' in the GitHub repository https://github.com/iorisa/snake-game.git', and create a new pull request about the issue"
],
)
async def test_git_create_issue(content: str):
context = Context()
di = DataInterpreter(context=context, tools=["<all>"])
prerequisite = "from metagpt.tools.libs import get_env"
await di.execute_code.run(code=prerequisite, language="python")
di.put_message(UserMessage(content=content))
while not di.is_idle:
await di.run()
if __name__ == "__main__":
pytest.main([__file__, "-s"])

50
tests/metagpt/tools/libs/test_software_development.py
View file

@ -1,59 +1,17 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
from typing import Dict
import pytest
from metagpt.tools.libs import (
fix_bug,
git_archive,
run_qa_test,
write_codes,
write_design,
write_prd,
write_project_plan,
)
from metagpt.tools.libs.software_development import import_git_repo
@pytest.mark.asyncio
async def test_software_team():
path = await write_prd("snake game")
assert path
path = await write_design(path)
assert path
path = await write_project_plan(path)
assert path
path = await write_codes(path)
assert path
path = await run_qa_test(path)
assert path
issue = """
pygame 2.0.1 (SDL 2.0.14, Python 3.9.17)
Hello from the pygame community. https://www.pygame.org/contribute.html
Traceback (most recent call last):
File "/Users/ix/github/bak/MetaGPT/workspace/snake_game/snake_game/main.py", line 10, in <module>
main()
File "/Users/ix/github/bak/MetaGPT/workspace/snake_game/snake_game/main.py", line 7, in main
game.start_game()
File "/Users/ix/github/bak/MetaGPT/workspace/snake_game/snake_game/game.py", line 81, in start_game
x
NameError: name 'x' is not defined
"""
path = await fix_bug(path, issue)
assert path
new_path = await write_prd("snake game with moving enemy", path)
assert new_path == path
git_log = await git_archive(new_path)
assert git_log
async def get_env_description() -> Dict[str, str]:
return {'await get_env(key="access_token", app_name="github")': "get the access token for github authentication."}
@pytest.mark.skip
@pytest.mark.asyncio
async def test_import_repo():
url = "https://github.com/spec-first/connexion.git"