How to Build a Production-Ready Claude Chatbot in Python

pip install anthropic
export ANTHROPIC_API_KEY="your-api-key-here"

import os
from typing import Generator, List, Dict
from anthropic import Anthropic

class ClaudeChatEngine:
    def __init__(self, system_prompt: str = "You are a helpful, concise assistant."):
        # Initialize the client. It automatically looks for ANTHROPIC_API_KEY in env
        self.client = Anthropic()
        self.model = "claude-3-5-sonnet-20241022"
        self.system_prompt = system_prompt
        self.history: List[Dict[str, str]] = []

    def send_message(self, user_message: str) -> Generator[str, None, None]:
        """
        Sends a message to Claude, updates history, and yields tokens as they arrive.
        """
        # Append user message to state history
        self.history.append({"role": "user", "content": user_message})

        assistant_response = ""

        # Initiate a streaming request
        with self.client.messages.stream(
            model=self.model,
            max_tokens=1024,
            system=self.system_prompt,
            messages=self.history
        ) as stream:
            for text in stream.text_stream:
                assistant_response += text
                yield text

        # Append Claude's completed response to history to maintain context
        self.history.append({"role": "assistant", "content": assistant_response})

# Usage Example
if __name__ == "__main__":
    bot = ClaudeChatEngine()
    print("Chatbot initialized. Type 'exit' to quit.\n")

    while True:
        user_input = input("You: ")
        if user_input.lower() == "exit":
            break

        print("Claude: ", end="", flush=True)
        for token in bot.send_message(user_input):
            print(token, end="", flush=True)
        print("\n")

import time
import logging
from anthropic import (
    Anthropic, 
    APIConnectionError, 
    RateLimitError, 
    APIStatusError
)

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("ClaudeChat")

class ResilientClaudeChatEngine(ClaudeChatEngine):
    def send_message_safe(self, user_message: str, max_retries: int = 3) -> Generator[str, None, None]:
        """
        Sends a message with exponential backoff for rate limits and robust error handling.
        """
        self.history.append({"role": "user", "content": user_message})
        assistant_response = ""
        delay = 1.0  # Initial delay for backoff

        for attempt in range(max_retries):
            try:
                with self.client.messages.stream(
                    model=self.model,
                    max_tokens=1024,
                    system=self.system_prompt,
                    messages=self.history
                ) as stream:
                    for text in stream.text_stream:
                        assistant_response += text
                        yield text

                # If we successfully finished streaming, break out of retry loop
                break

            except RateLimitError as e:
                logger.warning(f"Rate limited (429). Retrying in {delay}s... Error: {e}")
                if attempt == max_retries - 1:
                    yield "\n[Error: The system is currently busy. Please try again in a moment.]"
                    return
                time.sleep(delay)
                delay *= 2  # Exponential backoff

            except APIConnectionError as e:
                logger.error(f"Failed to connect to [Anthropic API](https://console.anthropic.com): {e}")
                yield "\n[Error: Connection issue. Please check your internet connection.]"
                return

            except APIStatusError as e:
                logger.error(f"Anthropic API returned status code {e.status_code}: {e.message}")
                yield f"\n[Error: An API error occurred (Status {e.status_code}).]"
                return

            except Exception as e:
                logger.error(f"Unexpected error: {str(e)}")
                yield "\n[Error: An unexpected system error occurred.]"
                return

        # Save the history only if we got a successful response
        if assistant_response:
            self.history.append({"role": "assistant", "content": assistant_response})