wanderlust.ai / src /wanderlust_ai /api /anthropic_client.py
BlakeL's picture
Upload 115 files
3f9f85b verified
"""
Anthropic Claude API client with secure configuration management.
This module demonstrates how to use environment variables securely
with the configuration system.
"""
import asyncio
from typing import Optional
import httpx
from anthropic import Anthropic
from ..utils.config import get_settings
class AnthropicClient:
"""Secure Anthropic Claude API client."""
def __init__(self, api_key: Optional[str] = None):
"""
Initialize the Anthropic client.
Args:
api_key: Optional API key. If not provided, will use settings.
"""
self.settings = get_settings()
# Use provided API key or get from settings
self.api_key = api_key or self.settings.anthropic_api_key
if not self.api_key:
raise ValueError(
"Anthropic API key is required. "
"Set ANTHROPIC_API_KEY in your .env file."
)
# Initialize the client
self.client = Anthropic(api_key=self.api_key)
async def generate_response(
self,
prompt: str,
max_tokens: int = 1000,
model: str = "claude-3-haiku-20240307"
) -> str:
"""
Generate a response using Claude.
Args:
prompt: The input prompt
max_tokens: Maximum tokens to generate
model: Claude model to use
Returns:
Generated response text
"""
try:
# Use httpx for async requests with timeout
async with httpx.AsyncClient(timeout=self.settings.request_timeout) as client:
response = await asyncio.to_thread(
self.client.messages.create,
model=model,
max_tokens=max_tokens,
messages=[{"role": "user", "content": prompt}]
)
return response.content[0].text
except Exception as e:
if self.settings.debug:
print(f"Error generating response: {e}")
raise
def get_model_info(self) -> dict:
"""Get information about available models."""
return {
"api_key_configured": bool(self.api_key),
"api_key_length": len(self.api_key) if self.api_key else 0,
"timeout": self.settings.request_timeout,
"debug_mode": self.settings.debug
}
# Example usage
async def main():
"""Example usage of the Anthropic client."""
try:
# Initialize client (will automatically load from .env)
client = AnthropicClient()
# Get model info
info = client.get_model_info()
print("πŸ€– Anthropic Client Info:")
print(f" API Key configured: {info['api_key_configured']}")
print(f" API Key length: {info['api_key_length']} characters")
print(f" Timeout: {info['timeout']} seconds")
print(f" Debug mode: {info['debug_mode']}")
# Generate a test response
response = await client.generate_response(
"Hello! Can you help me plan a trip to Japan?"
)
print(f"\nπŸ“ Claude Response:\n{response}")
except ValueError as e:
print(f"❌ Configuration Error: {e}")
print("Please check your .env file and ensure ANTHROPIC_API_KEY is set.")
except Exception as e:
print(f"❌ Error: {e}")
if __name__ == "__main__":
asyncio.run(main())