Skip to content

Python tool API

This source-derived reference renders the registered tool modules. Import tool classes from their defining modules; the package root exports only version metadata. See tool contracts for arguments and core helpers for internal models and functions.

axm_git

axm-git — Git workflow automation for AXM agents.

commit_preflight

GitPreflightTool — show working tree status for agent decision-making.

GitPreflightTool

Bases: AXMTool

Report working tree changes so the agent can plan commits.

Registered as git_preflight via axm.tools entry point.

Source code in packages/axm-git/src/axm_git/tools/commit_preflight.py
Python
class GitPreflightTool(AXMTool):
    """Report working tree changes so the agent can plan commits.

    Registered as ``git_preflight`` via axm.tools entry point.
    """

    expose_directly = True
    domain = "git"
    tags = frozenset({"status", "diff", "preflight"})

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

    def execute(
        self,
        *,
        path: str = ".",
        diff_lines: int = 200,
        **kwargs: object,
    ) -> ToolResult:
        """Show current working tree status and diff summary.

        Args:
            path: Project root (required).
            diff_lines: Max diff lines to include (default 200, 0 to
                disable).

        Returns:
            ToolResult with file list, statuses, diff stats, and diff content.
        """
        resolved = Path(path).resolve()
        max_diff_lines = diff_lines

        try:
            pathspec, cwd = _resolve_scope(resolved)

            status = run_git(["status", "--porcelain", *pathspec], cwd)
            if status.returncode != 0:
                return not_a_repo_error(status.stderr, resolved)
            files = _parse_status(status.stdout)

            # git diff --stat [-- rel_path] (only when dirty)
            diff_stat_out = ""
            if files:
                diff_stat_out = run_git(
                    ["diff", "--stat", *pathspec], cwd
                ).stdout.strip()

            diff_content, diff_truncated = _collect_diff(pathspec, cwd, max_diff_lines)
        except subprocess.TimeoutExpired as exc:
            return timeout_error_result(exc)

        text = render_text(
            files=files,
            diff_stat=diff_stat_out,
            diff=diff_content,
            diff_truncated=diff_truncated,
            max_diff_lines=max_diff_lines,
        )

        return ToolResult(
            success=True,
            data={
                "files": files,
                "file_count": len(files),
                "diff_stat": diff_stat_out,
                "diff": diff_content,
                "diff_truncated": diff_truncated,
                "clean": len(files) == 0,
            },
            text=text,
        )
name property

Tool name used for MCP registration.

execute(*, path='.', diff_lines=200, **kwargs)

Show current working tree status and diff summary.

Parameters:

Name Type Description Default
path str

Project root (required).

'.'
diff_lines int

Max diff lines to include (default 200, 0 to disable).

200

Returns:

Type Description
ToolResult

ToolResult with file list, statuses, diff stats, and diff content.

Source code in packages/axm-git/src/axm_git/tools/commit_preflight.py
Python
def execute(
    self,
    *,
    path: str = ".",
    diff_lines: int = 200,
    **kwargs: object,
) -> ToolResult:
    """Show current working tree status and diff summary.

    Args:
        path: Project root (required).
        diff_lines: Max diff lines to include (default 200, 0 to
            disable).

    Returns:
        ToolResult with file list, statuses, diff stats, and diff content.
    """
    resolved = Path(path).resolve()
    max_diff_lines = diff_lines

    try:
        pathspec, cwd = _resolve_scope(resolved)

        status = run_git(["status", "--porcelain", *pathspec], cwd)
        if status.returncode != 0:
            return not_a_repo_error(status.stderr, resolved)
        files = _parse_status(status.stdout)

        # git diff --stat [-- rel_path] (only when dirty)
        diff_stat_out = ""
        if files:
            diff_stat_out = run_git(
                ["diff", "--stat", *pathspec], cwd
            ).stdout.strip()

        diff_content, diff_truncated = _collect_diff(pathspec, cwd, max_diff_lines)
    except subprocess.TimeoutExpired as exc:
        return timeout_error_result(exc)

    text = render_text(
        files=files,
        diff_stat=diff_stat_out,
        diff=diff_content,
        diff_truncated=diff_truncated,
        max_diff_lines=max_diff_lines,
    )

    return ToolResult(
        success=True,
        data={
            "files": files,
            "file_count": len(files),
            "diff_stat": diff_stat_out,
            "diff": diff_content,
            "diff_truncated": diff_truncated,
            "clean": len(files) == 0,
        },
        text=text,
    )

render_text(*, files, diff_stat, diff, diff_truncated, max_diff_lines)

Render a compact text summary of preflight results.

Source code in packages/axm-git/src/axm_git/tools/commit_preflight.py
Python
def render_text(
    *,
    files: list[dict[str, str]],
    diff_stat: str,
    diff: str,
    diff_truncated: bool,
    max_diff_lines: int,
) -> str:
    """Render a compact text summary of preflight results."""
    if not files:
        return "git_preflight | clean"

    parts: list[str] = [f"git_preflight | {len(files)} files · dirty", ""]

    for f in files:
        status = f["status"]
        parts.append(f"{status:<{_STATUS_PAD}}{f['path']}")

    if diff_stat:
        parts.append("")
        parts.append(diff_stat)

    if diff:
        parts.append("")
        parts.append(diff)

    if diff_truncated:
        parts.append(f"[diff truncated at {max_diff_lines} lines]")

    return "\n".join(parts)

branch

GitBranchTool — create or checkout git branches.

GitBranchTool

Bases: AXMTool

Create or checkout a git branch in one call.

Registered as git_branch via axm.tools entry point.

Source code in packages/axm-git/src/axm_git/tools/branch.py
Python
class GitBranchTool(AXMTool):
    """Create or checkout a git branch in one call.

    Registered as ``git_branch`` via axm.tools entry point.
    """

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

    def execute(  # type: ignore[override]
        self,
        *,
        name: str,
        from_ref: str | None = None,
        checkout_only: bool = False,
        delete: bool = False,
        path: str = ".",
        **kwargs: object,
    ) -> ToolResult:
        """Create, checkout, or delete a git branch.

        Args:
            name: Branch name (required).
            from_ref: Optional ref to branch from (tag, commit, branch).
            checkout_only: If True, checkout existing branch without creating.
            delete: If True, delete the branch (``git branch -D``) instead of
                creating/checking out. Mutually exclusive with the create path.
            path: Project root directory.

        Returns:
            ToolResult with branch name on success.
        """
        resolved = Path(path).resolve()

        try:
            # Verify this is a git repo.
            check = run_git(["rev-parse", "--git-dir"], resolved)
            if check.returncode != 0:
                repo_err = not_a_repo_error(check.stderr, resolved)
                return ToolResult(
                    success=repo_err.success,
                    error=repo_err.error,
                    data=repo_err.data,
                    text=render_failure_text(
                        error=repo_err.error or "", data=repo_err.data
                    ),
                )

            if delete:
                return self._delete(name, resolved)

            # Build the checkout command.
            if checkout_only:
                cmd = ["checkout", name]
            else:
                cmd = ["checkout", "-b", name]
                if from_ref is not None:
                    cmd.append(from_ref)

            result = run_git(cmd, resolved)
            if result.returncode != 0:
                error = result.stderr.strip() or result.stdout.strip()
                return ToolResult(
                    success=False,
                    error=error,
                    text=render_failure_text(error=error, data=None),
                )

            # Confirm the current branch.
            current = run_git(["branch", "--show-current"], resolved)
            branch = current.stdout.strip()
        except subprocess.TimeoutExpired as exc:
            return timeout_error_result(exc)

        data: dict[str, object] = {"branch": branch}
        return ToolResult(success=True, data=data, text=render_text(data))

    @staticmethod
    def _delete(name: str, resolved: Path) -> ToolResult:
        """Delete branch *name* via ``git branch -D``."""
        result = run_git(["branch", "-D", name], resolved)
        if result.returncode != 0:
            error = result.stderr.strip() or result.stdout.strip()
            return ToolResult(
                success=False,
                error=error,
                text=render_failure_text(error=error, data=None),
            )
        data: dict[str, object] = {"branch": name, "deleted": True}
        return ToolResult(
            success=True, data=data, text=f"git_branch | ✓ | deleted {name}"
        )
