Anthropic

1
import asyncio
2
import os
3
import json
4
import base64
5
from io import BytesIO
6
from PIL import Image
7
from IPython.display import display
8
from typing import Any, cast
9
from datetime import datetime
10
11
from anthropic import Anthropic
12
from anthropic.types.beta import (
13
    BetaContentBlockParam,
14
    BetaTextBlockParam,
15
    BetaImageBlockParam,
16
    BetaToolResultBlockParam,
17
    BetaToolUseBlockParam,
18
    BetaMessageParam,
19
)
20
21
from scrapybara.anthropic import BashTool, ComputerTool, EditTool, ToolResult, ToolCollection
22
from scrapybara import Scrapybara
23
24
# Initialize Scrapybara
25
scrapybara_client = Scrapybara(api_key="your_scrapybara_api_key")
26
instance = scrapybara_client.start(instance_type="medium")
27
28
# Initialize Anthropic
29
anthropic_client = Anthropic(api_key="your_claude_api_key")
30
31
# System prompt from original Computer Use implementation
32
SYSTEM_PROMPT = """<SYSTEM_CAPABILITY>
33
* You are utilising an Ubuntu virtual machine using linux architecture with internet access.
34
* You can feel free to install Ubuntu applications with your bash tool. Use curl instead of wget.
35
* To open chromium, please just click on the web browser icon or use the (DISPLAY=:1 chromium &) command.  Note, chromium is what is installed on your system.
36
* Using bash tool you can start GUI applications, but you need to set export DISPLAY=:1 and use a subshell. For example "(DISPLAY=:1 xterm &)". GUI apps run with bash tool will appear within your desktop environment, but they may take some time to appear. Take a screenshot to confirm it did.
37
* When using your bash tool with commands that are expected to output very large quantities of text, redirect into a tmp file and use str_replace_editor or `grep -n -B <lines before> -A <lines after> <query> <filename>` to confirm output.
38
* When viewing a page it can be helpful to zoom out so that you can see everything on the page.  Either that, or make sure you scroll down to see everything before deciding something isn't available.
39
* When using your computer function calls, they take a while to run and send back to you.  Where possible/feasible, try to chain multiple of these calls all into one function calls request.
40
* The current date is {datetime.today().strftime('%A, %B %-d, %Y')}.
41
</SYSTEM_CAPABILITY>
42
43
<IMPORTANT>
44
* When using Chromium, if a startup wizard appears, IGNORE IT.  Do not even click "skip this step".  Instead, click on the address bar where it says "Search or enter address", and enter the appropriate search term or URL there.
45
* If the item you are looking at is a pdf, if after taking a single screenshot of the pdf it seems that you want to read the entire document instead of trying to continue to read the pdf from your screenshots + navigation, determine the URL, use curl to download the pdf, install and use pdftotext to convert it to a text file, and then read that text file directly with your StrReplaceEditTool.
46
</IMPORTANT>"""

1
def _make_api_tool_result(result: ToolResult, tool_use_id: str) -> BetaToolResultBlockParam:
2
    tool_result_content: list[BetaTextBlockParam | BetaImageBlockParam] | str = []  # Changed this line
3
    is_error = False
4
    if result.error:
5
        is_error = True
6
        tool_result_content = result.error
7
    else:
8
        if result.output:
9
            tool_result_content.append({
10
                "type": "text",
11
                "text": result.output,
12
            })
13
        if result.base64_image:
14
            tool_result_content.append({
15
                "type": "image",
16
                "source": {
17
                    "type": "base64",
18
                    "media_type": "image/png",
19
                    "data": result.base64_image,
20
                },
21
            })
22
    return {
23
        "type": "tool_result",
24
        "content": tool_result_content,
25
        "tool_use_id": tool_use_id,
26
        "is_error": is_error,
27
    }
28
29
def _response_to_params(response):
30
    res = []
31
    for block in response.content:
32
        if block.type == "text":
33
            res.append({"type": "text", "text": block.text})
34
        else:
35
            res.append(block.model_dump())
36
    return res
37
38
def _maybe_filter_to_n_most_recent_images(
39
    messages: list[BetaMessageParam],
40
    images_to_keep: int,
41
    min_removal_threshold: int,
42
):
43
    if images_to_keep is None:
44
        return messages
