Merge branch 'feature/import_repo' into featur/intent_detect

metagpt/tools/libs/__init__.py

@@ -12,6 +12,15 @@ from metagpt.tools.libs import (
     web_scraping,
     email_login,
 )
+from metagpt.tools.libs.software_development import (
+    write_prd,
+    write_design,
+    write_project_plan,
+    write_codes,
+    run_qa_test,
+    fix_bug,
+    git_archive,
+)
 
 _ = (
     data_preprocess,
@@ -20,4 +29,11 @@ _ = (
     gpt_v_generator,
     web_scraping,
     email_login,
+    write_prd,
+    write_design,
+    write_project_plan,
+    write_codes,
+    run_qa_test,
+    fix_bug,
+    git_archive,
 )  # Avoid pre-commit error

metagpt/tools/libs/git.py

@@ -0,0 +1,65 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+from __future__ import annotations
+
+from pathlib import Path
+
+from metagpt.tools.tool_registry import register_tool
+from metagpt.utils.git_repository import GitRepository
+
+
+@register_tool(tags=["git"])
+async def git_clone(url: str, output_dir: 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.
+
+    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
+    """
+    repo = await GitRepository.clone_from(url, output_dir)
+    return repo.workdir
+
+
+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.
+
+    Raises:
+        ValueError: If the specified Git root is invalid.
+
+    Example:
+        >>> repo_dir = '/TO/GIT/REPO'
+        >>> commit_id = 'main'
+        >>> await git_checkout(repo_dir=repo_dir, commit_id=commit_id)
+        git checkout main
+    """
+    repo = GitRepository(local_path=repo_dir, auto_init=False)
+    if not repo.is_valid:
+        ValueError(f"Invalid git root: {repo_dir}")
+    await repo.checkout(commit_id)

metagpt/tools/libs/shell.py

@@ -0,0 +1,63 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+from __future__ import annotations
+
+import subprocess
+from pathlib import Path
+from typing import Dict, List, Tuple, Union
+
+from metagpt.tools.tool_registry import register_tool
+
+
+@register_tool(tags=["shell"])
+async def shell_execute(
+    command: Union[List[str], str], cwd: str | Path = None, env: Dict = None, timeout: int = 600
+) -> Tuple[str, str, int]:
+    """
+    Execute a command asynchronously and return its standard output and standard error.
+
+    Args:
+        command (Union[List[str], str]): The command to execute and its arguments. It can be provided either as a list
+            of strings or as a single string.
+        cwd (str | Path, optional): The current working directory for the command. Defaults to None.
+        env (Dict, optional): Environment variables to set for the command. Defaults to None.
+        timeout (int, optional): Timeout for the command execution in seconds. Defaults to 600.
+
+    Returns:
+        Tuple[str, str, int]: A tuple containing the string type standard output and string type standard error of the executed command and int type return code.
+
+    Raises:
+        ValueError: If the command times out, this error is raised. The error message contains both standard output and
+         standard error of the timed-out process.
+
+    Example:
+        >>> # command is a list
+        >>> stdout, stderr, returncode = await shell_execute(command=["ls", "-l"], cwd="/home/user", env={"PATH": "/usr/bin"})
+        >>> print(stdout)
+        total 8
+        -rw-r--r-- 1 user user    0 Mar 22 10:00 file1.txt
+        -rw-r--r-- 1 user user    0 Mar 22 10:00 file2.txt
+        ...
+
+        >>> # command is a string of shell script
+        >>> stdout, stderr, returncode = await shell_execute(command="ls -l", cwd="/home/user", env={"PATH": "/usr/bin"})
+        >>> print(stdout)
+        total 8
+        -rw-r--r-- 1 user user    0 Mar 22 10:00 file1.txt
+        -rw-r--r-- 1 user user    0 Mar 22 10:00 file2.txt
+        ...
+
+    References:
+        This function uses `subprocess.Popen` for executing shell commands asynchronously.
+    """
+    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')}")

metagpt/tools/libs/software_development.py

@@ -0,0 +1,301 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Optional
+
+from metagpt.const import BUGFIX_FILENAME, REQUIREMENT_FILENAME
+from metagpt.schema import BugFixContext, Message
+from metagpt.tools.tool_registry import register_tool
+from metagpt.utils.common import any_to_str
+
+
+@register_tool(tags=["software development", "ProductManager"])
+async def write_prd(idea: str, project_path: Optional[str | Path] = None) -> Path:
+    """Writes a PRD based on user requirements.
+
+    Args:
+        idea (str): The idea or concept for the PRD.
+        project_path (Optional[str|Path], optional): The path to an existing project directory.
+            If it's None, a new project path will be created. Defaults to None.
+
+    Returns:
+        Path: The path to the PRD files under the project directory
+
+    Example:
+        >>> # Create a new project:
+        >>> from metagpt.tools.libs.software_development import write_prd
+        >>> prd_path = await write_prd("Create a new feature for the application")
+        >>> print(prd_path)
+        '/path/to/project_path/docs/prd/'
+
+        >>> # Add user requirements to the exists project:
+        >>> from metagpt.tools.libs.software_development import write_prd
+        >>> project_path = '/path/to/exists_project_path'
+        >>> prd_path = await write_prd("Create a new feature for the application", project_path=project_path)
+        >>> print(prd_path = )
+        '/path/to/project_path/docs/prd/'
+    """
+    from metagpt.actions import UserRequirement
+    from metagpt.context import Context
+    from metagpt.roles import ProductManager
+
+    ctx = Context()
+    if project_path:
+        ctx.config.project_path = Path(project_path)
+        ctx.config.inc = True
+    role = ProductManager(context=ctx)
+    msg = await role.run(with_message=Message(content=idea, cause_by=UserRequirement))
+    await role.run(with_message=msg)
+    return ctx.repo.docs.prd.workdir
+
+
+@register_tool(tags=["software development", "Architect"])
+async def write_design(prd_path: str | Path) -> Path:
+    """Writes a design to the project repository, based on the PRD of the project.
+
+    Args:
+        prd_path (str|Path): The path to the PRD files under the project directory.
+
+    Returns:
+        Path: The path to the system design files under the project directory.
+
+    Example:
+        >>> from metagpt.tools.libs.software_development import write_design
+        >>> prd_path = '/path/to/project_path/docs/prd' # Returned by `write_prd`
+        >>> system_design_path = await write_desgin(prd_path)
+        >>> print(system_design_path)
+        '/path/to/project_path/docs/system_design/'
+
+    """
+    from metagpt.actions import WritePRD
+    from metagpt.context import Context
+    from metagpt.roles import Architect
+
+    ctx = Context()
+    project_path = Path(prd_path).parent.parent
+    ctx.set_repo_dir(project_path)
+
+    role = Architect(context=ctx)
+    await role.run(with_message=Message(content="", cause_by=WritePRD))
+    return ctx.repo.docs.system_design.workdir
+
+
+@register_tool(tags=["software development", "Architect"])
+async def write_project_plan(system_design_path: str | Path) -> Path:
+    """Writes a project plan to the project repository, based on the design of the project.
+
+    Args:
+        system_design_path (str|Path): The path to the system design files under the project directory.
+
+    Returns:
+        Path: The path to task files under the project directory.
+
+    Example:
+        >>> from metagpt.tools.libs.software_development import write_project_plan
+        >>> system_design_path = '/path/to/project_path/docs/system_design/' # Returned by `write_design`
+        >>> task_path = await write_project_plan(system_design_path)
+        >>> print(task_path)
+        '/path/to/project_path/docs/task'
+
+    """
+    from metagpt.actions import WriteDesign
+    from metagpt.context import Context
+    from metagpt.roles import ProjectManager
+
+    ctx = Context()
+    project_path = Path(system_design_path).parent.parent
+    ctx.set_repo_dir(project_path)
+
+    role = ProjectManager(context=ctx)
+    await role.run(with_message=Message(content="", cause_by=WriteDesign))
+    return ctx.repo.docs.task.workdir
+
+
+@register_tool(tags=["software development", "Engineer"])
+async def write_codes(task_path: str | Path, inc: bool = False) -> Path:
+    """Writes codes to the project repository, based on the project plan of the project.
+
+    Args:
+        task_path (str|Path): The path to task files under the project directory.
+        inc (bool, optional): Whether to write incremental codes. Defaults to False.
+
+    Returns:
+        Path: The path to the source code files under the project directory.
+
+    Example:
+        # Write codes to a new project
+        >>> from metagpt.tools.libs.software_development import write_codes
+        >>> task_path = '/path/to/project_path/docs/task' # Returned by `write_project_plan`
+        >>> src_path = await write_codes(task_path)
+        >>> print(src_path)
+        '/path/to/project_path/src/'
+
+        # Write increment codes to the exists project
+        >>> from metagpt.tools.libs.software_development import write_codes
+        >>> task_path = '/path/to/project_path/docs/task' # Returned by `write_prd`
+        >>> src_path = await write_codes(task_path, inc=True)
+        >>> print(src_path)
+        '/path/to/project_path/src/'
+    """
+    from metagpt.actions import WriteTasks
+    from metagpt.context import Context
+    from metagpt.roles import Engineer
+
+    ctx = Context()
+    ctx.config.inc = inc
+    project_path = Path(task_path).parent.parent
+    ctx.set_repo_dir(project_path)
+
+    role = Engineer(context=ctx)
+    msg = Message(content="", cause_by=WriteTasks, send_to=role)
+    me = {any_to_str(role), role.name}
+    while me.intersection(msg.send_to):
+        msg = await role.run(with_message=msg)
+    return ctx.repo.srcs.workdir
+
+
+@register_tool(tags=["software development", "QaEngineer"])
+async def run_qa_test(src_path: str | Path) -> Path:
+    """Run QA test on the project repository.
+
+    Args:
+        src_path (str|Path): The path to the source code files under the project directory.
+
+    Returns:
+        Path: The path to the unit tests under the project directory
+
+    Example:
+        >>> from metagpt.tools.libs.software_development import run_qa_test
+        >>> src_path = '/path/to/project_path/src/' # Returned by `write_codes`
+        >>> test_path = await run_qa_test(src_path)
+        >>> print(test_path)
+        '/path/to/project_path/tests'
+    """
+    from metagpt.actions.summarize_code import SummarizeCode
+    from metagpt.context import Context
+    from metagpt.environment import Environment
+    from metagpt.roles import QaEngineer
+
+    ctx = Context()
+    project_path = Path(src_path).parent
+    ctx.set_repo_dir(project_path)
+    ctx.src_workspace = ctx.git_repo.workdir / ctx.git_repo.workdir.name
+
+    env = Environment(context=ctx)
+    role = QaEngineer(context=ctx)
+    env.add_role(role)
+
+    msg = Message(content="", cause_by=SummarizeCode, send_to=role)
+    env.publish_message(msg)
+
+    while not env.is_idle:
+        await env.run()
+    return ctx.repo.tests.workdir
+
+
+@register_tool(tags=["software development", "Engineer"])
+async def fix_bug(project_path: str | Path, issue: str) -> Path:
+    """Fix bugs in the project repository.
+
+    Args:
+        project_path (str|Path): The path to the project repository.
+        issue (str): Description of the bug or issue.
+
+    Returns:
+        Path: The path to the project directory
+
+    Example:
+        >>> from metagpt.tools.libs.software_development import fix_bug
+        >>> project_path = '/path/to/project_path' # Returned by `write_codes`
+        >>> issue = 'Exception: exception about ...; Bug: bug about ...; Issue: issue about ...'
+        >>> project_path = await fix_bug(project_path=project_path, issue=issue)
+        >>> print(project_path)
+        '/path/to/project_path'
+    """
+    from metagpt.actions.fix_bug import FixBug
+    from metagpt.context import Context
+    from metagpt.roles import Engineer
+
+    ctx = Context()
+    ctx.set_repo_dir(project_path)
+    ctx.src_workspace = ctx.git_repo.workdir / ctx.git_repo.workdir.name
+    await ctx.repo.docs.save(filename=BUGFIX_FILENAME, content=issue)
+    await ctx.repo.docs.save(filename=REQUIREMENT_FILENAME, content="")
+
+    role = Engineer(context=ctx)
+    bug_fix = BugFixContext(filename=BUGFIX_FILENAME)
+    msg = Message(
+        content=bug_fix.model_dump_json(),
+        instruct_content=bug_fix,
+        role="",
+        cause_by=FixBug,
+        sent_from=role,
+        send_to=role,
+    )
+    me = {any_to_str(role), role.name}
+    while me.intersection(msg.send_to):
+        msg = await role.run(with_message=msg)
+    return project_path
+
+
+@register_tool(tags=["software development", "git"])
+async def git_archive(project_path: str | Path) -> str:
+    """Stage and commit changes for the project repository using Git.
+
+    Args:
+        project_path (str|Path): The path to the project repository.
+
+
+    Returns:
+        git log
+
+    Example:
+        >>> from metagpt.tools.libs.software_development import git_archive
+        >>> project_path = '/path/to/project_path' # Returned by `write_prd`
+        >>> git_log = await git_archive(project_path=project_path)
+        >>> print(git_log)
+        commit a221d1c418c07f2b4fc07001e486285ead1a520a (HEAD -> feature/toollib/software_company, geekan/main)
+        Merge: e01afd09 4a72f398
+        Author: Sirui Hong <x@xx.github.com>
+        Date:   Tue Mar 19 15:16:03 2024 +0800
+        Merge pull request #1037 from iorisa/fixbug/issues/1018
+        fixbug: #1018
+
+    """
+    from metagpt.context import Context
+
+    ctx = Context()
+    ctx.set_repo_dir(project_path)
+    ctx.git_repo.archive()
+    return ctx.git_repo.log()
+
+
+@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.
+
+    Args:
+        url (str): The Git project URL, such as "https://github.com/geekan/MetaGPT.git".
+
+    Returns:
+        Path: The path of the formatted project.
+
+    Example:
+        # The Git project URL to input
+        >>> git_url = "https://github.com/geekan/MetaGPT.git"
+
+        # Import the Git repository and get the formatted project path
+        >>> formatted_project_path = await import_git_repo(git_url)
+        >>> print("Formatted project path:", formatted_project_path)
+        /PATH/TO/THE/FORMMATTED/PROJECT
+    """
+    from metagpt.actions.import_repo import ImportRepo
+    from metagpt.context import Context
+
+    ctx = Context()
+    action = ImportRepo(repo_path=url, context=ctx)
+    await action.run()
+    return ctx.repo.workdir