Skip to content

Await merge

await_merge

GitAwaitMergeTool — poll a GitHub PR until it is merged or times out.

GitAwaitMergeTool

Bases: AXMTool

Poll a GitHub PR until it reaches the MERGED state.

Blocks, querying gh pr view --json state every interval seconds until the PR is merged, is closed, or timeout elapses. Registered as git_await_merge via axm.tools entry point.

Source code in packages/axm-git/src/axm_git/tools/await_merge.py
Python
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
class GitAwaitMergeTool(AXMTool):
    """Poll a GitHub PR until it reaches the ``MERGED`` state.

    Blocks, querying ``gh pr view --json state`` every *interval* seconds
    until the PR is merged, is closed, or *timeout* elapses. Registered as
    ``git_await_merge`` via axm.tools entry point.
    """

    domain = "git"
    tags = frozenset({"pr", "merge", "poll", "await"})

    @property
    def name(self) -> str:
        """Tool name used for MCP registration."""
        return "git_await_merge"

    def execute(  # type: ignore[override]
        self,
        *,
        pr: str,
        timeout: int = _DEFAULT_TIMEOUT,
        interval: int = _DEFAULT_INTERVAL,
        path: str = ".",
        **kwargs: object,
    ) -> ToolResult:
        """Poll PR *pr* until merged or timed out.

        Args:
            pr: PR number or URL (required).
            timeout: Maximum seconds to wait (default 600).
            interval: Seconds between polls (default 30).
            path: Repository path.

        Returns:
            ToolResult with ``merged=True`` and ``pr_ref`` on success.
        """
        resolved = Path(path).resolve()
        if not gh_available():
            error = "gh CLI not available"
            return ToolResult(
                success=False, error=error, text=render_failure_text(error=error)
            )

        try:
            start = time.monotonic()
            while time.monotonic() - start < timeout:
                state = _poll_pr_state(pr, resolved)
                if state is None:
                    error = f"failed to query PR {pr} state"
                    return ToolResult(
                        success=False,
                        error=error,
                        text=render_failure_text(error=error),
                    )
                if state == "MERGED":
                    data: dict[str, object] = {"merged": True, "pr_ref": pr}
                    return ToolResult(success=True, data=data, text=render_text(data))
                if state == "CLOSED":
                    error = f"PR {pr} was closed without merging"
                    return ToolResult(
                        success=False,
                        error=error,
                        text=render_failure_text(error=error),
                    )
                time.sleep(interval)
        except subprocess.TimeoutExpired as exc:
            return timeout_error_result(exc)

        error = f"PR {pr} not merged after {timeout}s timeout"
        return ToolResult(
            success=False, error=error, text=render_failure_text(error=error)
        )
name property

Tool name used for MCP registration.

execute(*, pr, timeout=_DEFAULT_TIMEOUT, interval=_DEFAULT_INTERVAL, path='.', **kwargs)

Poll PR pr until merged or timed out.

Parameters:

Name Type Description Default
pr str

PR number or URL (required).

 required
timeout int

Maximum seconds to wait (default 600).

 _DEFAULT_TIMEOUT
interval int

Seconds between polls (default 30).

 _DEFAULT_INTERVAL
path str

Repository path.

 '.'

Returns:

Type Description
ToolResult

ToolResult with merged=True and pr_ref on success.
Source code in packages/axm-git/src/axm_git/tools/await_merge.py
Python
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def execute(  # type: ignore[override]
    self,
    *,
    pr: str,
    timeout: int = _DEFAULT_TIMEOUT,
    interval: int = _DEFAULT_INTERVAL,
    path: str = ".",
    **kwargs: object,
) -> ToolResult:
    """Poll PR *pr* until merged or timed out.

    Args:
        pr: PR number or URL (required).
        timeout: Maximum seconds to wait (default 600).
        interval: Seconds between polls (default 30).
        path: Repository path.

    Returns:
        ToolResult with ``merged=True`` and ``pr_ref`` on success.
    """
    resolved = Path(path).resolve()
    if not gh_available():
        error = "gh CLI not available"
        return ToolResult(
            success=False, error=error, text=render_failure_text(error=error)
        )

    try:
        start = time.monotonic()
        while time.monotonic() - start < timeout:
            state = _poll_pr_state(pr, resolved)
            if state is None:
                error = f"failed to query PR {pr} state"
                return ToolResult(
                    success=False,
                    error=error,
                    text=render_failure_text(error=error),
                )
            if state == "MERGED":
                data: dict[str, object] = {"merged": True, "pr_ref": pr}
                return ToolResult(success=True, data=data, text=render_text(data))
            if state == "CLOSED":
                error = f"PR {pr} was closed without merging"
                return ToolResult(
                    success=False,
                    error=error,
                    text=render_failure_text(error=error),
                )
            time.sleep(interval)
    except subprocess.TimeoutExpired as exc:
        return timeout_error_result(exc)

    error = f"PR {pr} not merged after {timeout}s timeout"
    return ToolResult(
        success=False, error=error, text=render_failure_text(error=error)
    )