""" Anthropic Claude provider for FleetMind chat """ import os import logging from typing import Tuple, List, Dict from anthropic import Anthropic, APIError, APIConnectionError, AuthenticationError from chat.providers.base_provider import AIProvider from chat.tools import TOOLS_SCHEMA, execute_tool logger = logging.getLogger(__name__) class ClaudeProvider(AIProvider): """Anthropic Claude AI provider""" def __init__(self): self.api_key = os.getenv("ANTHROPIC_API_KEY", "") self.api_available = bool(self.api_key and not self.api_key.startswith("your_")) # Debug logging key_status = "not set" if not self.api_key else f"set ({len(self.api_key)} chars)" logger.info(f"ClaudeProvider init: ANTHROPIC_API_KEY {key_status}") if self.api_available: try: self.client = Anthropic(api_key=self.api_key) logger.info("ClaudeProvider: Initialized successfully") except Exception as e: logger.error(f"ClaudeProvider: Failed to initialize: {e}") self.api_available = False else: self.client = None logger.warning("ClaudeProvider: ANTHROPIC_API_KEY not configured") self.model = "claude-3-5-sonnet-20241022" self.system_prompt = self._get_system_prompt() def _get_system_prompt(self) -> str: """Get the system prompt for Claude""" return """You are an AI assistant for FleetMind, a delivery dispatch system. Your job is to help coordinators create and query delivery orders efficiently. **IMPORTANT: When a user wants to create an order, FIRST show them this order form:** 📋 **Order Information Form** Please provide the following details: **Required Fields:** • Customer Name: [Full name] • Delivery Address: [Street address, city, state, zip] • Contact: [Phone number OR email address] **Optional Fields:** • Delivery Deadline: [Date/time, or "ASAP" - default: 6 hours from now] • Priority: [standard/express/urgent - default: standard] • Special Instructions: [Any special notes] • Package Weight: [In kg - default: 5.0 kg] **Example:** "Customer: John Doe, Address: 123 Main St, San Francisco, CA 94103, Phone: 555-1234, Deliver by 5 PM today" --- **Your Workflow for ORDER CREATION:** 1. **If user says "create order" or similar:** Show the form above and ask them to provide the information 2. **If they provide all/most info:** Proceed immediately with geocoding and order creation 3. **If information is missing:** Show what's missing from the form and ask for those specific fields 4. **After collecting required fields:** - Use `geocode_address` tool to validate the address - Use `create_order` tool to save the order - Provide a clear confirmation with order ID **Your Workflow for ORDER QUERYING (INTERACTIVE):** When user asks to "fetch orders", "show orders", or "get orders": 1. First call `count_orders` (with any filters user mentioned) 2. Tell user: "I found X orders. How many would you like to see?" 3. Wait for user response 4. Call `fetch_orders` with the limit they specify 5. Display the results clearly When user asks "which orders are incomplete/not complete/pending": - Call `get_incomplete_orders` directly - Show results with priority and deadline When user asks about a specific order ID: - Call `get_order_details` with the order_id - Display all 26 fields clearly organized **Available Tools:** **Order Creation:** - geocode_address: Convert address to GPS coordinates - create_order: Create customer delivery order (REQUIRES geocoded address) **Order Querying:** - count_orders: Count orders with optional filters - fetch_orders: Fetch N orders with pagination and filters - get_order_details: Get complete info about specific order by ID - search_orders: Search by customer name/email/phone/order ID - get_incomplete_orders: Get all pending/assigned/in_transit orders **Driver Creation:** - create_driver: Add new driver to fleet **Driver Querying:** - count_drivers: Count drivers with optional filters - fetch_drivers: Fetch N drivers with pagination and filters - get_driver_details: Get complete info about specific driver by ID - search_drivers: Search by name/email/phone/plate/driver ID - get_available_drivers: Get all active/offline drivers **Available Order Filters:** - Status: pending, assigned, in_transit, delivered, failed, cancelled - Priority: standard, express, urgent - Payment: pending, paid, cod - Booleans: is_fragile, requires_signature, requires_cold_storage - Driver: assigned_driver_id **Available Driver Filters:** - Status: active, busy, offline, unavailable - Vehicle Type: van, truck, car, motorcycle, etc. **Your Workflow for DRIVER QUERYING (INTERACTIVE):** When user asks to "show drivers", "fetch drivers", or "get drivers": 1. First call `count_drivers` (with any filters user mentioned) 2. Tell user: "I found X drivers. How many would you like to see?" 3. Wait for user response 4. Call `fetch_drivers` with the limit they specify 5. Display the results clearly When user asks "which drivers are available/free": - Call `get_available_drivers` directly - Show results with status and vehicle info When user asks about a specific driver ID: - Call `get_driver_details` with the driver_id - Display all 15 fields clearly organized **Important Rules:** - ALWAYS geocode the address BEFORE creating an order - Be efficient - don't ask questions one at a time - Accept information in any format (natural language, bullet points, etc.) - Keep responses concise and professional - Show enthusiasm when orders/drivers are successfully created - For order/driver queries, be interactive and helpful with summaries Remember: Dispatch coordinators are busy - help them work efficiently!""" def is_available(self) -> bool: return self.api_available def get_status(self) -> str: if self.api_available: return f"✅ Connected - Model: {self.model}" return "⚠️ Not configured (add ANTHROPIC_API_KEY)" def get_provider_name(self) -> str: return "Claude (Anthropic)" def get_model_name(self) -> str: return self.model def process_message( self, user_message: str, conversation ) -> Tuple[str, List[Dict]]: """Process user message with Claude""" if not self.api_available: return self._handle_no_api(), [] # Add user message to history conversation.add_message("user", user_message) try: # Make API call to Claude response = self.client.messages.create( model=self.model, max_tokens=4096, system=self.system_prompt, tools=TOOLS_SCHEMA, messages=conversation.get_history() ) # Process response and handle tool calls return self._process_response(response, conversation) except AuthenticationError: error_msg = "⚠️ Invalid API key. Please check your ANTHROPIC_API_KEY in .env file." logger.error("Authentication error with Anthropic API") return error_msg, [] except APIConnectionError: error_msg = "⚠️ Cannot connect to Anthropic API. Please check your internet connection." logger.error("Connection error with Anthropic API") return error_msg, [] except APIError as e: error_msg = f"⚠️ API error: {str(e)}" logger.error(f"Anthropic API error: {e}") return error_msg, [] except Exception as e: error_msg = f"⚠️ Unexpected error: {str(e)}" logger.error(f"Claude provider error: {e}") return error_msg, [] def _process_response( self, response, conversation ) -> Tuple[str, List[Dict]]: """Process Claude's response and handle tool calls""" tool_calls_made = [] # Check if Claude wants to use tools if response.stop_reason == "tool_use": # Execute tools tool_results = [] for content_block in response.content: if content_block.type == "tool_use": tool_name = content_block.name tool_input = content_block.input logger.info(f"Claude executing tool: {tool_name}") # Execute the tool tool_result = execute_tool(tool_name, tool_input) # Track for transparency tool_calls_made.append({ "tool": tool_name, "input": tool_input, "result": tool_result }) conversation.add_tool_result(tool_name, tool_input, tool_result) # Prepare result for Claude tool_results.append({ "type": "tool_result", "tool_use_id": content_block.id, "content": str(tool_result) }) # Add assistant's tool use to history conversation.add_message("assistant", response.content) # Add tool results to history conversation.add_message("user", tool_results) # Continue conversation with tool results followup_response = self.client.messages.create( model=self.model, max_tokens=4096, system=self.system_prompt, tools=TOOLS_SCHEMA, messages=conversation.get_history() ) # Extract final text response final_text = self._extract_text_response(followup_response) conversation.add_message("assistant", final_text) return final_text, tool_calls_made else: # No tool use, just text response text_response = self._extract_text_response(response) conversation.add_message("assistant", text_response) return text_response, tool_calls_made def _extract_text_response(self, response) -> str: """Extract text content from Claude's response""" text_parts = [] for block in response.content: if hasattr(block, 'text'): text_parts.append(block.text) elif block.type == "text": text_parts.append(block.text if hasattr(block, 'text') else str(block)) return "\n".join(text_parts) if text_parts else "I apologize, but I couldn't generate a response." def _handle_no_api(self) -> str: """Return error message when API is not available""" return """⚠️ **Claude API requires Anthropic API key** To use Claude: 1. Get an API key from: https://console.anthropic.com/ - Sign up for free ($5 credit available) - Or use hackathon credits 2. Add to your `.env` file: ``` ANTHROPIC_API_KEY=sk-ant-your-key-here ``` 3. Restart the application **Alternative:** Switch to Gemini by setting `AI_PROVIDER=gemini` in .env """ def get_welcome_message(self) -> str: if not self.api_available: return self._handle_no_api() return """👋 Hello! I'm your AI dispatch assistant powered by **Claude Sonnet 3.5**. I can help you create and query delivery orders and drivers quickly and efficiently! --- 📋 **To Create an Order, Provide:** **Required:** • Customer Name • Delivery Address • Contact (Phone OR Email) **Optional:** • Delivery Deadline (default: 6 hours) • Priority: standard/express/urgent (default: standard) • Special Instructions • Package Weight in kg (default: 5.0) --- 🔍 **To Query Orders:** • "Fetch orders" or "Show orders" - I'll ask how many • "Which orders are incomplete?" - See all pending/in-progress orders • "Tell me about order ORD-XXX" - Get complete order details • "Show me 10 urgent orders" - Filter by priority • "Search for orders from John" - Find by customer name --- 🚚 **To Create a Driver:** **Required:** Driver Name **Optional:** Phone, Email, Vehicle Type, Plate, Capacity, Skills --- 👥 **To Query Drivers:** • "Show me drivers" or "Fetch drivers" - I'll ask how many • "Which drivers are available?" - See all active/offline drivers • "Tell me about driver DRV-XXX" - Get complete driver details • "Show 5 active drivers with vans" - Filter by status and vehicle • "Search for Tom" - Find by driver name --- **Quick Start Examples:** ✅ **Create Order:** "Create order for John Doe, 123 Main St San Francisco CA, phone 555-1234, deliver by 5 PM" ✅ **Query Orders:** "Fetch the orders" or "Show me incomplete orders" ✅ **Create Driver:** "Add driver Tom Wilson, phone 555-0101, drives a van, plate ABC-123" ✅ **Query Drivers:** "Show me drivers" or "Which drivers are available?" --- What would you like to do?"""