Bash 工具

import subprocess
import uuid


class BashSession:
    """A bash process that stays alive between commands so state persists."""

    def __init__(self):
        self.process = subprocess.Popen(
            ["/bin/bash"],
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,  # interleave errors with output, in order
            start_new_session=True,  # own process group: a timeout can kill every child
            text=True,
        )

    def execute_command(self, command):
        """Run a command in the session and return its output."""
        sentinel = f"__CLAUDE_BASH_DONE_{uuid.uuid4().hex}__"  # unique per call
        self.process.stdin.write(f"{command}\necho {sentinel}\n")
        self.process.stdin.flush()

        output = []
        for line in self.process.stdout:
            if sentinel in line:  # this command's output is complete
                break
            output.append(line)
        return "".join(output)

    def restart(self):
        self.process.kill()
        self.process.wait()
        self.__init__()


bash_session = BashSession()
print(bash_session.execute_command("cd /tmp && pwd"))
print(bash_session.execute_command("pwd"))  # still /tmp: the session kept its state

tool_results = []
for content in response.content:
    if content.type == "tool_use" and content.name == "bash":
        if content.input.get("restart"):
            bash_session.restart()
            result = "Bash session restarted"
        else:
            command = content.input.get("command")
            result = bash_session.execute_command(command)

        # 每个 tool_use 块对应一个 tool_result，全部在下一条用户消息中返回
        tool_results.append(
            {"type": "tool_result", "tool_use_id": content.id, "content": result}
        )

client = anthropic.Anthropic()

response = client.messages.create(
    model="claude-opus-5",
    max_tokens=1024,
    tools=[{"type": "bash_20250124", "name": "bash"}],
    messages=[
        {"role": "user", "content": "List all Python files in the current directory."},
        {
            "role": "assistant",
            "content": [
                {
                    "type": "tool_use",
                    "id": "toolu_01A09q90qw90lq917835lq9",
                    "name": "bash",
                    "input": {"command": "ls *.py"},
                }
            ],
        },
        {
            "role": "user",
            "content": [
                {
                    "type": "tool_result",
                    "tool_use_id": "toolu_01A09q90qw90lq917835lq9",
                    "content": "analysis.py\nprocess_data.py\n",
                }
            ],
        },
    ],
)

print(response.content)

import shlex

ALLOWED_COMMANDS = {"ls", "cat", "echo", "pwd", "grep", "find", "wc", "head", "tail"}
SHELL_OPERATORS = {"&&", "||", "|", ";", "&", ">", "<", ">>"}


def validate_command(command):
    # 仅允许显式允许列表中的命令
    try:
        tokens = shlex.split(command)
    except ValueError:
        return False, "Could not parse command"

    if not tokens:
        return False, "Empty command"

    executable = tokens[0]
    if executable not in ALLOWED_COMMANDS:
        return False, f"Command '{executable}' is not in the allowlist"

    # 拒绝以独立单词形式书写的 shell 运算符
    for token in tokens[1:]:
        if token in SHELL_OPERATORS or token.startswith(("$", "`")):
            return False, f"Shell operator '{token}' is not allowed"

    return True, None