app.py

import gradio as gr
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.tools.decorator import tool
from datetime import datetime, timedelta
from dotenv import load_dotenv
load_dotenv()

# ✅ Tool definition
@tool(name="treasury_data", description="Returns detailed account balances, FX exposure, and liabilities")
def get_treasury_data(input: str) -> str:
    today = datetime.now().strftime("%Y-%m-%d")
    due_date = (datetime.now() + timedelta(days=5)).strftime("%Y-%m-%d")
    return f"""
    As of {today}, here is the treasury snapshot:

    ✅ Cash Balances:
    - USD Operating Account: $2,800,000
    - EUR Revenue Account: €1,500,000
    - GBP Payroll Account: £600,000

    📉 FX Market Rates:
    - EUR/USD: 1.08
    - GBP/USD: 1.27

    🧾 Upcoming Liabilities:
    - USD Vendor Payments: $2,200,000 due on {due_date}
    - EUR Convertible Bond Maturity: €900,000 due on {due_date}
    - GBP Payroll Run: £500,000 due on {due_date}

    📊 Policy Thresholds:
    - Minimum USD liquidity buffer: $500,000
    - FX hedge threshold for EUR: 70% exposure
    """

# ✅ Agent definition
agent = Agent(
    model=OpenAIChat(id="gpt-4o"),
    tools=[get_treasury_data],
    instructions=[
        "You are a Treasury Analyst AI.",
        "Analyze liquidity and FX exposure.",
        "Recommend actions such as FX hedging, internal fund transfers, or delay of liabilities.",
        "Always ensure minimum liquidity buffers are met.",
        "Convert foreign currency exposures to USD for a consolidated view.",
        "Provide your analysis in bullet points with numbers.",
    ],
    markdown=True,
)

def ask_agent(messages, history):
    print("====== DEBUG: ask_agent START ======")
    print("📥 Incoming messages:", messages)
    print("📚 Chat history:", history)

    # ✅ Handle Hugging Face's tendency to send just a string
    if isinstance(messages, str):
        messages = [{"role": "user", "content": messages}]
    elif isinstance(messages, list):
        # Defensive fix: if list of strings, wrap each
        if all(isinstance(m, str) for m in messages):
            messages = [{"role": "user", "content": m} for m in messages]
        # If it's a list but not valid messages, throw a descriptive error
        elif not all(isinstance(m, dict) and "role" in m and "content" in m for m in messages):
            print("❌ Malformed message list")
            return {"role": "assistant", "content": "Sorry, I couldn't understand the message format."}
    else:
        print("❌ Completely invalid input type")
        return {"role": "assistant", "content": "Sorry, I didn't understand that input."}

    try:
        latest_user_message = messages[-1]["content"]
        print(f"💬 Latest user message: {latest_user_message}")
    except Exception as e:
        print(f"❌ Error extracting latest message: {e}")
        return {"role": "assistant", "content": "Sorry, I couldn't read your message."}

    try:
        response = agent.run(latest_user_message)
        print(f"✅ Agent response:\n{response}")
        return {"role": "assistant", "content": response.content}
    except Exception as e:
        print(f"❌ Agent failed: {e}")
        return {"role": "assistant", "content": "Something went wrong on my end. Try again?"}

def yes(message, history):
    return "yes"

# ✅ Launch Gradio UI
gr.ChatInterface(
    fn=ask_agent,
    title="AI Treasury Assistant",
    description="Ask about cash positions, FX risk, or liquidity outlook.",
    chatbot=gr.Chatbot(show_copy_button=True, type="messages"),  # ✅ Fix here
    type="messages"  # ✅ And here
).launch()