ant_ai.acp.tools

acp_get_cwd async

acp_get_cwd() -> str

Return the working directory set by the user for this IDE session.

Call this to get the base directory when the user refers to files without giving a full path. Then use os.path.join(cwd, filename) to build the full path and pass it to acp_fs_read_file or acp_terminal_run.

cwd="/Users/alice/myproject", read "README.md" →

acp_fs_read_file("/Users/alice/myproject/README.md")

Returns:

Type	Description
str	Path to the session's working directory (may be absolute or relative
str	depending on what the user typed in the client).

Source code in src/ant_ai/acp/tools.py

@tool
async def acp_get_cwd() -> str:
    """Return the working directory set by the user for this IDE session.

    Call this to get the base directory when the user refers to files without
    giving a full path. Then use os.path.join(cwd, filename) to build the
    full path and pass it to acp_fs_read_file or acp_terminal_run.

    Example: cwd="/Users/alice/myproject", read "README.md" →
        acp_fs_read_file("/Users/alice/myproject/README.md")

    Returns:
        Path to the session's working directory (may be absolute or relative
        depending on what the user typed in the client).
    """
    cwd = _acp_cwd.get()
    if cwd is None:
        raise RuntimeError(
            "Not running inside an ACP prompt turn. "
            "ACP tool functions can only be called during a session/prompt invocation."
        )
    return cwd

acp_list_directory async

acp_list_directory(path: str | None = None) -> str

List files and subdirectories at a path on the agent's filesystem.

Use this to explore the project before reading specific files. Directories are shown with a trailing '/'. Relative paths are resolved against the session working directory. Defaults to the working directory when no path is given.

Parameters:

Name	Type	Description	Default
path	str | None	Directory to list. Absolute or relative to cwd. Defaults to cwd.	None

Returns:

Type	Description
str	Newline-separated entries — directories have a trailing '/'.

Source code in src/ant_ai/acp/tools.py

@tool
async def acp_list_directory(path: str | None = None) -> str:
    """List files and subdirectories at a path on the agent's filesystem.

    Use this to explore the project before reading specific files.
    Directories are shown with a trailing '/'. Relative paths are resolved
    against the session working directory. Defaults to the working directory
    when no path is given.

    Args:
        path: Directory to list. Absolute or relative to cwd. Defaults to cwd.

    Returns:
        Newline-separated entries — directories have a trailing '/'.
    """
    cwd = _acp_cwd.get()
    target = cwd or "." if path is None else _resolve_path(path)

    logger.debug("acp_list_directory: listing '{}'", target)
    try:
        entries = sorted(os.listdir(target))
        lines = [
            e + "/" if os.path.isdir(os.path.join(target, e)) else e for e in entries
        ]
        result = "\n".join(lines) if lines else "(empty directory)"
        logger.debug("acp_list_directory: {} entries", len(lines))
        return result
    except Exception as exc:
        logger.error("acp_list_directory failed: path='{}' error={}", target, exc)
        raise

acp_fs_read_file async

acp_fs_read_file(
    path: str,
    line: int | None = None,
    limit: int | None = None,
) -> str

Read a text file from the IDE's filesystem via ACP.

Relative paths are automatically resolved against the session working directory, so you can pass just a filename like "README.md" and it will be read from the cwd. Use acp_get_cwd() to see what that directory is.

Parameters:

Name	Type	Description	Default
path	str	Path to the file — absolute or relative to the working directory.	required
line	int | None	1-based starting line (optional).	None
limit	int | None	Maximum number of lines to return (optional).	None

Returns:

Type	Description
str	File content as a string.

Source code in src/ant_ai/acp/tools.py

@tool
async def acp_fs_read_file(
    path: str,
    line: int | None = None,
    limit: int | None = None,
) -> str:
    """Read a text file from the IDE's filesystem via ACP.

    Relative paths are automatically resolved against the session working
    directory, so you can pass just a filename like "README.md" and it will
    be read from the cwd. Use acp_get_cwd() to see what that directory is.

    Args:
        path: Path to the file — absolute or relative to the working directory.
        line: 1-based starting line (optional).
        limit: Maximum number of lines to return (optional).

    Returns:
        File content as a string.
    """
    client, session_id = _require_fs_read()
    resolved = _resolve_path(path)
    logger.debug("acp_fs_read_file: path='{}' resolved='{}'", path, resolved)
    try:
        response = await client.read_text_file(
            path=resolved, session_id=session_id, line=line, limit=limit
        )
        logger.debug("acp_fs_read_file: got {} chars", len(response.content))
        return response.content
    except Exception as exc:
        logger.error("acp_fs_read_file failed: path='{}' error={}", resolved, exc)
        raise

acp_fs_write_file async

acp_fs_write_file(path: str, content: str) -> None

