prettify errors

Author: x5a

computer-use-demo/computer_use_demo/loop.py

@@ -8,15 +8,22 @@
 from enum import StrEnum
 from typing import Any, cast
 
-from anthropic import Anthropic, AnthropicBedrock, AnthropicVertex, APIResponse
+import httpx
+from anthropic import (
+    Anthropic,
+    AnthropicBedrock,
+    AnthropicVertex,
+    APIError,
+    APIResponseValidationError,
+    APIStatusError,
+)
 from anthropic.types import (
     ToolResultBlockParam,
 )
 from anthropic.types.beta import (
     BetaContentBlock,
     BetaContentBlockParam,
     BetaImageBlockParam,
-    BetaMessage,
     BetaMessageParam,
     BetaTextBlockParam,
     BetaToolResultBlockParam,
@@ -70,7 +77,9 @@ async def sampling_loop(
     messages: list[BetaMessageParam],
     output_callback: Callable[[BetaContentBlock], None],
     tool_output_callback: Callable[[ToolResult, str], None],
-    api_response_callback: Callable[[APIResponse[BetaMessage]], None],
+    api_response_callback: Callable[
+        [httpx.Request, httpx.Response | object | None, Exception | None], None
+    ],
     api_key: str,
     only_n_most_recent_images: int | None = None,
     max_tokens: int = 4096,
@@ -102,16 +111,25 @@ async def sampling_loop(
         # we use raw_response to provide debug information to streamlit. Your
         # implementation may be able call the SDK directly with:
         # `response = client.messages.create(...)` instead.
-        raw_response = client.beta.messages.with_raw_response.create(
-            max_tokens=max_tokens,
-            messages=messages,
-            model=model,
-            system=system,
-            tools=tool_collection.to_params(),
-            betas=["computer-use-2024-10-22"],
-        )
+        try:
+            raw_response = client.beta.messages.with_raw_response.create(
+                max_tokens=max_tokens,
+                messages=messages,
+                model=model,
+                system=system,
+                tools=tool_collection.to_params(),
+                betas=["computer-use-2024-10-22"],
+            )
+        except (APIStatusError, APIResponseValidationError) as e:
+            api_response_callback(e.request, e.response, e)
+            return messages
+        except APIError as e:
+            api_response_callback(e.request, e.body, e)
+            return messages
 
-        api_response_callback(cast(APIResponse[BetaMessage], raw_response))
+        api_response_callback(
+            raw_response.http_response.request, raw_response.http_response, None
+        )
 
         response = raw_response.parse()
 

computer-use-demo/computer_use_demo/streamlit.py

@@ -6,18 +6,19 @@
 import base64
 import os
 import subprocess
-from datetime import datetime
+from datetime import datetime, timedelta
 from enum import StrEnum
 from functools import partial
 from pathlib import PosixPath
 from typing import cast
 
+import httpx
 import streamlit as st
-from anthropic import APIResponse
+from anthropic import RateLimitError
 from anthropic.types import (
     TextBlock,
 )
-from anthropic.types.beta import BetaMessage, BetaTextBlock, BetaToolUseBlock
+from anthropic.types.beta import BetaTextBlock, BetaToolUseBlock
 from anthropic.types.tool_use_block import ToolUseBlock
 from streamlit.delta_generator import DeltaGenerator
 
@@ -186,8 +187,8 @@ def _reset_api_provider():
                         )
 
         # render past http exchanges
-        for identity, response in st.session_state.responses.items():
-            _render_api_response(response, identity, http_logs)
+        for identity, (request, response) in st.session_state.responses.items():
+            _render_api_response(request, response, identity, http_logs)
 
         # render past chats
         if new_message:
@@ -278,16 +279,20 @@ def save_to_storage(filename: str, data: str) -> None:
 
 
 def _api_response_callback(
-    response: APIResponse[BetaMessage],
+    request: httpx.Request,
+    response: httpx.Response | object | None,
+    error: Exception | None,
     tab: DeltaGenerator,
-    response_state: dict[str, APIResponse[BetaMessage]],
+    response_state: dict[str, tuple[httpx.Request, httpx.Response | object | None]],
 ):
     """
     Handle an API response by storing it to state and rendering it.
     """
     response_id = datetime.now().isoformat()
-    response_state[response_id] = response
-    _render_api_response(response, response_id, tab)
+    response_state[response_id] = (request, response)
+    if error:
+        _render_error(error)
+    _render_api_response(request, response, response_id, tab)
 
 
 def _tool_output_callback(
@@ -299,20 +304,38 @@ def _tool_output_callback(
 
 
 def _render_api_response(
-    response: APIResponse[BetaMessage], response_id: str, tab: DeltaGenerator
+    request: httpx.Request,
+    response: httpx.Response | object | None,
+    response_id: str,
+    tab: DeltaGenerator,
 ):
     """Render an API response to a streamlit tab"""
     with tab:
         with st.expander(f"Request/Response ({response_id})"):
             newline = "\n\n"
             st.markdown(
-                f"`{response.http_request.method} {response.http_request.url}`{newline}{newline.join(f'`{k}: {v}`' for k, v in response.http_request.headers.items())}"
-            )
-            st.json(response.http_request.read().decode())
-            st.markdown(
-                f"`{response.http_response.status_code}`{newline}{newline.join(f'`{k}: {v}`' for k, v in response.headers.items())}"
+                f"`{request.method} {request.url}`{newline}{newline.join(f'`{k}: {v}`' for k, v in request.headers.items())}"
             )
-            st.json(response.http_response.text)
+            st.json(request.read().decode())
+            st.markdown("---")
+            if isinstance(response, httpx.Response):
+                st.markdown(
+                    f"`{response.status_code}`{newline}{newline.join(f'`{k}: {v}`' for k, v in response.headers.items())}"
+                )
+                st.json(response.text)
+            else:
+                st.write(response)
+
+
+def _render_error(error: Exception):
+    if isinstance(error, RateLimitError):
+        body = "You have been rate limited."
+        if retry_after := error.response.headers.get("retry-after"):
+            body += f" **Retry after {str(timedelta(seconds=int(retry_after)))} (HH:MM:SS).**"
+        body += f"\n\n{error.message}"
+    else:
+        body = str(error)
+    st.error(f"**{error.__class__.__name__}**\n\n{body}", icon=":material/error:")
 
 
 def _render_message(