Skip to content

Copier

copier

Copier adapter for template-based scaffolding.

CopierAdapter

Adapter for Copier template operations.

Wraps Copier's run_copy function with a Pydantic-based interface and returns structured ScaffoldResult.

Source code in packages/axm-init/src/axm_init/adapters/copier.py 
 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
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
class CopierAdapter:
    """Adapter for Copier template operations.

    Wraps Copier's run_copy function with a Pydantic-based interface
    and returns structured ScaffoldResult.
    """

    @staticmethod
    def _do_copy(config: CopierConfig) -> None:
        """Run copier, offloading to a thread if an event loop is active.

        Copier (via prompt_toolkit) calls ``asyncio.run()`` internally.
        When we are already inside an async event loop (e.g. MCP server),
        this raises ``RuntimeError: asyncio.run() cannot be called from
        a running event loop``.  The fix: detect the running loop and
        execute the blocking copy in a **separate thread** which gets
        its own event loop context.
        """

        def _run() -> None:
            run_copy(
                src_path=str(config.template_path),
                dst_path=config.destination,
                data=config.data,
                defaults=config.defaults,
                overwrite=config.overwrite,
                unsafe=config.trust_template,
            )

        try:
            asyncio.get_running_loop()
        except RuntimeError:
            # No event loop — safe to call directly (CLI context).
            _run()
        else:
            # Inside an event loop (MCP server) — offload to a thread.
            with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
                future = pool.submit(_run)
                future.result()  # propagate exceptions

    def copy(self, config: CopierConfig) -> ScaffoldResult:
        """Execute Copier copy operation.

        Redirects stdout/stderr so that post-copy tasks (git init,
        uv sync, pre-commit install) don't pollute the parent process
        stdio — critical when running inside an MCP server.

        File descriptors are individually guarded so that a failure at
        any point (e.g. fd limit reached) cannot leak previously
        acquired descriptors.

        Args:
            config: Copier configuration with template path, destination, and data.

        Returns:
            ScaffoldResult with success status and path.
        """
        old_stdout, old_stderr = sys.stdout, sys.stderr
        devnull = -1
        old_fd_out = -1
        old_fd_err = -1

        def _cleanup_fds() -> None:
            """Close any fds that were successfully acquired (idempotent)."""
            nonlocal devnull, old_fd_out, old_fd_err
            if old_fd_out != -1:
                os.dup2(old_fd_out, 1)
                os.close(old_fd_out)
                old_fd_out = -1
            if old_fd_err != -1:
                os.dup2(old_fd_err, 2)
                os.close(old_fd_err)
                old_fd_err = -1
            if devnull != -1:
                os.close(devnull)
                devnull = -1
            sys.stdout = old_stdout
            sys.stderr = old_stderr

        try:
            # Redirect stdout/stderr to prevent subprocess output from
            # corrupting MCP JSON-RPC stdio transport.
            sys.stdout = StringIO()
            sys.stderr = StringIO()
            devnull = os.open(os.devnull, os.O_WRONLY)
            old_fd_out = os.dup(1)
            old_fd_err = os.dup(2)
            os.dup2(devnull, 1)
            os.dup2(devnull, 2)
            if config.trust_template:
                logger.warning(
                    "Running Copier with unsafe=True — template may execute "
                    "arbitrary post-copy tasks."
                )
            try:
                self._do_copy(config)
            finally:
                _cleanup_fds()
            # Walk destination to collect created files, excluding noise
            # from post-copy tasks (.git, .venv, __pycache__, node_modules).
            _excluded = {
                ".git",
                ".venv",
                "__pycache__",
                "node_modules",
                ".mypy_cache",
            }
            created: list[str] = sorted(
                str(p.relative_to(config.destination))
                for p in config.destination.rglob("*")
                if p.is_file()
                and not any(
                    part in _excluded or part.startswith(".")
                    for part in p.relative_to(config.destination).parts[:-1]
                )
            )
            return ScaffoldResult(
                success=True,
                path=str(config.destination),
                message="Project scaffolded via Copier",
                files_created=created,
            )
        except Exception as e:
            _cleanup_fds()
            return ScaffoldResult(
                success=False,
                path=str(config.destination),
                message=f"Copier failed: {e}",
            )