Write or overwrite a text file in the IDE's filesystem via ACP.

Relative paths are resolved against the session working directory.

Parameters:

Name	Type	Description	Default
path	str	Path to the file — absolute or relative to the working directory.	required
content	str	Text content to write.	required

Source code in src/ant_ai/acp/tools.py

@tool
async def acp_fs_write_file(path: str, content: str) -> None:
    """Write or overwrite a text file in the IDE's filesystem via ACP.

    Relative paths are resolved against the session working directory.

    Args:
        path: Path to the file — absolute or relative to the working directory.
        content: Text content to write.
    """
    client, session_id = _require_fs_write()
    resolved = _resolve_path(path)
    logger.debug(
        "acp_fs_write_file: path='{}' resolved='{}' len={}",
        path,
        resolved,
        len(content),
    )
    try:
        await client.write_text_file(
            path=resolved, content=content, session_id=session_id
        )
        logger.debug("acp_fs_write_file: done")
    except Exception as exc:
        logger.error("acp_fs_write_file failed: path='{}' error={}", resolved, exc)
        raise

acp_terminal_create async

acp_terminal_create(
    command: str,
    args: list[str] | None = None,
    cwd: str | None = None,
    env: list[dict[str, str]] | None = None,
    output_byte_limit: int | None = None,
) -> str

Create a terminal in the IDE and start executing a command.

Returns the terminal_id immediately without waiting for completion. Use acp_terminal_wait_for_exit() to block until done.

Parameters:

Name	Type	Description	Default
command	str	The command to execute.	required
args	list[str] | None	Optional list of arguments.	None
cwd	str | None	Optional absolute working directory.	None
env	list[dict[str, str]] | None	Optional environment variables as list of {"name": ..., "value": ...}.	None
output_byte_limit	int | None	Maximum output bytes before truncation.	None

Returns:

Type	Description
str	terminal_id string for subsequent terminal operations.

Source code in src/ant_ai/acp/tools.py

@tool
async def acp_terminal_create(
    command: str,
    args: list[str] | None = None,
    cwd: str | None = None,
    env: list[dict[str, str]] | None = None,
    output_byte_limit: int | None = None,
) -> str:
    """Create a terminal in the IDE and start executing a command.

    Returns the terminal_id immediately without waiting for completion.
    Use acp_terminal_wait_for_exit() to block until done.

    Args:
        command: The command to execute.
        args: Optional list of arguments.
        cwd: Optional absolute working directory.
        env: Optional environment variables as list of {"name": ..., "value": ...}.
        output_byte_limit: Maximum output bytes before truncation.

    Returns:
        terminal_id string for subsequent terminal operations.
    """
    from acp.schema import EnvVariable

    client, session_id = _require_terminal()
    env_vars = (
        [EnvVariable(name=e["name"], value=e["value"]) for e in env] if env else None
    )
    response = await client.create_terminal(
        command=command,
        session_id=session_id,
        args=args,
        cwd=cwd,
        env=env_vars,
        output_byte_limit=output_byte_limit,
    )
    return response.terminal_id

acp_terminal_output async

acp_terminal_output(terminal_id: str) -> str

Get the current output of a running or completed terminal.

Parameters:

Name	Type	Description	Default
terminal_id	str	Terminal identifier from acp_terminal_create().	required

Returns:

Type	Description
str	Captured output string (may be truncated if output_byte_limit was hit).

Source code in src/ant_ai/acp/tools.py

@tool
async def acp_terminal_output(terminal_id: str) -> str:
    """Get the current output of a running or completed terminal.

    Args:
        terminal_id: Terminal identifier from acp_terminal_create().

    Returns:
        Captured output string (may be truncated if output_byte_limit was hit).
    """
    client, session_id = _require_context()
    response = await client.terminal_output(
        session_id=session_id, terminal_id=terminal_id
    )
    return response.output

acp_terminal_wait_for_exit async

acp_terminal_wait_for_exit(
    terminal_id: str,
) -> dict[str, Any]

Block until a terminal command completes.

Parameters:

Name	Type	Description	Default
terminal_id	str	Terminal identifier from acp_terminal_create().	required

Returns:

Type	Description
dict[str, Any]	Dict with 'exit_code' (int | None) and 'signal' (str | None).

Source code in src/ant_ai/acp/tools.py

@tool
async def acp_terminal_wait_for_exit(terminal_id: str) -> dict[str, Any]:
    """Block until a terminal command completes.

    Args:
        terminal_id: Terminal identifier from acp_terminal_create().

    Returns:
        Dict with 'exit_code' (int | None) and 'signal' (str | None).
    """
    client, session_id = _require_context()
    response = await client.wait_for_terminal_exit(
        session_id=session_id, terminal_id=terminal_id
    )
    return {"exit_code": response.exit_code, "signal": response.signal}