name property

Tool name used for MCP registration.

execute(*, name, from_ref=None, checkout_only=False, delete=False, path='.', **kwargs)

Create, checkout, or delete a git branch.

Parameters:

Name Type Description Default
name str

Branch name (required).

required
from_ref str | None

Optional ref to branch from (tag, commit, branch).

None
checkout_only bool

If True, checkout existing branch without creating.

False
delete bool

If True, delete the branch (git branch -D) instead of creating/checking out. Mutually exclusive with the create path.

False
path str

Project root directory.

'.'

Returns:

Type Description
ToolResult

ToolResult with branch name on success.

Source code in packages/axm-git/src/axm_git/tools/branch.py
Python
def execute(  # type: ignore[override]
    self,
    *,
    name: str,
    from_ref: str | None = None,
    checkout_only: bool = False,
    delete: bool = False,
    path: str = ".",
    **kwargs: object,
) -> ToolResult:
    """Create, checkout, or delete a git branch.

    Args:
        name: Branch name (required).
        from_ref: Optional ref to branch from (tag, commit, branch).
        checkout_only: If True, checkout existing branch without creating.
        delete: If True, delete the branch (``git branch -D``) instead of
            creating/checking out. Mutually exclusive with the create path.
        path: Project root directory.

    Returns:
        ToolResult with branch name on success.
    """
    resolved = Path(path).resolve()

    try:
        # Verify this is a git repo.
        check = run_git(["rev-parse", "--git-dir"], resolved)
        if check.returncode != 0:
            repo_err = not_a_repo_error(check.stderr, resolved)
            return ToolResult(
                success=repo_err.success,
                error=repo_err.error,
                data=repo_err.data,
                text=render_failure_text(
                    error=repo_err.error or "", data=repo_err.data
                ),
            )

        if delete:
            return self._delete(name, resolved)

        # Build the checkout command.
        if checkout_only:
            cmd = ["checkout", name]
        else:
            cmd = ["checkout", "-b", name]
            if from_ref is not None:
                cmd.append(from_ref)

        result = run_git(cmd, resolved)
        if result.returncode != 0:
            error = result.stderr.strip() or result.stdout.strip()
            return ToolResult(
                success=False,
                error=error,
                text=render_failure_text(error=error, data=None),
            )

        # Confirm the current branch.
        current = run_git(["branch", "--show-current"], resolved)
        branch = current.stdout.strip()
    except subprocess.TimeoutExpired as exc:
        return timeout_error_result(exc)

    data: dict[str, object] = {"branch": branch}
    return ToolResult(success=True, data=data, text=render_text(data))

commit

GitCommitTool — batched atomic commits with commit-hook handling.

GitCommitTool

Bases: AXMTool

Execute one or more atomic commits in a single call.