copy(config)

Execute Copier copy operation.

Redirects stdout/stderr so that post-copy tasks (git init, uv sync, pre-commit install) don't pollute the parent process stdio — critical when running inside an MCP server.

File descriptors are individually guarded so that a failure at any point (e.g. fd limit reached) cannot leak previously acquired descriptors.

Parameters:

Name Type Description Default
config CopierConfig

Copier configuration with template path, destination, and data.

 required

Returns:

Type Description
ScaffoldResult

ScaffoldResult with success status and path.
Source code in packages/axm-init/src/axm_init/adapters/copier.py 
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
def copy(self, config: CopierConfig) -> ScaffoldResult:
    """Execute Copier copy operation.

    Redirects stdout/stderr so that post-copy tasks (git init,
    uv sync, pre-commit install) don't pollute the parent process
    stdio — critical when running inside an MCP server.

    File descriptors are individually guarded so that a failure at
    any point (e.g. fd limit reached) cannot leak previously
    acquired descriptors.

    Args:
        config: Copier configuration with template path, destination, and data.

    Returns:
        ScaffoldResult with success status and path.
    """
    old_stdout, old_stderr = sys.stdout, sys.stderr
    devnull = -1
    old_fd_out = -1
    old_fd_err = -1

    def _cleanup_fds() -> None:
        """Close any fds that were successfully acquired (idempotent)."""
        nonlocal devnull, old_fd_out, old_fd_err
        if old_fd_out != -1:
            os.dup2(old_fd_out, 1)
            os.close(old_fd_out)
            old_fd_out = -1
        if old_fd_err != -1:
            os.dup2(old_fd_err, 2)
            os.close(old_fd_err)
            old_fd_err = -1
        if devnull != -1:
            os.close(devnull)
            devnull = -1
        sys.stdout = old_stdout
        sys.stderr = old_stderr

    try:
        # Redirect stdout/stderr to prevent subprocess output from
        # corrupting MCP JSON-RPC stdio transport.
        sys.stdout = StringIO()
        sys.stderr = StringIO()
        devnull = os.open(os.devnull, os.O_WRONLY)
        old_fd_out = os.dup(1)
        old_fd_err = os.dup(2)
        os.dup2(devnull, 1)
        os.dup2(devnull, 2)
        if config.trust_template:
            logger.warning(
                "Running Copier with unsafe=True — template may execute "
                "arbitrary post-copy tasks."
            )
        try:
            self._do_copy(config)
        finally:
            _cleanup_fds()
        # Walk destination to collect created files, excluding noise
        # from post-copy tasks (.git, .venv, __pycache__, node_modules).
        _excluded = {
            ".git",
            ".venv",
            "__pycache__",
            "node_modules",
            ".mypy_cache",
        }
        created: list[str] = sorted(
            str(p.relative_to(config.destination))
            for p in config.destination.rglob("*")
            if p.is_file()
            and not any(
                part in _excluded or part.startswith(".")
                for part in p.relative_to(config.destination).parts[:-1]
            )
        )
        return ScaffoldResult(
            success=True,
            path=str(config.destination),
            message="Project scaffolded via Copier",
            files_created=created,
        )
    except Exception as e:
        _cleanup_fds()
        return ScaffoldResult(
            success=False,
            path=str(config.destination),
            message=f"Copier failed: {e}",
        )

CopierConfig

Bases: BaseModel

Configuration for Copier execution.

Source code in packages/axm-init/src/axm_init/adapters/copier.py 
22
23
24
25
26
27
28
29
30
31
32
class CopierConfig(BaseModel):
    """Configuration for Copier execution."""

    template_path: Path
    destination: Path
    data: dict[str, Any]
    defaults: bool = True
    overwrite: bool = False
    trust_template: bool = False

    model_config = {"extra": "forbid"}