On the last stage, for the second test (#OZ7 (Implement the write tool)) on the third API call I keep getting:
Error code: 500 - {'error': {'message': 'Internal Server Error', 'code': 500}}!)
I would guess that this is rate-limit issue. No amount of cycling through the providers or sleeping for 30s helped, though (see my code below. I realize it’s a little more complicated than it needs to be because I am constraining Bash - see the Course Feedback topic. But removing the os path lookup didn’t make a difference). I can’t sleep for longer as the Test times out in 45s.
import argparse
import os
import sys
import json
import re
import subprocess
import time
import random
from openai import OpenAI
API_KEY = os.getenv("OPENROUTER_API_KEY")
BASE_URL = os.getenv("OPENROUTER_BASE_URL", default="https://openrouter.ai/api/v1")
def safe_bash_list(current_dir):
""" Handle list command - only list current directory """
try:
# Use subprocess.run with capture_output to get ls output
# This runs 'ls' in the current directory only
process = subprocess.run(
['ls'],
capture_output=True,
text=True,
cwd=current_dir # Ensure it runs in the correct directory
)
result = process.stdout if process.returncode == 0 else process.stderr
except Exception as e:
result = f"Error executing ls: {str(e)}"
return result
def safe_bash_delete(tool_args):
"""
Allows deleting a file via bash, while constraining it to
only running a non-recursive rm, and only within
the test's directory.
Parameters
----------
tool_args
Returns
-------
result: string
"""
# Get the current working directory to restrict file operations
# This prevents the model from deleting files outside the project directory
current_dir = os.getcwd()
command = tool_args.get("command")
# Check if command is a simple delete (rm) command
# Pattern: rm <filename> (no flags like -rf, no pipes, no &&)
command = command.strip()
# Only allow simple "rm <filename>" commands
# This prevents: rm -rf, rm *, sudo rm, etc.
is_rm_command = re.match(r'^rm\s+[^\s]+$', command)
is_ls_command = re.match(r'^ls(\s+)?$', command)
if is_rm_command:
# Extract the file path from the command
parts = command.split(None, 1) # Split on whitespace
if len(parts) == 2:
file_to_delete = parts[1]
# Security: Ensure the file path doesn't escape current directory
if not file_to_delete.startswith("/") and not "../" in file_to_delete:
full_path = os.path.join(current_dir, file_to_delete)
full_path = os.path.abspath(full_path)
# Verify the file is within current directory
if full_path.startswith(current_dir):
try:
os.remove(full_path)
result = "File deleted successfully"
except Exception as e:
result = f"Error deleting file: {str(e)}"
else:
result = f"Error: File path '{file_to_delete}' escapes working directory"
else:
result = f"Error: Invalid file path '{file_to_delete}' - absolute paths and ../ not allowed"
else:
result = "Error: Invalid rm command format"
elif is_ls_command:
result = safe_bash_list(current_dir)
else:
# Reject any other commands for security
# This blocks: ls -la, ls -l, ls /path, cd, rm -rf, sudo, etc.
result = "Error: Only simple 'rm <filename>' or 'ls' commands are allowed. Other commands are blocked for security."
return result
def make_api_call(client, messages, tools, providers, provider, max_retries=5):
"""Make API call with retry logic for rate limiting"""
for attempt in range(max_retries):
try:
return client.chat.completions.create(
model="anthropic/claude-haiku-4.5",
messages=messages,
tools=tools,
extra_body={
"provider": {
"order": [providers[provider]],
"allow_fallbacks": False
}
}
)
except Exception as e:
print(e)
if attempt < max_retries - 1 and "500" in str(e):
seconds = 15
print(f"API rate limit hit, retrying in {seconds}s...", file=sys.stderr)
time.sleep(seconds) # Increasing backoff
if provider == 2:
provider = 0
else:
provider += 1
else:
raise e
def main():
p = argparse.ArgumentParser()
p.add_argument("-p", required=True)
args = p.parse_args()
if not API_KEY:
raise RuntimeError("OPENROUTER_API_KEY is not set")
client = OpenAI(api_key=API_KEY, base_url=BASE_URL)
messages=[{"role": "user", "content": args.p}]
providers = ["Google"]
others = ["Anthropic", "Amazon Bedrock"]
next = random.choice(others)
providers.append(next)
providers.remove(next)
providers.extend(others)
provider = 0
last_provider = None
while True:
tools=[
{
"type": "function",
"function": {
"name": "Read",
"description": "Read and return the contents of a file",
"parameters": {
"type": "object",
"required": ["file_path"],
"properties": {
"file_path": {
"type": "string",
"description": "The path to the file to read"
}
}
}
}
},
{
"type": "function",
"function": {
"name": "Write",
"description": "Write content to a file",
"parameters": {
"type": "object",
"required": ["file_path", "content"],
"properties": {
"file_path": {
"type": "string",
"description": "The path of the file to write to"
},
"content": {
"type": "string",
"description": "The content to write to the file"
}
}
}
}
},
{
"type": "function",
"function": {
"name": "Bash",
"description": "Execute a shell command",
"parameters": {
"type": "object",
"required": ["command"],
"properties": {
"command": {
"type": "string",
"description": "The command to execute"
}
}
}
}
}
]
chat = make_api_call(client, messages, tools, providers, provider)
if not chat.choices or len(chat.choices) == 0:
raise RuntimeError("no choices in response")
print("chat returned as:", chat, file=sys.stderr)
message = chat.choices[0].message
if message:
messages.append(message)
content = message.content
tool_calls = message.tool_calls
if tool_calls and len(tool_calls) > 0:
for call in tool_calls:
tool = call.function
args = json.loads(tool.arguments)
if tool:
name = tool.name
result = None
if name == 'Read':
with open(args.get("file_path"), 'r') as file:
result = file.read()
elif name == 'Write':
with open(args.get("file_path"), 'w') as file:
file.write(args.get("content"))
elif name == 'Bash':
result = safe_bash_delete(args)
messages.append({
"role": "tool",
"tool_call_id": call.id,
"content" : result
})
last_provider = chat.provider
if provider == 2:
provider = 0
else:
provider += 1
else:
# TODO: Uncomment the following line to pass the first stage
print(content)
break
if __name__ == "__main__":
main()
seconds = 30
print("sleeping for:", seconds)
time.sleep(seconds)
If it is a rate-limit issue, can we have the option of falling back to some other free model when this happens?