AffordableAI/ComputerUseDocker
3
1import os2import base643import gradio as gr4from typing import List, Tuple, Optional, Dict, Any5from enum import Enum6from datetime import datetime7import httpx8from dataclasses import dataclass9from anthropic import Anthropic10import json11 12# Get API key from environment variable13ANTHROPIC_API_KEY = os.getenv('ANTHROPIC_API_KEY')14if not ANTHROPIC_API_KEY:15 raise ValueError("ANTHROPIC_API_KEY environment variable must be set")16 17# Initialize Anthropic client18anthropic = Anthropic(api_key=ANTHROPIC_API_KEY)19 20@dataclass21class ToolResult:22 output: Optional[str] = None23 error: Optional[str] = None24 base64_image: Optional[str] = None25 26 def to_dict(self) -> Dict[str, Any]:27 return {28 "type": "tool_result",29 "output": self.output,30 "error": self.error,31 "base64_image": self.base64_image32 }33 34# Define the computer use tools35COMPUTER_USE_TOOLS = [36 {37 "type": "computer_20241022",38 "name": "computer",39 "display_width_px": 1024,40 "display_height_px": 768,41 "display_number": 1,42 },43 {44 "type": "text_editor_20241022",45 "name": "str_replace_editor"46 },47 {48 "type": "bash_20241022",49 "name": "bash"50 }51]52 53class ComputerUseDemo:54 def __init__(self):55 self.messages: List[Dict[str, Any]] = []56 self.model = "claude-3-5-sonnet-20241022"57 self.max_tokens = 409658 59 def validate_auth(self) -> Optional[str]:60 if not ANTHROPIC_API_KEY:61 return "ANTHROPIC_API_KEY environment variable is not set."62 return None63 64 def format_messages(self, chat_history: List[Tuple[str, str]]) -> List[Dict[str, Any]]:65 """Convert chat history to Anthropic message format"""66 formatted_messages = []67 for user_msg, assistant_msg in chat_history:68 formatted_messages.append({"role": "user", "content": user_msg})69 if assistant_msg:70 formatted_messages.append({"role": "assistant", "content": assistant_msg})71 return formatted_messages72 73 def execute_tool(self, tool_name: str, tool_input: str) -> ToolResult:74 """Execute the requested tool and return results"""75 try:76 if tool_name == "computer":77 return ToolResult(78 output=f"Executed computer command: {tool_input}",79 base64_image=None80 )81 elif tool_name == "str_replace_editor":82 return ToolResult(83 output=f"Created/modified file with content: {tool_input}"84 )85 elif tool_name == "bash":86 return ToolResult(87 output=f"Executed bash command: {tool_input}"88 )89 else:90 return ToolResult(91 error=f"Unknown tool: {tool_name}"92 )93 except Exception as e:94 return ToolResult(error=str(e))95 96 def process_message(97 self, 98 message: str, 99 chat_history: List[Tuple[str, str]],100 system_prompt: str101 ) -> Tuple[List[Tuple[str, str]], Optional[str]]:102 """Process a message with proper error handling"""103 if error := self.validate_auth():104 return chat_history, error105 106 try:107 # Format messages for Claude108 messages = self.format_messages(chat_history)109 messages.append({"role": "user", "content": message})110 111 # Create system prompt for computer use112 base_system_prompt = """You have access to a computing environment through specialized tools.113 Use the computer tool to interact with the GUI, the text editor for file operations, 114 and bash for command-line tasks. After each action, explicitly describe what you did and what happened.115 Always show the results of your actions."""116 117 full_system_prompt = f"{base_system_prompt}\n{system_prompt}" if system_prompt else base_system_prompt118 119 conversation_log = []120 conversation_log.append((message, "")) # Add user message121 122 while True:123 # Call Claude API with computer use enabled124 response = anthropic.beta.messages.create(125 model=self.model,126 max_tokens=self.max_tokens,127 messages=messages,128 system=full_system_prompt,129 tools=COMPUTER_USE_TOOLS,130 betas=["computer-use-2024-10-22"]131 )132 133 # Get the assistant's response text134 assistant_message = response.content[0].text if response.content else ""135 136 # Check for tool calls in the response137 tool_calls = []138 for content in response.content:139 if hasattr(content, 'tool_calls') and content.tool_calls:140 tool_calls.extend(content.tool_calls)141 142 # Add assistant's initial response to conversation143 if assistant_message:144 conversation_log.append(("", assistant_message))145 146 # Handle tool calls if present147 if tool_calls:148 tool_call = tool_calls[0] # Handle first tool call149 tool_result = self.execute_tool(tool_call.name, tool_call.parameters)150 151 # Add tool execution to conversation152 if tool_result.error:153 conversation_log.append(("", f"Error: {tool_result.error}"))154 else:155 conversation_log.append(("", f"Tool execution: {tool_result.output}"))156 157 # Add tool interactions to message history158 messages.append({159 "role": "assistant",160 "content": [{"type": "tool_use", "name": tool_call.name, "input": tool_call.parameters}]161 })162 messages.append({163 "role": "user",164 "content": [tool_result.to_dict()]165 })166 else:167 break168 169 return conversation_log, None170 171 except Exception as e:172 return chat_history, f"Error processing message: {str(e)}"173 174def create_demo() -> gr.Interface:175 demo = ComputerUseDemo()176 177 with gr.Blocks(title="Claude Computer Use Demo") as interface:178 gr.Markdown("# Claude Computer Use Demo")179 gr.Markdown("""## Security Notice180 ⚠️ This demo runs in a sandboxed environment. Never provide access to sensitive 181 information or credentials, as Claude's behavior could be influenced by malicious content.""")182 183 with gr.Row():184 with gr.Column(scale=4):185 chatbot = gr.Chatbot(186 label="Chat History",187 height=600,188 container=True189 )190 with gr.Row():191 msg = gr.Textbox(192 label="Message",193 placeholder="Type your instructions for Claude...",194 lines=3,195 scale=8196 )197 with gr.Column(scale=1):198 submit = gr.Button("Send", variant="primary")199 clear = gr.Button("Reset")200 201 with gr.Column(scale=1):202 system_prompt = gr.Textbox(203 label="System Instructions",204 lines=3,205 placeholder="Add custom instructions for Claude..."206 )207 with gr.Accordion("Advanced Settings", open=False):208 model = gr.Textbox(209 label="Model",210 value=demo.model,211 interactive=False212 )213 max_tokens = gr.Slider(214 label="Max Tokens",215 minimum=1024,216 maximum=8192,217 value=demo.max_tokens,218 step=1024219 )220 221 def on_message(222 message: str,223 chat_history: List[Tuple[str, str]],224 system_prompt: str225 ) -> Tuple[List[Tuple[str, str]], str]:226 if not message.strip():227 return chat_history, ""228 new_history, error = demo.process_message(message, chat_history or [], system_prompt)229 if error:230 gr.Warning(error)231 return new_history, ""232 233 # Set up event handlers234 msg.submit(235 on_message,236 inputs=[msg, chatbot, system_prompt],237 outputs=[chatbot, msg]238 )239 240 submit.click(241 on_message,242 inputs=[msg, chatbot, system_prompt],243 outputs=[chatbot, msg]244 )245 246 clear.click(247 lambda: ([], ""),248 outputs=[chatbot, msg]249 )250 251 return interface252 253# Create and launch the demo254demo = create_demo()255 256if __name__ == "__main__":257 demo.launch(server_name="0.0.0.0", server_port=7860)