acp_terminal_kill async

acp_terminal_kill(terminal_id: str) -> None

Kill a running terminal process (keeps the terminal for output retrieval).

Parameters:

Name	Type	Description	Default
terminal_id	str	Terminal identifier from acp_terminal_create().	required

Source code in src/ant_ai/acp/tools.py

@tool
async def acp_terminal_kill(terminal_id: str) -> None:
    """Kill a running terminal process (keeps the terminal for output retrieval).

    Args:
        terminal_id: Terminal identifier from acp_terminal_create().
    """
    client, session_id = _require_context()
    await client.kill_terminal(session_id=session_id, terminal_id=terminal_id)

acp_terminal_release async

acp_terminal_release(terminal_id: str) -> None

Kill any running process and release terminal resources.

After release, the terminal_id is no longer valid.

Parameters:

Name	Type	Description	Default
terminal_id	str	Terminal identifier from acp_terminal_create().	required

Source code in src/ant_ai/acp/tools.py

@tool
async def acp_terminal_release(terminal_id: str) -> None:
    """Kill any running process and release terminal resources.

    After release, the terminal_id is no longer valid.

    Args:
        terminal_id: Terminal identifier from acp_terminal_create().
    """
    client, session_id = _require_context()
    await client.release_terminal(session_id=session_id, terminal_id=terminal_id)

acp_terminal_run async

acp_terminal_run(
    command: str,
    args: list[str] | None = None,
    cwd: str | None = None,
    env: list[dict[str, str]] | None = None,
    timeout_sec: float | None = None,
) -> str

Run a command in the IDE terminal and return its output.

Convenience wrapper: creates a terminal, waits for the command to finish (killing it on timeout), retrieves output, and releases resources.

Parameters:

Name	Type	Description	Default
command	str	The command to execute.	required
args	list[str] | None	Optional list of arguments.	None
cwd	str | None	Optional absolute working directory.	None
env	list[dict[str, str]] | None	Optional environment variables as list of {"name": ..., "value": ...}.	None
timeout_sec	float | None	If set, kills the process after this many seconds.	None

Returns:

Type	Description
str	Combined stdout/stderr output from the command.

Source code in src/ant_ai/acp/tools.py

@tool
async def acp_terminal_run(
    command: str,
    args: list[str] | None = None,
    cwd: str | None = None,
    env: list[dict[str, str]] | None = None,
    timeout_sec: float | None = None,
) -> str:
    """Run a command in the IDE terminal and return its output.

    Convenience wrapper: creates a terminal, waits for the command to finish
    (killing it on timeout), retrieves output, and releases resources.

    Args:
        command: The command to execute.
        args: Optional list of arguments.
        cwd: Optional absolute working directory.
        env: Optional environment variables as list of {"name": ..., "value": ...}.
        timeout_sec: If set, kills the process after this many seconds.

    Returns:
        Combined stdout/stderr output from the command.
    """
    terminal_id = await acp_terminal_create(
        command=command, args=args, cwd=cwd, env=env
    )
    try:
        if timeout_sec is not None:
            try:
                await asyncio.wait_for(
                    acp_terminal_wait_for_exit(terminal_id), timeout=timeout_sec
                )
            except TimeoutError:
                await acp_terminal_kill(terminal_id)
        else:
            await acp_terminal_wait_for_exit(terminal_id)
        return await acp_terminal_output(terminal_id)
    finally:
        await acp_terminal_release(terminal_id)

acp_send_plan async

acp_send_plan(entries: list[dict[str, str]]) -> str

Send an agent plan update to the IDE.

Each entry must have 'content', 'priority' ("high"|"medium"|"low"), and 'status' ("pending"|"in_progress"|"completed").

Parameters:

Name	Type	Description	Default
entries	list[dict[str, str]]	List of plan entry dicts.	required

Returns:

Type	Description
str	Confirmation string "plan sent".

Source code in src/ant_ai/acp/tools.py

@tool
async def acp_send_plan(entries: list[dict[str, str]]) -> str:
    """Send an agent plan update to the IDE.

    Each entry must have 'content', 'priority' ("high"|"medium"|"low"),
    and 'status' ("pending"|"in_progress"|"completed").

    Args:
        entries: List of plan entry dicts.

    Returns:
        Confirmation string "plan sent".
    """
    client, session_id = _require_context()
    plan_entries = [
        PlanEntry(
            content=e["content"],
            priority=cast(Literal["high", "medium", "low"], e["priority"]),
            status=cast(Literal["pending", "in_progress", "completed"], e["status"]),
        )
        for e in entries
    ]
    await client.session_update(
        session_id=session_id,
        update=AgentPlanUpdate(session_update="plan", entries=plan_entries),
    )
    return "plan sent"