Each commit in the batch is processed sequentially: stage files, run git commit (the repo's commit hooks fire automatically), and capture the result. If a commit fails (e.g. a hook rejects it), processing stops and the error is returned alongside any commits that already succeeded.

When a commit hook auto-fixes files (e.g. ruff --fix), the tool automatically re-stages and retries the commit once.

Verdict-Carrying Patch invariant — when a hook mutates staged content on the autofix-retry path, ToolResult.data reports exactly which files changed under hook_autofixed_files: list[str] (repo-root relative). The field is always present: it is an empty list on the clean path (no hooks, or no mutation) and never None. This lets a consumer see that the patch it committed is not byte-for-byte the patch it staged — the commit still carries a truthful verdict of what landed.

The hook runner is whatever the repo installs at .git/hooks/pre-commit (pre-commit OR prek); axm-git never invokes the runner directly — git does.

Registered as git_commit via axm.tools entry point.

Source code in packages/axm-git/src/axm_git/tools/commit.py
Python
class GitCommitTool(AXMTool):
    """Execute one or more atomic commits in a single call.

    Each commit in the batch is processed sequentially: stage files,
    run ``git commit`` (the repo's commit hooks fire automatically), and
    capture the result.  If a commit fails (e.g. a hook rejects it),
    processing stops and the error is returned alongside any commits
    that already succeeded.

    When a commit hook auto-fixes files (e.g. ruff ``--fix``),
    the tool automatically re-stages and retries the commit once.

    **Verdict-Carrying Patch invariant** — when a hook mutates staged
    content on the autofix-retry path, ``ToolResult.data`` reports exactly
    which files changed under ``hook_autofixed_files: list[str]`` (repo-root
    relative). The field is *always present*: it is an empty list on the
    clean path (no hooks, or no mutation) and never ``None``. This lets a
    consumer see that the patch it committed is not byte-for-byte the patch
    it staged — the commit still carries a truthful verdict of what landed.

    The hook runner is whatever the repo installs at
    ``.git/hooks/pre-commit`` (pre-commit OR prek); axm-git never
    invokes the runner directly — git does.

    Registered as ``git_commit`` via axm.tools entry point.
    """

    expose_directly = True
    domain = "git"
    tags = frozenset({"commit", "stage", "conventional"})

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

    def execute(
        self,
        *,
        path: str = ".",
        commits: list[dict[str, object]] | None = None,
        profile: str | None = None,
        strict: bool = False,
        **kwargs: object,
    ) -> ToolResult:
        """Execute batched commits.

        Args:
            path: Project root (required).
            commits: List of commit specs, each a dict with keys:
                - ``files`` (list[str]): Files to stage.
                - ``message`` (str): Commit summary line.
                - ``body`` (str, optional): Commit body.
            profile: Optional identity profile name. Overrides
                schedule-based resolution from ``git-profiles.toml``.
            strict: When ``True``, a non-Conventional-Commit message is a
                hard failure instead of a warning. Defaults to ``False``
                (warn-by-default guardrail).

        Returns:
            ToolResult with a list of committed results, an ``author`` key
            (``{name, email}`` or ``None``), and ``hook_autofixed_files``
            (``list[str]``, repo-root relative) naming any staged files a
            commit hook mutated during the autofix-retry — always present,
            ``[]`` when no hook auto-fixed anything.
        """
        resolved = Path(path).resolve()
        commit_list: list[dict[str, object]] = commits or []
        total = len(commit_list)

        if not commit_list:
            error = "No commits provided"
            return ToolResult(
                success=False,
                error=error,
                text=render_failure_text(error=error, data=None),
            )

        try:
            # Fail fast with suggestions if not a git repo
            check = run_git(["rev-parse", "--git-dir"], resolved)
            if check.returncode != 0:
                repo_err = not_a_repo_error(check.stderr, resolved)
                return ToolResult(
                    success=repo_err.success,
                    error=repo_err.error,
                    data=repo_err.data,
                    text=render_failure_text(
                        error=repo_err.error or "", data=repo_err.data
                    ),
                )

            # Resolve identity once for the entire batch
            identity = resolve_identity(resolved, profile_override=profile)
            identity_args = author_args(identity)

            results: list[dict[str, object]] = []
            autofixed: list[str] = []

            for i, spec in enumerate(commit_list):
                failure = _process_single_commit(
                    spec,
                    i + 1,
                    identity_args,
                    resolved,
                    results,
                    total,
                    autofixed,
                    strict=strict,
                )
                if failure:
                    return failure
        except subprocess.TimeoutExpired as exc:
            return timeout_error_result(exc)

        data = {
            "results": results,
            "total": len(results),
            "succeeded": len(results),
            # Verdict-Carrying Patch invariant: the paths a commit hook
            # auto-fixed during the re-stage + retry (deduplicated, sorted);
            # always present, ``[]`` on the clean path (AC1/AC2).
            "hook_autofixed_files": sorted(set(autofixed)),
            "author": (
                {"name": identity.name, "email": identity.email} if identity else None
            ),
        }
        return ToolResult(success=True, data=data, text=render_text(data))
name property

Tool name used for MCP registration.

execute(*, path='.', commits=None, profile=None, strict=False, **kwargs)

Execute batched commits.

Parameters:

Name Type Description Default
path str

Project root (required).

'.'
commits list[dict[str, object]] | None

List of commit specs, each a dict with keys: - files (list[str]): Files to stage. - message (str): Commit summary line. - body (str, optional): Commit body.

None
profile str | None

Optional identity profile name. Overrides schedule-based resolution from git-profiles.toml.

None
strict bool

When True, a non-Conventional-Commit message is a hard failure instead of a warning. Defaults to False (warn-by-default guardrail).

False

Returns:

Type Description
ToolResult

ToolResult with a list of committed results, an author key

ToolResult

({name, email} or None), and hook_autofixed_files

ToolResult

(list[str], repo-root relative) naming any staged files a

ToolResult

commit hook mutated during the autofix-retry — always present,

ToolResult

[] when no hook auto-fixed anything.

Source code in packages/axm-git/src/axm_git/tools/commit.py
Python
def execute(
    self,
    *,
    path: str = ".",
    commits: list[dict[str, object]] | None = None,
    profile: str | None = None,
    strict: bool = False,
    **kwargs: object,
) -> ToolResult:
    """Execute batched commits.

    Args:
        path: Project root (required).
        commits: List of commit specs, each a dict with keys:
            - ``files`` (list[str]): Files to stage.
            - ``message`` (str): Commit summary line.
            - ``body`` (str, optional): Commit body.
        profile: Optional identity profile name. Overrides
            schedule-based resolution from ``git-profiles.toml``.
        strict: When ``True``, a non-Conventional-Commit message is a
            hard failure instead of a warning. Defaults to ``False``
            (warn-by-default guardrail).

    Returns:
        ToolResult with a list of committed results, an ``author`` key
        (``{name, email}`` or ``None``), and ``hook_autofixed_files``
        (``list[str]``, repo-root relative) naming any staged files a
        commit hook mutated during the autofix-retry — always present,
        ``[]`` when no hook auto-fixed anything.
    """
    resolved = Path(path).resolve()
    commit_list: list[dict[str, object]] = commits or []
    total = len(commit_list)

    if not commit_list:
        error = "No commits provided"
        return ToolResult(
            success=False,
            error=error,
            text=render_failure_text(error=error, data=None),
        )

    try:
        # Fail fast with suggestions if not a git repo
        check = run_git(["rev-parse", "--git-dir"], resolved)
        if check.returncode != 0:
            repo_err = not_a_repo_error(check.stderr, resolved)
            return ToolResult(
                success=repo_err.success,
                error=repo_err.error,
                data=repo_err.data,
                text=render_failure_text(
                    error=repo_err.error or "", data=repo_err.data
                ),
            )

        # Resolve identity once for the entire batch
        identity = resolve_identity(resolved, profile_override=profile)
        identity_args = author_args(identity)

        results: list[dict[str, object]] = []
        autofixed: list[str] = []

        for i, spec in enumerate(commit_list):
            failure = _process_single_commit(
                spec,
                i + 1,
                identity_args,
                resolved,
                results,
                total,
                autofixed,
                strict=strict,
            )
            if failure:
                return failure
    except subprocess.TimeoutExpired as exc:
        return timeout_error_result(exc)

    data = {
        "results": results,
        "total": len(results),
        "succeeded": len(results),
        # Verdict-Carrying Patch invariant: the paths a commit hook
        # auto-fixed during the re-stage + retry (deduplicated, sorted);
        # always present, ``[]`` on the clean path (AC1/AC2).
        "hook_autofixed_files": sorted(set(autofixed)),
        "author": (
            {"name": identity.name, "email": identity.email} if identity else None
        ),
    }
    return ToolResult(success=True, data=data, text=render_text(data))

clone

GitCloneTool — clone a remote or local git repository.

GitCloneTool

Bases: AXMTool

Clone a git repository into a local directory.

Registered as git_clone via axm.tools entry point.

Source code in packages/axm-git/src/axm_git/tools/clone.py
Python
class GitCloneTool(AXMTool):
    """Clone a git repository into a local directory.

    Registered as ``git_clone`` via axm.tools entry point.
    """

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

    def execute(  # type: ignore[override]
        self,
        *,
        url: str,
        dest: str,
        path: str = ".",
        **kwargs: object,
    ) -> ToolResult:
        """Clone *url* into *dest* relative to *path*.

        Args:
            url: Repository URL or local path to clone from.
            dest: Destination directory name (relative to *path*).
            path: Parent directory in which to create the clone
                (default: current working directory).

        Returns:
            ToolResult with url, dest, absolute clone path, and
            ``cloned: True`` on success.
        """
        cwd = Path(path).resolve()

        try:
            # Clone can be slow over a network — use timeout=None so large
            # repos are not killed mid-transfer.  Local clones are fast;
            # callers that need a hard limit can wrap this tool themselves.
            result = run_git(["clone", url, dest], cwd, timeout=None)
        except subprocess.TimeoutExpired as exc:
            return timeout_error_result(exc)

        if result.returncode != 0:
            error = result.stderr.strip()
            return ToolResult(
                success=False,
                error=error,
                text=render_failure_text(error=error),
            )

        data: dict[str, object] = {
            "url": url,
            "dest": dest,
            "path": str(cwd / dest),
            "cloned": True,
        }
        return ToolResult(success=True, data=data, text=render_text(data))
name property

Tool name used for MCP registration.

execute(*, url, dest, path='.', **kwargs)

Clone url into dest relative to path.

Parameters:

Name Type Description Default
url str

Repository URL or local path to clone from.

required
dest str

Destination directory name (relative to path).

required
path str

Parent directory in which to create the clone (default: current working directory).

'.'

Returns:

Type Description
ToolResult

ToolResult with url, dest, absolute clone path, and

ToolResult

cloned: True on success.

Source code in packages/axm-git/src/axm_git/tools/clone.py
Python
def execute(  # type: ignore[override]
    self,
    *,
    url: str,
    dest: str,
    path: str = ".",
    **kwargs: object,
) -> ToolResult:
    """Clone *url* into *dest* relative to *path*.

    Args:
        url: Repository URL or local path to clone from.
        dest: Destination directory name (relative to *path*).
        path: Parent directory in which to create the clone
            (default: current working directory).

    Returns:
        ToolResult with url, dest, absolute clone path, and
        ``cloned: True`` on success.
    """
    cwd = Path(path).resolve()

    try:
        # Clone can be slow over a network — use timeout=None so large
        # repos are not killed mid-transfer.  Local clones are fast;
        # callers that need a hard limit can wrap this tool themselves.
        result = run_git(["clone", url, dest], cwd, timeout=None)
    except subprocess.TimeoutExpired as exc:
        return timeout_error_result(exc)

    if result.returncode != 0:
        error = result.stderr.strip()
        return ToolResult(
            success=False,
            error=error,
            text=render_failure_text(error=error),
        )

    data: dict[str, object] = {
        "url": url,
        "dest": dest,
        "path": str(cwd / dest),
        "cloned": True,
    }
    return ToolResult(success=True, data=data, text=render_text(data))

worktree

GitWorktreeTool — add, remove, and list git worktrees.

GitWorktreeTool

Bases: AXMTool

Add, remove, or list git worktrees.

Registered as git_worktree via axm.tools entry point.

Source code in packages/axm-git/src/axm_git/tools/worktree.py
Python
class GitWorktreeTool(AXMTool):
    """Add, remove, or list git worktrees.

    Registered as ``git_worktree`` via axm.tools entry point.
    """

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

    def execute(  # type: ignore[override]  # noqa: PLR0913
        self,
        *,
        action: str,
        path: str = ".",
        worktree_path: str | None = None,
        branch: str | None = None,
        base: str | None = None,
        force: bool = False,
        **kwargs: object,
    ) -> ToolResult:
        """Manage git worktrees.

        Args:
            action: One of ``add``, ``remove``, ``list``.
            path: Repository path. Git-root resolution and
                  ``resolve_default_branch`` run here — so a fresh
                  worktree (a sibling directory that does not exist yet)
                  is created correctly.
            worktree_path: For ``add``/``remove``: the worktree location
                  (the future or existing worktree dir). When omitted,
                  *path* doubles as the worktree location (legacy form,
                  which only works when that dir already lives inside a
                  repo).
            branch: Branch name for ``add`` action.
            base: Base ref for ``add`` (default: the repo's resolved
                  default branch).
            force: Force removal for ``remove`` action.

        Returns:
            ToolResult with worktree data on success.
        """
        resolved = Path(path).resolve()
        # When worktree_path is given, *path* is the repo (an existing
        # dir) and *worktree_path* is the target. Otherwise legacy: the
        # sole path is both repo-locator and worktree target.
        wt_target = Path(worktree_path).resolve() if worktree_path else resolved
        repo_locator = resolved if worktree_path else wt_target

        match action:
            case "list":
                return self._list(resolved)
            case "add":
                try:
                    effective_base = base or resolve_default_branch(repo_locator)
                except FileNotFoundError:
                    return _not_a_repo_result(repo_locator)
                return self._add(
                    repo_locator,
                    wt_target,
                    branch=branch,
                    base=effective_base,
                )
            case "remove":
                return self._remove(repo_locator, wt_target, force=force)
            case _:
                error = f"Invalid action {action!r}. Use 'add', 'remove', or 'list'."
                return ToolResult(
                    success=False,
                    error=error,
                    text=render_failure_text(error=error, data=None),
                )

    def _list(self, path: Path) -> ToolResult:
        """List all worktrees."""
        git_root = find_git_root(path)
        if git_root is None:
            return _not_a_repo_result(path)

        try:
            result = run_git(["worktree", "list", "--porcelain"], git_root)
        except subprocess.TimeoutExpired as exc:
            return timeout_error_result(exc)
        if result.returncode != 0:
            return _git_error_result(result)

        worktrees = _parse_worktree_porcelain(result.stdout)
        data: dict[str, object] = {"worktrees": worktrees}
        return ToolResult(success=True, data=data, text=render_list_text(data))

    def _add(
        self,
        repo: Path,
        wt_target: Path,
        *,
        branch: str | None,
        base: str,
    ) -> ToolResult:
        """Add a new worktree at *wt_target*, resolving git from *repo*."""
        git_root = find_git_root(repo)
        if git_root is None:
            return _not_a_repo_result(repo)

        cmd: list[str] = ["worktree", "add"]
        if branch:
            cmd.extend(["-b", branch])
        cmd.append(str(wt_target))
        cmd.append(base)

        try:
            result = run_git(cmd, git_root)
        except subprocess.TimeoutExpired as exc:
            return timeout_error_result(exc)
        if result.returncode != 0:
            return _git_error_result(result)

        data: dict[str, object] = {
            "path": str(wt_target),
            "branch": branch or base,
            "base": base,
        }
        return ToolResult(success=True, data=data, text=render_add_text(data))

    def _remove(self, repo: Path, wt_target: Path, *, force: bool) -> ToolResult:
        """Remove worktree *wt_target*, resolving git from *repo*."""
        git_root = find_git_root(repo)
        if git_root is None:
            return _not_a_repo_result(repo)

        cmd: list[str] = ["worktree", "remove", str(wt_target)]
        if force:
            cmd.append("--force")

        try:
            result = run_git(cmd, git_root)
        except subprocess.TimeoutExpired as exc:
            return timeout_error_result(exc)
        if result.returncode != 0:
            return _git_error_result(result)

        data: dict[str, object] = {"removed": str(wt_target)}
        return ToolResult(success=True, data=data, text=render_remove_text(data))
name property

Tool name used for MCP registration.

execute(*, action, path='.', worktree_path=None, branch=None, base=None, force=False, **kwargs)

Manage git worktrees.

Parameters:

Name Type Description Default
action str

One of add, remove, list.

required
path str

Repository path. Git-root resolution and resolve_default_branch run here — so a fresh worktree (a sibling directory that does not exist yet) is created correctly.

'.'
worktree_path str | None

For add/remove: the worktree location (the future or existing worktree dir). When omitted, path doubles as the worktree location (legacy form, which only works when that dir already lives inside a repo).

None
branch str | None

Branch name for add action.

None
base str | None

Base ref for add (default: the repo's resolved default branch).

None
force bool

Force removal for remove action.

False

Returns:

Type Description
ToolResult

ToolResult with worktree data on success.

Source code in packages/axm-git/src/axm_git/tools/worktree.py
Python
def execute(  # type: ignore[override]  # noqa: PLR0913
    self,
    *,
    action: str,
    path: str = ".",
    worktree_path: str | None = None,
    branch: str | None = None,
    base: str | None = None,
    force: bool = False,
    **kwargs: object,
) -> ToolResult:
    """Manage git worktrees.

    Args:
        action: One of ``add``, ``remove``, ``list``.
        path: Repository path. Git-root resolution and
              ``resolve_default_branch`` run here — so a fresh
              worktree (a sibling directory that does not exist yet)
              is created correctly.
        worktree_path: For ``add``/``remove``: the worktree location
              (the future or existing worktree dir). When omitted,
              *path* doubles as the worktree location (legacy form,
              which only works when that dir already lives inside a
              repo).
        branch: Branch name for ``add`` action.
        base: Base ref for ``add`` (default: the repo's resolved
              default branch).
        force: Force removal for ``remove`` action.

    Returns:
        ToolResult with worktree data on success.
    """
    resolved = Path(path).resolve()
    # When worktree_path is given, *path* is the repo (an existing
    # dir) and *worktree_path* is the target. Otherwise legacy: the
    # sole path is both repo-locator and worktree target.
    wt_target = Path(worktree_path).resolve() if worktree_path else resolved
    repo_locator = resolved if worktree_path else wt_target

    match action:
        case "list":
            return self._list(resolved)
        case "add":
            try:
                effective_base = base or resolve_default_branch(repo_locator)
            except FileNotFoundError:
                return _not_a_repo_result(repo_locator)
            return self._add(
                repo_locator,
                wt_target,
                branch=branch,
                base=effective_base,
            )
        case "remove":
            return self._remove(repo_locator, wt_target, force=force)
        case _:
            error = f"Invalid action {action!r}. Use 'add', 'remove', or 'list'."
            return ToolResult(
                success=False,
                error=error,
                text=render_failure_text(error=error, data=None),
            )

merge

GitMergeTool — squash-merge a branch into a target branch.

GitMergeTool

Bases: AXMTool

Squash-merge a branch into a target branch and commit.

Checks out target_branch, runs git merge --squash <branch>, then commits the squashed changes (honouring the identity-profile system). Registered as git_merge via axm.tools entry point.

Source code in packages/axm-git/src/axm_git/tools/merge.py
Python
class GitMergeTool(AXMTool):
    """Squash-merge a branch into a target branch and commit.

    Checks out *target_branch*, runs ``git merge --squash <branch>``, then
    commits the squashed changes (honouring the identity-profile system).
    Registered as ``git_merge`` via axm.tools entry point.
    """

    domain = "git"
    tags = frozenset({"merge", "squash", "branch"})

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

    def execute(  # type: ignore[override]
        self,
        *,
        branch: str,
        target_branch: str | None = None,
        message: str | None = None,
        path: str = ".",
        **kwargs: object,
    ) -> ToolResult:
        """Squash-merge *branch* into *target_branch*.

        Args:
            branch: The branch to merge in (required).
            target_branch: The branch to merge into (default: the repo's
                resolved default branch).
            message: Commit message for the squash commit. Defaults to
                ``Merge <branch> (squash)``.
            path: Repository path.

        Returns:
            ToolResult with ``merged``, ``into`` and ``message`` on success.
        """
        resolved = Path(path).resolve()
        target_branch = target_branch or resolve_default_branch(resolved)
        msg = message or f"Merge {branch} (squash)"
        try:
            precheck = _check_clean_tree(resolved)
            if precheck is not None:
                return precheck

            failure = _run_merge_steps(branch, target_branch, msg, resolved)
            if failure is not None:
                return failure
        except subprocess.TimeoutExpired as exc:
            return timeout_error_result(exc)

        data: dict[str, object] = {
            "merged": branch,
            "into": target_branch,
            "message": msg,
        }
        return ToolResult(success=True, data=data, text=render_text(data))
name property

Tool name used for MCP registration.

execute(*, branch, target_branch=None, message=None, path='.', **kwargs)

Squash-merge branch into target_branch.

Parameters:

Name Type Description Default
branch str

The branch to merge in (required).

required
target_branch str | None

The branch to merge into (default: the repo's resolved default branch).

None
message str | None

Commit message for the squash commit. Defaults to Merge <branch> (squash).

None
path str

Repository path.

'.'

Returns:

Type Description
ToolResult

ToolResult with merged, into and message on success.

Source code in packages/axm-git/src/axm_git/tools/merge.py
Python
def execute(  # type: ignore[override]
    self,
    *,
    branch: str,
    target_branch: str | None = None,
    message: str | None = None,
    path: str = ".",
    **kwargs: object,
) -> ToolResult:
    """Squash-merge *branch* into *target_branch*.

    Args:
        branch: The branch to merge in (required).
        target_branch: The branch to merge into (default: the repo's
            resolved default branch).
        message: Commit message for the squash commit. Defaults to
            ``Merge <branch> (squash)``.
        path: Repository path.

    Returns:
        ToolResult with ``merged``, ``into`` and ``message`` on success.
    """
    resolved = Path(path).resolve()
    target_branch = target_branch or resolve_default_branch(resolved)
    msg = message or f"Merge {branch} (squash)"
    try:
        precheck = _check_clean_tree(resolved)
        if precheck is not None:
            return precheck

        failure = _run_merge_steps(branch, target_branch, msg, resolved)
        if failure is not None:
            return failure
    except subprocess.TimeoutExpired as exc:
        return timeout_error_result(exc)

    data: dict[str, object] = {
        "merged": branch,
        "into": target_branch,
        "message": msg,
    }
    return ToolResult(success=True, data=data, text=render_text(data))

pull

GitPullTool — pull a remote branch into the local repository.

GitPullTool

Bases: AXMTool

Pull a remote branch (default origin main) into the local repo.

Registered as git_pull via axm.tools entry point.

Source code in packages/axm-git/src/axm_git/tools/pull.py
Python
class GitPullTool(AXMTool):
    """Pull a remote branch (default ``origin main``) into the local repo.

    Registered as ``git_pull`` via axm.tools entry point.
    """

    domain = "git"
    tags = frozenset({"pull", "sync", "remote"})

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

    def execute(
        self,
        *,
        branch: str = "main",
        remote: str = "origin",
        path: str = ".",
        **kwargs: object,
    ) -> ToolResult:
        """Pull *remote*/*branch* into the local repository.

        Args:
            branch: Remote branch to pull (default ``main``).
            remote: Remote name (default ``origin``).
            path: Repository path.

        Returns:
            ToolResult with ``pulled``, ``remote`` and ``branch`` on success.
        """
        resolved = Path(path).resolve()
        try:
            check = run_git(["rev-parse", "--git-dir"], resolved)
            if check.returncode != 0:
                repo_err = not_a_repo_error(check.stderr, resolved)
                return ToolResult(
                    success=repo_err.success,
                    error=repo_err.error,
                    data=repo_err.data,
                    text=render_failure_text(error=repo_err.error or ""),
                )

            result = run_git(["pull", remote, branch], resolved)
            if result.returncode != 0:
                error = result.stderr.strip() or result.stdout.strip()
                return ToolResult(
                    success=False,
                    error=f"git pull failed: {error}",
                    text=render_failure_text(error=error),
                )
        except subprocess.TimeoutExpired as exc:
            return timeout_error_result(exc)

        data: dict[str, object] = {
            "pulled": True,
            "remote": remote,
            "branch": branch,
        }
        return ToolResult(success=True, data=data, text=render_text(data))
name property

Tool name used for MCP registration.

execute(*, branch='main', remote='origin', path='.', **kwargs)

Pull remote/branch into the local repository.

Parameters:

Name Type Description Default
branch str

Remote branch to pull (default main).

'main'
remote str

Remote name (default origin).

'origin'
path str

Repository path.

'.'

Returns:

Type Description
ToolResult

ToolResult with pulled, remote and branch on success.

Source code in packages/axm-git/src/axm_git/tools/pull.py
Python
def execute(
    self,
    *,
    branch: str = "main",
    remote: str = "origin",
    path: str = ".",
    **kwargs: object,
) -> ToolResult:
    """Pull *remote*/*branch* into the local repository.

    Args:
        branch: Remote branch to pull (default ``main``).
        remote: Remote name (default ``origin``).
        path: Repository path.

    Returns:
        ToolResult with ``pulled``, ``remote`` and ``branch`` on success.
    """
    resolved = Path(path).resolve()
    try:
        check = run_git(["rev-parse", "--git-dir"], resolved)
        if check.returncode != 0:
            repo_err = not_a_repo_error(check.stderr, resolved)
            return ToolResult(
                success=repo_err.success,
                error=repo_err.error,
                data=repo_err.data,
                text=render_failure_text(error=repo_err.error or ""),
            )

        result = run_git(["pull", remote, branch], resolved)
        if result.returncode != 0:
            error = result.stderr.strip() or result.stdout.strip()
            return ToolResult(
                success=False,
                error=f"git pull failed: {error}",
                text=render_failure_text(error=error),
            )
    except subprocess.TimeoutExpired as exc:
        return timeout_error_result(exc)

    data: dict[str, object] = {
        "pulled": True,
        "remote": remote,
        "branch": branch,
    }
    return ToolResult(success=True, data=data, text=render_text(data))

push

GitPushTool — push current branch with dirty-check and upstream detection.

GitPushTool

Bases: AXMTool

Push the current branch after verifying a clean working tree.

Registered as git_push via axm.tools entry point.

Source code in packages/axm-git/src/axm_git/tools/push.py
Python
class GitPushTool(AXMTool):
    """Push the current branch after verifying a clean working tree.

    Registered as ``git_push`` via axm.tools entry point.
    """

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

    def execute(
        self,
        *,
        path: str = ".",
        remote: str = "origin",
        set_upstream: bool = True,
        force: bool = False,
        force_unconditional: bool = False,
        **kwargs: object,
    ) -> ToolResult:
        """Push the current branch to a remote.

        Args:
            path: Project root directory.
            remote: Remote name (default ``origin``).
            set_upstream: Auto-set upstream for new branches.
            force: If True, force-push using ``--force-with-lease`` (safe:
                the remote is only overwritten if it has not advanced
                beyond our remote-tracking ref).
            force_unconditional: If True (and ``force`` is set), use a bare
                ``--force`` instead, overwriting the remote unconditionally.
                DATA-LOSS RISK: this discards remote commits we never saw.
                Leave False unless a deliberate hard overwrite is intended.

        Returns:
            ToolResult with branch, remote, and push status.
        """
        resolved = Path(path).resolve()

        try:
            # 1. Verify this is a git repo.
            check = run_git(["rev-parse", "--git-dir"], resolved)
            if check.returncode != 0:
                repo_err = not_a_repo_error(check.stderr, resolved)
                return ToolResult(
                    success=repo_err.success,
                    error=repo_err.error,
                    data=repo_err.data,
                    text=render_failure_text(
                        error=repo_err.error or "", data=repo_err.data
                    ),
                )

            # 2. Dirty check.
            dirty_err = _check_dirty(resolved)
            if dirty_err is not None:
                return dirty_err

            # 3. Get current branch.
            branch_result = run_git(["branch", "--show-current"], resolved)
            branch = branch_result.stdout.strip()
            if not branch:
                error = "No branch checked out (detached HEAD)."
                return ToolResult(
                    success=False,
                    error=error,
                    text=render_failure_text(error=error, data=None),
                )

            # 4. Detect upstream.
            upstream = run_git(
                ["rev-parse", "--abbrev-ref", "@{u}"],
                resolved,
            )
            has_upstream = upstream.returncode == 0

            # 5. Push.
            force_flag = _resolve_force_flag(
                force=force, force_unconditional=force_unconditional
            )
            cmd = _build_push_cmd(
                force_flag=force_flag,
                has_upstream=has_upstream,
                set_upstream=set_upstream,
                remote=remote,
                branch=branch,
            )
            push_result = run_git(cmd, resolved)
            if push_result.returncode != 0:
                error = push_result.stderr.strip() or push_result.stdout.strip()
                return ToolResult(
                    success=False,
                    error=error,
                    text=render_failure_text(error=error, data=None),
                )
        except subprocess.TimeoutExpired as exc:
            return timeout_error_result(exc)

        force_mode = force_flag.removeprefix("--") if force_flag else None
        data: dict[str, object] = {
            "branch": branch,
            "remote": remote,
            "pushed": True,
            "set_upstream": not has_upstream and set_upstream,
            "force_mode": force_mode,
        }
        return ToolResult(success=True, data=data, text=render_text(data))
name property

Tool name used for MCP registration.

execute(*, path='.', remote='origin', set_upstream=True, force=False, force_unconditional=False, **kwargs)

Push the current branch to a remote.

Parameters:

Name Type Description Default
path str

Project root directory.

'.'
remote str

Remote name (default origin).

'origin'
set_upstream bool

Auto-set upstream for new branches.

True
force bool

If True, force-push using --force-with-lease (safe: the remote is only overwritten if it has not advanced beyond our remote-tracking ref).

False
force_unconditional bool

If True (and force is set), use a bare --force instead, overwriting the remote unconditionally. DATA-LOSS RISK: this discards remote commits we never saw. Leave False unless a deliberate hard overwrite is intended.

False

Returns:

Type Description
ToolResult

ToolResult with branch, remote, and push status.

Source code in packages/axm-git/src/axm_git/tools/push.py
Python
def execute(
    self,
    *,
    path: str = ".",
    remote: str = "origin",
    set_upstream: bool = True,
    force: bool = False,
    force_unconditional: bool = False,
    **kwargs: object,
) -> ToolResult:
    """Push the current branch to a remote.

    Args:
        path: Project root directory.
        remote: Remote name (default ``origin``).
        set_upstream: Auto-set upstream for new branches.
        force: If True, force-push using ``--force-with-lease`` (safe:
            the remote is only overwritten if it has not advanced
            beyond our remote-tracking ref).
        force_unconditional: If True (and ``force`` is set), use a bare
            ``--force`` instead, overwriting the remote unconditionally.
            DATA-LOSS RISK: this discards remote commits we never saw.
            Leave False unless a deliberate hard overwrite is intended.

    Returns:
        ToolResult with branch, remote, and push status.
    """
    resolved = Path(path).resolve()

    try:
        # 1. Verify this is a git repo.
        check = run_git(["rev-parse", "--git-dir"], resolved)
        if check.returncode != 0:
            repo_err = not_a_repo_error(check.stderr, resolved)
            return ToolResult(
                success=repo_err.success,
                error=repo_err.error,
                data=repo_err.data,
                text=render_failure_text(
                    error=repo_err.error or "", data=repo_err.data
                ),
            )

        # 2. Dirty check.
        dirty_err = _check_dirty(resolved)
        if dirty_err is not None:
            return dirty_err

        # 3. Get current branch.
        branch_result = run_git(["branch", "--show-current"], resolved)
        branch = branch_result.stdout.strip()
        if not branch:
            error = "No branch checked out (detached HEAD)."
            return ToolResult(
                success=False,
                error=error,
                text=render_failure_text(error=error, data=None),
            )

        # 4. Detect upstream.
        upstream = run_git(
            ["rev-parse", "--abbrev-ref", "@{u}"],
            resolved,
        )
        has_upstream = upstream.returncode == 0

        # 5. Push.
        force_flag = _resolve_force_flag(
            force=force, force_unconditional=force_unconditional
        )
        cmd = _build_push_cmd(
            force_flag=force_flag,
            has_upstream=has_upstream,
            set_upstream=set_upstream,
            remote=remote,
            branch=branch,
        )
        push_result = run_git(cmd, resolved)
        if push_result.returncode != 0:
            error = push_result.stderr.strip() or push_result.stdout.strip()
            return ToolResult(
                success=False,
                error=error,
                text=render_failure_text(error=error, data=None),
            )
    except subprocess.TimeoutExpired as exc:
        return timeout_error_result(exc)

    force_mode = force_flag.removeprefix("--") if force_flag else None
    data: dict[str, object] = {
        "branch": branch,
        "remote": remote,
        "pushed": True,
        "set_upstream": not has_upstream and set_upstream,
        "force_mode": force_mode,
    }
    return ToolResult(success=True, data=data, text=render_text(data))

pr

GitPRTool — create GitHub pull requests via gh CLI.

GitPRTool

Bases: AXMTool

Create a GitHub pull request with optional auto-merge.

Registered as git_pr via axm.tools entry point.

Source code in packages/axm-git/src/axm_git/tools/pr.py
Python
class GitPRTool(AXMTool):
    """Create a GitHub pull request with optional auto-merge.

    Registered as ``git_pr`` via axm.tools entry point.
    """

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

    def execute(  # type: ignore[override]
        self,
        *,
        title: str,
        body: str | None = None,
        base: str | None = None,
        auto_merge: bool = False,
        path: str = ".",
        **kwargs: object,
    ) -> ToolResult:
        """Create a GitHub pull request.

        Args:
            title: PR title (required).
            body: PR body/description.
            base: Base branch (default: the repo's resolved default branch).
            auto_merge: Enable auto-merge with squash (default ``False``).
            path: Repository path.

        Returns:
            ToolResult with ``pr_url`` and ``pr_number`` on success.
        """
        resolved = Path(path).resolve()
        base = base or resolve_default_branch(resolved)

        try:
            precheck = _check_pr_preconditions(resolved)
            if precheck is not None:
                return precheck

            created = _create_pr(title, body, base, resolved)
            if isinstance(created, ToolResult):
                return created
            pr_url, pr_number = created

            auto_merge_ok = _maybe_auto_merge(pr_number, auto_merge, resolved)

            data: dict[str, object] = {
                "pr_url": pr_url,
                "pr_number": pr_number,
                "auto_merge": auto_merge_ok,
            }
            return ToolResult(success=True, data=data, text=render_text(data))
        except subprocess.TimeoutExpired as exc:
            return timeout_error_result(exc)
name property

Tool name used for MCP registration.

execute(*, title, body=None, base=None, auto_merge=False, path='.', **kwargs)

Create a GitHub pull request.

Parameters:

Name Type Description Default
title str

PR title (required).

required
body str | None

PR body/description.

None
base str | None

Base branch (default: the repo's resolved default branch).

None
auto_merge bool

Enable auto-merge with squash (default False).

False
path str

Repository path.

'.'

Returns:

Type Description
ToolResult

ToolResult with pr_url and pr_number on success.

Source code in packages/axm-git/src/axm_git/tools/pr.py
Python
def execute(  # type: ignore[override]
    self,
    *,
    title: str,
    body: str | None = None,
    base: str | None = None,
    auto_merge: bool = False,
    path: str = ".",
    **kwargs: object,
) -> ToolResult:
    """Create a GitHub pull request.

    Args:
        title: PR title (required).
        body: PR body/description.
        base: Base branch (default: the repo's resolved default branch).
        auto_merge: Enable auto-merge with squash (default ``False``).
        path: Repository path.

    Returns:
        ToolResult with ``pr_url`` and ``pr_number`` on success.
    """
    resolved = Path(path).resolve()
    base = base or resolve_default_branch(resolved)

    try:
        precheck = _check_pr_preconditions(resolved)
        if precheck is not None:
            return precheck

        created = _create_pr(title, body, base, resolved)
        if isinstance(created, ToolResult):
            return created
        pr_url, pr_number = created

        auto_merge_ok = _maybe_auto_merge(pr_number, auto_merge, resolved)

        data: dict[str, object] = {
            "pr_url": pr_url,
            "pr_number": pr_number,
            "auto_merge": auto_merge_ok,
        }
        return ToolResult(success=True, data=data, text=render_text(data))
    except subprocess.TimeoutExpired as exc:
        return timeout_error_result(exc)

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

release_diff

GitReleaseDiffTool — read-only SemVer bump decision for a package subdir.

GitReleaseDiffTool

Bases: AXMTool

Summarise commits/diff since the last tag to decide a SemVer bump.

Strictly read-only: issues only log, diff, tag and rev-parse — never creates or pushes a tag. Scopes every log and diff to the resolved package subdir so monorepo attribution is correct. Registered as git_release_diff via axm.tools.

Source code in packages/axm-git/src/axm_git/tools/release_diff.py
Python
class GitReleaseDiffTool(AXMTool):
    """Summarise commits/diff since the last tag to decide a SemVer bump.

    Strictly read-only: issues only ``log``, ``diff``, ``tag`` and
    ``rev-parse`` — never creates or pushes a tag. Scopes every ``log``
    and ``diff`` to the resolved package subdir so monorepo attribution
    is correct. Registered as ``git_release_diff`` via axm.tools.
    """

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

    def execute(self, *, path: str = ".", **kwargs: object) -> ToolResult:
        """Compute a read-only release diff for the package at *path*.

        Args:
            path: Package root (defaults to the current directory).

        Returns:
            ToolResult with current tag, commit summary, diffstat and a
            suggested next version.
        """
        resolved = Path(path).resolve()
        if find_git_root(resolved) is None:
            # False-green guard: without a repo, every read-only git call
            # returns empty, which would masquerade as a clean "first release
            # → 0.1.0". Surface the not-a-repo error instead.
            repo_err = not_a_repo_error("not a git repository", resolved)
            return ToolResult(
                success=repo_err.success,
                error=repo_err.error,
                data=repo_err.data,
                text=render_failure_text(
                    error=repo_err.error or "", data=repo_err.data
                ),
            )
        prefix = get_tag_prefix(resolved)
        try:
            data = self._collect(resolved, prefix)
        except subprocess.TimeoutExpired as exc:
            return _timeout_error_result(exc)
        return ToolResult(success=True, data=data, text=render_text(data))

    def _collect(self, resolved: Path, prefix: str) -> dict[str, object]:
        """Gather all read-only data for the success-path payload."""
        root = find_git_root(resolved)
        subdir = self._subdir(root, resolved)
        cwd = root or resolved
        current_tag = _current_tag(cwd, prefix)
        commits = _scoped_log(cwd, current_tag, subdir)
        files_changed, diffstat = _diffstat(cwd, current_tag, subdir)
        suggested_bump, suggested_next, breaking = _suggest(commits, current_tag)
        return {
            "current_tag": current_tag,
            "suggested_bump": suggested_bump,
            "suggested_next": suggested_next,
            "breaking": breaking,
            "commits_since": commits,
            "counts": _aggregate_counts(commits),
            "files_changed": files_changed,
            "diffstat": diffstat,
            "public_api_touched": _public_api_touched(cwd, current_tag, subdir),
        }

    @staticmethod
    def _subdir(root: Path | None, resolved: Path) -> str:
        """Package path relative to the git root (``.`` when at the root)."""
        if root is None:
            return "."
        try:
            rel = resolved.relative_to(root)
        except ValueError:
            return "."
        return str(rel)
name property

Tool name used for MCP registration.

execute(*, path='.', **kwargs)

Compute a read-only release diff for the package at path.

Parameters:

Name Type Description Default
path str

Package root (defaults to the current directory).

'.'

Returns:

Type Description
ToolResult

ToolResult with current tag, commit summary, diffstat and a

ToolResult

suggested next version.

Source code in packages/axm-git/src/axm_git/tools/release_diff.py
Python
def execute(self, *, path: str = ".", **kwargs: object) -> ToolResult:
    """Compute a read-only release diff for the package at *path*.

    Args:
        path: Package root (defaults to the current directory).

    Returns:
        ToolResult with current tag, commit summary, diffstat and a
        suggested next version.
    """
    resolved = Path(path).resolve()
    if find_git_root(resolved) is None:
        # False-green guard: without a repo, every read-only git call
        # returns empty, which would masquerade as a clean "first release
        # → 0.1.0". Surface the not-a-repo error instead.
        repo_err = not_a_repo_error("not a git repository", resolved)
        return ToolResult(
            success=repo_err.success,
            error=repo_err.error,
            data=repo_err.data,
            text=render_failure_text(
                error=repo_err.error or "", data=repo_err.data
            ),
        )
    prefix = get_tag_prefix(resolved)
    try:
        data = self._collect(resolved, prefix)
    except subprocess.TimeoutExpired as exc:
        return _timeout_error_result(exc)
    return ToolResult(success=True, data=data, text=render_text(data))

tag

GitTagTool — one-shot semver tag: preflight + compute + create + verify + push.

GitTagTool

Bases: AXMTool

Create a semver release tag in one call.

Performs preflight checks (clean tree, CI status), computes the next version from Conventional Commits, creates an annotated tag, verifies hatch-vcs resolution, and pushes to origin.

Registered as git_tag via axm.tools entry point.

Source code in packages/axm-git/src/axm_git/tools/tag.py
Python
class GitTagTool(AXMTool):
    """Create a semver release tag in one call.

    Performs preflight checks (clean tree, CI status), computes the
    next version from Conventional Commits, creates an annotated tag,
    verifies hatch-vcs resolution, and pushes to origin.

    Registered as ``git_tag`` via axm.tools entry point.
    """

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

    def execute(
        self,
        *,
        path: str = ".",
        version: str | None = None,
        **kwargs: object,
    ) -> ToolResult:
        """Create and push a semver tag.

        Args:
            path: Project root (required).
            version: Version override (optional, e.g. ``"v1.0.0"``).

        Returns:
            ToolResult with tag, version, and push status.
        """
        resolved = Path(path).resolve()
        tag_prefix = get_tag_prefix(resolved)

        try:
            # 1. Preflight: repo, clean tree, CI, commits
            result = _preflight(resolved, tag_prefix=tag_prefix)
            if isinstance(result, ToolResult):
                return result
            ci_check, current_tag, commits = result

            # 2. Compute version
            next_version, bump_type, breaking = resolve_version(
                version, current_tag, commits, tag_prefix=tag_prefix
            )
            logger.info(
                "Tagging %s (bump=%s, breaking=%s)",
                next_version,
                bump_type,
                breaking,
            )

            # 3. Create annotated tag
            full_tag = f"{tag_prefix}{next_version}"
            tag_result = run_git(["tag", "-a", full_tag, "-m", full_tag], resolved)
            if tag_result.returncode != 0:
                error = f"Failed to create tag: {tag_result.stderr.strip()}"
                return ToolResult(
                    success=False,
                    error=error,
                    text=render_failure_text(error=error, data=None),
                )

            # 4. Verify hatch-vcs (best-effort)
            resolved_version = None
            pkg_name = detect_package_name(resolved)
            if pkg_name:
                resolved_version = verify_hatch_vcs(resolved, pkg_name)

            # 5. Push tag
            push = run_git(["push", "origin", full_tag], resolved)
        except ValueError as exc:
            error = str(exc)
            return ToolResult(
                success=False,
                error=error,
                text=render_failure_text(error=error, data=None),
            )
        except subprocess.TimeoutExpired as exc:
            return _timeout_error_result(exc)

        pushed = push.returncode == 0
        data: dict[str, object] = {
            "tag": next_version,
            "full_tag": full_tag,
            "bump": bump_type,
            "breaking": breaking,
            "resolved_version": resolved_version,
            "pushed": pushed,
            "ci_check": ci_check,
            "commits_included": len(commits),
            "current_tag": current_tag or "none",
        }
        if not pushed:
            # False-green guard: the annotated tag is created locally but the
            # remote push failed — the Publish CI will never fire. Report the
            # failure (local tag preserved in ``data``) so the warden/agent
            # does not believe the release shipped.
            error = (
                f"Tag {full_tag} created locally but push to origin failed: "
                f"{push.stderr.strip()}"
            )
            return ToolResult(
                success=False,
                error=error,
                data=data,
                text=render_failure_text(error=error, data=data),
            )
        return ToolResult(success=True, data=data, text=render_text(data))
name property

Tool name used for MCP registration.

execute(*, path='.', version=None, **kwargs)

Create and push a semver tag.

Parameters:

Name Type Description Default
path str

Project root (required).

'.'
version str | None

Version override (optional, e.g. "v1.0.0").

None

Returns:

Type Description
ToolResult

ToolResult with tag, version, and push status.

Source code in packages/axm-git/src/axm_git/tools/tag.py
Python
def execute(
    self,
    *,
    path: str = ".",
    version: str | None = None,
    **kwargs: object,
) -> ToolResult:
    """Create and push a semver tag.

    Args:
        path: Project root (required).
        version: Version override (optional, e.g. ``"v1.0.0"``).

    Returns:
        ToolResult with tag, version, and push status.
    """
    resolved = Path(path).resolve()
    tag_prefix = get_tag_prefix(resolved)

    try:
        # 1. Preflight: repo, clean tree, CI, commits
        result = _preflight(resolved, tag_prefix=tag_prefix)
        if isinstance(result, ToolResult):
            return result
        ci_check, current_tag, commits = result

        # 2. Compute version
        next_version, bump_type, breaking = resolve_version(
            version, current_tag, commits, tag_prefix=tag_prefix
        )
        logger.info(
            "Tagging %s (bump=%s, breaking=%s)",
            next_version,
            bump_type,
            breaking,
        )

        # 3. Create annotated tag
        full_tag = f"{tag_prefix}{next_version}"
        tag_result = run_git(["tag", "-a", full_tag, "-m", full_tag], resolved)
        if tag_result.returncode != 0:
            error = f"Failed to create tag: {tag_result.stderr.strip()}"
            return ToolResult(
                success=False,
                error=error,
                text=render_failure_text(error=error, data=None),
            )

        # 4. Verify hatch-vcs (best-effort)
        resolved_version = None
        pkg_name = detect_package_name(resolved)
        if pkg_name:
            resolved_version = verify_hatch_vcs(resolved, pkg_name)

        # 5. Push tag
        push = run_git(["push", "origin", full_tag], resolved)
    except ValueError as exc:
        error = str(exc)
        return ToolResult(
            success=False,
            error=error,
            text=render_failure_text(error=error, data=None),
        )
    except subprocess.TimeoutExpired as exc:
        return _timeout_error_result(exc)

    pushed = push.returncode == 0
    data: dict[str, object] = {
        "tag": next_version,
        "full_tag": full_tag,
        "bump": bump_type,
        "breaking": breaking,
        "resolved_version": resolved_version,
        "pushed": pushed,
        "ci_check": ci_check,
        "commits_included": len(commits),
        "current_tag": current_tag or "none",
    }
    if not pushed:
        # False-green guard: the annotated tag is created locally but the
        # remote push failed — the Publish CI will never fire. Report the
        # failure (local tag preserved in ``data``) so the warden/agent
        # does not believe the release shipped.
        error = (
            f"Tag {full_tag} created locally but push to origin failed: "
            f"{push.stderr.strip()}"
        )
        return ToolResult(
            success=False,
            error=error,
            data=data,
            text=render_failure_text(error=error, data=data),
        )
    return ToolResult(success=True, data=data, text=render_text(data))

check_ci(path)

Check CI status via gh for the CI run matching HEAD.

Returns one of green/red/pending/skipped/error. The status is derived from the run whose headSha matches the current HEAD SHA, never from runs[0] unconditionally: a stale green (HEAD moved past it) or a red on an unrelated commit must not influence the verdict. When no run matches HEAD, returns pending (block tagging until CI exists for the exact commit being tagged).

Source code in packages/axm-git/src/axm_git/tools/tag.py
Python
def check_ci(path: Path) -> str:
    """Check CI status via ``gh`` for the CI run matching HEAD.

    Returns one of green/red/pending/skipped/error. The status is derived
    from the run whose ``headSha`` matches the current HEAD SHA, never from
    ``runs[0]`` unconditionally: a stale green (HEAD moved past it) or a red
    on an unrelated commit must not influence the verdict. When no run
    matches HEAD, returns ``pending`` (block tagging until CI exists for the
    exact commit being tagged).
    """
    if not gh_available():
        return "skipped"
    try:
        ci = run_gh(
            [
                "run",
                "list",
                "--branch",
                resolve_default_branch(path),
                "--limit",
                "3",
                "--json",
                "status,conclusion,headSha",
            ],
            path,
        )
        if ci.returncode != 0 or not ci.stdout.strip():
            return "skipped"
        runs = json.loads(ci.stdout)
        if not runs:
            return "skipped"
        head_sha = _head_sha(path)
        if head_sha is None:
            return "pending"
        for run in runs:
            if _sha_matches(str(run.get("headSha") or ""), head_sha):
                return _verdict(run)
        return "pending"
    except (json.JSONDecodeError, FileNotFoundError):
        return "error"

get_tag_prefix(path)

Read tag prefix from pyproject.toml tag-pattern (e.g. git/).

Returns the prefix string (e.g. "git/") or "" if none.

Source code in packages/axm-git/src/axm_git/tools/tag.py
Python
def get_tag_prefix(path: Path) -> str:
    """Read tag prefix from pyproject.toml ``tag-pattern`` (e.g. ``git/``).

    Returns the prefix string (e.g. ``"git/"``) or ``""`` if none.
    """
    pyproject = path / "pyproject.toml"
    if not pyproject.exists():
        return ""
    try:
        with open(pyproject, "rb") as f:
            data = tomllib.load(f)
        pattern = (
            data.get("tool", {})
            .get("hatch", {})
            .get("version", {})
            .get("tag-pattern", "")
        )
        # Extract prefix before "v" from patterns like "git/v(?P<version>.*)"
        m = re.match(r"^(.+?)v\(", pattern)
        return m.group(1) if m else ""
    except (OSError, tomllib.TOMLDecodeError):
        return ""

resolve_version(version_override, current_tag, commits, *, tag_prefix='')

Resolve the next version tag.

Returns:

Type Description
tuple[str, str, bool]

(next_version, bump_type, breaking).

Source code in packages/axm-git/src/axm_git/tools/tag.py
Python
def resolve_version(
    version_override: str | None,
    current_tag: str | None,
    commits: list[str],
    *,
    tag_prefix: str = "",
) -> tuple[str, str, bool]:
    """Resolve the next version tag.

    Returns:
        ``(next_version, bump_type, breaking)``.
    """
    if version_override:
        v = (
            version_override
            if version_override.startswith("v")
            else f"v{version_override}"
        )
        override_tuple = parse_tag(v)
        current_bare = current_tag
        if current_bare and tag_prefix and current_bare.startswith(tag_prefix):
            current_bare = current_bare[len(tag_prefix) :]
        if current_bare and override_tuple <= parse_tag(current_bare):
            msg = (
                f"Version override {v!r} is not strictly greater than "
                f"the current tag {current_bare!r}"
            )
            raise ValueError(msg)
        return v, "override", False

    base = current_tag or "v0.0.0"
    if tag_prefix and base.startswith(tag_prefix):
        base = base[len(tag_prefix) :]
    bump_result = compute_bump(commits, base)
    return bump_result.next, bump_result.bump, bump_result.breaking

verify_hatch_vcs(path, pkg_name)

Rebuild package and read resolved version (best-effort).

Source code in packages/axm-git/src/axm_git/tools/tag.py
Python
def verify_hatch_vcs(path: Path, pkg_name: str) -> str | None:
    """Rebuild package and read resolved version (best-effort)."""
    try:
        sync = subprocess.run(
            ["uv", "sync", "--reinstall-package", pkg_name],
            cwd=str(path),
            capture_output=True,
            text=True,
            check=False,
            timeout=600,
        )
        if sync.returncode != 0:
            return None
        ver = subprocess.run(
            [
                "uv",
                "run",
                "python",
                "-c",
                f"from importlib.metadata import version; print(version('{pkg_name}'))",
            ],
            cwd=str(path),
            capture_output=True,
            text=True,
            check=False,
            timeout=60,
        )
        if ver.returncode == 0:
            return ver.stdout.strip()
    except (FileNotFoundError, subprocess.TimeoutExpired):
        pass
    return None