45
46
    tool_result_blocks = cast(
47
        list[BetaToolResultBlockParam],
48
        [
49
            item
50
            for message in messages
51
            for item in (
52
                message["content"] if isinstance(message["content"], list) else []
53
            )
54
            if isinstance(item, dict) and item.get("type") == "tool_result"
55
        ],
56
    )
57
58
    total_images = sum(
59
        1
60
        for tool_result in tool_result_blocks
61
        for content in tool_result.get("content", [])
62
        if isinstance(content, dict) and content.get("type") == "image"
63
    )
64
65
    images_to_remove = total_images - images_to_keep
66
    images_to_remove -= images_to_remove % min_removal_threshold
67
68
    for tool_result in tool_result_blocks:
69
        if isinstance(tool_result.get("content"), list):
70
            new_content = []
71
            for content in tool_result.get("content", []):
72
                if isinstance(content, dict) and content.get("type") == "image":
73
                    if images_to_remove > 0:
74
                        images_to_remove -= 1
75
                        continue
76
                new_content.append(content)
77
            tool_result["content"] = new_content

1
def display_base64_image(base64_string, max_size=(800, 800)):
2
    image_data = base64.b64decode(base64_string)
3
    image = Image.open(BytesIO(image_data))
4
5
    # Resize if larger than max_size while maintaining aspect ratio
6
    if image.size[0] > max_size[0] or image.size[1] > max_size[1]:
7
        image.thumbnail(max_size, Image.Resampling.LANCZOS)
8
9
    display(image)
10
11
async def sampling_loop(command: str):
12
    """
13
    Run the sampling loop for a single command until completion.
14
    """
15
    messages: list[BetaMessageParam] = []
16
    tool_collection = ToolCollection(
17
        ComputerTool(instance),
18
        BashTool(instance),
19
        EditTool(instance),
20
    )
21
22
    # Add initial command to messages
23
    messages.append({
24
        "role": "user",
25
        "content": [{"type": "text", "text": command}],
26
    })
27
28
    while True:
29
        _maybe_filter_to_n_most_recent_images(messages, 2, 2)
30
31
        # Get Claude's response
32
        response = anthropic_client.beta.messages.create(
33
            model="claude-3-5-sonnet-20241022",
34
            max_tokens=4096,
35
            messages=messages,
36
            system=[{"type": "text", "text": SYSTEM_PROMPT}],
37
            tools=tool_collection.to_params(),
38
            betas=["computer-use-2024-10-22"]
39
        )
40
41
        # Convert response to params
42
        response_params = _response_to_params(response)
43
44
        # Process response content and handle tools before adding to messages
45
        tool_result_content: list[BetaToolResultBlockParam] = []
46
47
        for content_block in response_params:
48
            if content_block["type"] == "text":
49
                print(f"\nAssistant: {content_block['text']}")
50
51
            elif content_block["type"] == "tool_use":
52
                print(f"\nTool Use: {content_block['name']}")
53
                print(f"Input: {content_block['input']}")
54
55
                # Execute the tool
56
                result = await tool_collection.run(
57
                    path=content_block["name"],
58
                    tool_input=cast(dict[str, Any], content_block["input"])
59
                )
60
61
                print(f"Result: {result}")
62
                if content_block['name'] == 'bash' and not result:
63
                    result = await tool_collection.run(
64
                        path="computer",
65
                        tool_input={"action": "screenshot"}
66
                    )
67
                    print("Updated result: ", result)
68
69
                if result:
70
                    print("Converting tool result: ", result)
71
                    tool_result = _make_api_tool_result(result, content_block["id"])
72
                    print(f"Tool Result: {tool_result}")
73
74
                    if result.output:
75
                        print(f"\nTool Output: {result.output}")
76
                    if result.error:
77
                        print(f"\nTool Error: {result.error}")
78
                    if result.base64_image:
79
                        print("\nTool generated an image (base64 data available)")
80
                        display_base64_image(result.base64_image)
81
82
                    tool_result_content.append(tool_result)
83
84
                    print("\n---")
85
86
        # Add assistant's response to messages
87
        messages.append({
88
            "role": "assistant",
89
            "content": response_params,
90
        })
91
92
        # If tools were used, add their results to messages
93
        if tool_result_content:
94
            messages.append({
95
                "role": "user",
96
                "content": tool_result_content
97
            })
98
        else:
99
            # No tools used, task is complete
100
            break

1
command = "Google Scrapybara"
2
3
# Run the sampling loop for this command
4
await sampling_loop(command)