Creating Custom Tools

def search_products(category: str, min_price: float, in_stock: bool):
    """Search products with filters."""
    pass
 
def batch_process_orders(order_ids: list):
    """Process multiple orders at once."""
    pass
 
def create_customer_profile(data: dict):
    """Create customer with complex nested data."""
    pass

import requests
import frappe
 
def get_weather_forecast(city: str):
    """
    Get weather forecast for a city from OpenWeather API.
    
    Args:
        city (str): City name
        
    Returns:
        dict: Weather forecast data
    """
    api_key = frappe.conf.get("openweather_api_key")
    
    response = requests.get(
        f"https://api.openweathermap.org/data/2.5/weather",
        params={"q": city, "appid": api_key, "units": "metric"}
    )
    
    if response.status_code == 200:
        data = response.json()
        return {
            "city": city,
            "temperature": data["main"]["temp"],
            "condition": data["weather"][0]["description"],
            "humidity": data["main"]["humidity"]
        }
    else:
        frappe.throw(f"Weather API error: {response.status_code}")

import frappe
from datetime import datetime, timedelta
 
def calculate_reorder_quantity(item_code: str, days_ahead: int = 30):
    """
    Calculate optimal reorder quantity based on historical consumption.
    
    Args:
        item_code (str): Item code
        days_ahead (int): Days to plan ahead (default: 30)
        
    Returns:
        dict: Reorder recommendations
    """
    # Get current stock
    stock_qty = frappe.db.get_value("Bin", 
        {"item_code": item_code}, 
        "actual_qty"
    ) or 0
    
    # Get item details
    item = frappe.get_doc("Item", item_code)
    
    # Calculate average daily consumption (last 90 days)
    ninety_days_ago = datetime.now() - timedelta(days=90)
    
    consumed = frappe.db.sql("""
        SELECT SUM(actual_qty)
        FROM `tabStock Ledger Entry`
        WHERE item_code = %s
        AND posting_date >= %s
        AND actual_qty < 0
    """, (item_code, ninety_days_ago))[0][0] or 0
    
    daily_consumption = abs(consumed) / 90
    
    # Calculate needed quantity
    needed_qty = daily_consumption * days_ahead
    reorder_qty = max(0, needed_qty - stock_qty)
    
    # Round up to nearest safety stock level
    if item.safety_stock:
        reorder_qty = max(reorder_qty, item.safety_stock)
    
    return {
        "item_code": item_code,
        "current_stock": stock_qty,
        "daily_consumption": round(daily_consumption, 2),
        "days_ahead": days_ahead,
        "recommended_reorder_qty": round(reorder_qty, 2),
        "estimated_stockout_days": round(stock_qty / daily_consumption if daily_consumption > 0 else 999, 1)
    }

import frappe
 
def validate_invoice_pricing(invoice_name: str):
    """
    Validate invoice pricing against price list and flag anomalies.
    
    Args:
        invoice_name (str): Sales Invoice name
        
    Returns:
        dict: Validation results with any issues found
    """
    invoice = frappe.get_doc("Sales Invoice", invoice_name)
    issues = []
    
    for item in invoice.items:
        # Get price list rate
        price_list_rate = frappe.db.get_value(
            "Item Price",
            {
                "item_code": item.item_code,
                "price_list": invoice.selling_price_list
            },
            "price_list_rate"
        )
        
        if not price_list_rate:
            issues.append({
                "item": item.item_code,
                "issue": "No price list rate found",
                "severity": "high"
            })
            continue
        
        # Check for significant variance
        variance_pct = ((item.rate - price_list_rate) / price_list_rate) * 100
        
        if abs(variance_pct) > 10:
            issues.append({
                "item": item.item_code,
                "issue": f"Price varies {variance_pct:.1f}% from price list",
                "expected": price_list_rate,
                "actual": item.rate,
                "severity": "medium" if abs(variance_pct) < 20 else "high"
            })
    
    return {
        "invoice": invoice_name,
        "is_valid": len(issues) == 0,
        "issues_found": len(issues),
        "issues": issues
    }

def get_customer_balance(customer_id: str):
    """Get customer outstanding balance."""
    
    # Validate input
    if not customer_id:
        frappe.throw("Customer ID is required")
    
    # Check if customer exists
    if not frappe.db.exists("Customer", customer_id):
        frappe.throw(f"Customer {customer_id} not found")
    
    try:
        balance = frappe.db.get_value("Customer", customer_id, "outstanding_amount")
        return {"customer_id": customer_id, "balance": balance or 0}
    except Exception as e:
        frappe.log_error(f"Error getting balance for {customer_id}: {str(e)}")
        frappe.throw(f"Failed to retrieve customer balance: {str(e)}")

def delete_customer_document(doc_name: str):
    """Delete a customer (respects permissions)."""
    
    # Check if user has permission
    if not frappe.has_permission("Customer", "delete", doc_name):
        frappe.throw("You don't have permission to delete this customer")
    
    # Verify document exists
    if not frappe.db.exists("Customer", doc_name):
        frappe.throw(f"Customer {doc_name} not found")
    
    # Delete
    frappe.delete_doc("Customer", doc_name)
    
    return {"success": True, "message": f"Customer {doc_name} deleted"}

# BAD: Multiple queries in loop
def get_customer_orders_slow(customer_id):
    orders = frappe.get_all("Sales Order", {"customer": customer_id}, ["name"])
    results = []
    for order in orders:
        doc = frappe.get_doc("Sales Order", order.name)  # Separate query each time
        results.append(doc.as_dict())
    return results
 
# GOOD: Single query with all data
def get_customer_orders_fast(customer_id):
    orders = frappe.get_all(
        "Sales Order",
        filters={"customer": customer_id},
        fields=["name", "grand_total", "status", "delivery_date"]
    )
    return {"customer_id": customer_id, "orders": orders}

>>> from my_app.api.tools import calculate_customer_ltv
>>> result = calculate_customer_ltv("CUST-001")
>>> print(result)

def convert_currency(amount: float, from_currency: str, to_currency: str):
    """
    Convert an amount from one currency to another using current rates.
    
    Args:
        amount (float): Amount to convert
        from_currency (str): Source currency code (USD, EUR, GBP, etc.)
        to_currency (str): Target currency code
        
    Returns:
        dict: Converted amount and exchange rate used
        
    Example:
        >>> convert_currency(100, "USD", "EUR")
        {"amount": 100, "from": "USD", "to": "EUR", "converted": 85.50, "rate": 0.855}
    """
    # Implementation
    pass

def send_email_notification(recipient: str, subject: str, message: str):
    """Send email (with validation)."""
    
    # Validate email format
    if not frappe.utils.validate_email_address(recipient):
        frappe.throw(f"Invalid email address: {recipient}")
    
    # Whitelist allowed recipients (optional but safer)
    allowed_domains = ["yourcompany.com", "partner.com"]
    domain = recipient.split("@")[1]
    if domain not in allowed_domains:
        frappe.throw(f"Can only send to {', '.join(allowed_domains)} addresses")
    
    # Send email
    frappe.sendmail(recipients=[recipient], subject=subject, message=message)
    
    return {"success": True, "recipient": recipient}