main.py

"""

AgentForge - Hugging Face Space Template

This is a generic, reusable agent runner that reads configuration from environment variables.

"""
import os
import json
from fastapi import FastAPI, Request, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
from typing import Optional, List, Dict, Any
from agents import Agent, AsyncOpenAI as AgentsAsyncOpenAI, OpenAIChatCompletionsModel, function_tool, Runner, SQLiteSession

# ============================================
# Load Agent Configuration from Environment
# ============================================
AGENT_CONFIG_STR = os.getenv("AGENT_CONFIG")
if not AGENT_CONFIG_STR:
    raise ValueError("AGENT_CONFIG environment variable is required")

# Parse the config - handle both nested and flat structures
raw_config = json.loads(AGENT_CONFIG_STR)

# Handle nested structure (from full API response)
if isinstance(raw_config, dict):
    # Check if it's the full response structure with result.agent_build
    if "result" in raw_config and "agent_build" in raw_config.get("result", {}):
        AGENT_CONFIG = raw_config["result"]["agent_build"]
    # Check if it's nested under a different key
    elif "agent_build" in raw_config:
        AGENT_CONFIG = raw_config["agent_build"]
    # Otherwise assume it's already the flat agent_build structure
    else:
        AGENT_CONFIG = raw_config
else:
    AGENT_CONFIG = raw_config

# Validate that we have the required fields
if not isinstance(AGENT_CONFIG, dict):
    raise ValueError("AGENT_CONFIG must be a dictionary")

# API Keys from environment
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
GEMINI_API_KEY = os.getenv("GEMINI_API_KEY")
GROK_API_KEY = os.getenv("GROK_API_KEY")

# ============================================
# FastAPI App Setup
# ============================================
app = FastAPI(
    title=f"{AGENT_CONFIG.get('name', 'Agent')} API",
    description=f"Deployed agent for {AGENT_CONFIG.get('business_context', {}).get('business_name', 'Business')}",
    version="1.0.0"
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# ============================================
# Request/Response Models
# ============================================
class ChatRequest(BaseModel):
    message: str = Field(..., description="User message to the agent")
    session_id: Optional[str] = Field(default="default", description="Session ID for conversation tracking")

class ChatResponse(BaseModel):
    status: str
    agent_name: Optional[str] = None  # May be missing in config
    user_message: str
    agent_response: str
    tools_available: List[str]
    timestamp: float

# ============================================
# Dynamic Tool Recreation
# ============================================
def recreate_tools_from_config(domain: str, business_name: str):
    """

    Recreate tools based on domain.

    This mirrors the DynamicToolFactory logic from agent_architect.py

    """
    
    if domain == "pharmacy":
        @function_tool
        async def manage_prescription(action: str, prescription_id: str = None, patient_id: str = None, medication: str = None) -> dict:
            """Manage prescriptions - check, refill, or create"""
            from datetime import datetime
            return {"prescription_id": prescription_id or f"RX-{datetime.now().strftime('%Y%m%d%H%M')}", 
                    "action": action, "status": "Processed", "refills": 3}
        
        @function_tool
        async def check_drug_inventory(medication_name: str) -> dict:
            """Check medication stock and expiry"""
            return {"medication": medication_name, "in_stock": True, "quantity": 250, "expiry": "2026-06-15"}
        
        @function_tool
        async def get_patient_info(patient_id: str) -> dict:
            """Retrieve patient records and allergies"""
            return {"patient_id": patient_id, "allergies": ["Penicillin"], "medications": ["Metformin"]}
        
        @function_tool
        def web_search(query: str) -> dict:
            """Perform a web search for current information"""
            return {"query": query, "results": "Web search functionality - integrate with real API"}
        
        return [manage_prescription, check_drug_inventory, get_patient_info, web_search]
    
    elif domain == "ecommerce":
        @function_tool
        async def search_products(query: str, category: str = None) -> dict:
            """Search product catalog"""
            return {"query": query, "results": [{"id": "P001", "name": query, "price": 49.99, "stock": 50}]}
        
        @function_tool
        async def track_order(order_id: str) -> dict:
            """Track order status and delivery"""
            return {"order_id": order_id, "status": "In Transit", "eta": "2025-11-20", "location": "Distribution Center"}
        
        @function_tool
        async def manage_cart(action: str, product_id: str = None, quantity: int = 1) -> dict:
            """Add, remove, or view cart items"""
            return {"action": action, "product_id": product_id, "cart_total": 149.99, "items": 3}
        
        @function_tool
        def web_search(query: str) -> dict:
            """Perform a web search for current information"""
            return {"query": query, "results": "Web search functionality"}
        
        return [search_products, track_order, manage_cart, web_search]
    
    elif domain == "weather":
        @function_tool
        async def get_forecast(location: str, days: int = 7) -> dict:
            """Get weather forecast"""
            return {"location": location, "days": days, "forecast": [{"date": "2025-12-12", "high": 22, "low": 15, "condition": "partly cloudy"}]}
        
        @function_tool
        async def severe_weather_alert(location: str) -> dict:
            """Check for severe weather alerts"""
            return {"location": location, "alerts": [], "severity": "none", "preparedness_tips": ["Normal precautions"]}
        
        @function_tool
        async def historical_weather_comparison(location: str, date: str) -> dict:
            """Compare current weather to historical data"""
            return {"location": location, "date": date, "current_temp": 20, "historical_avg": 18, "difference": 2, "percentile": 65}
        
        @function_tool
        def web_search(query: str) -> dict:
            """Perform a web search for current information"""
            return {"query": query, "results": "Web search functionality"}
        
        return [get_forecast, severe_weather_alert, historical_weather_comparison, web_search]
    
    # Add more domains as needed...
    else:  # generic
        @function_tool
        async def generate_analytics(metric: str, time_range: str) -> dict:
            """Generate business analytics"""
            return {"metric": metric, "time_range": time_range, "value": 12500, "trend": "+15%", "insights": f"{metric} growing"}
        
        @function_tool
        async def send_notification(recipient: str, message: str, channel: str = "email") -> dict:
            """Send notifications"""
            return {"recipient": recipient, "message": message, "channel": channel, "status": "Sent"}
        
        @function_tool
        def web_search(query: str) -> dict:
            """Perform a web search for current information"""
            return {"query": query, "results": "Web search functionality"}
        
        return [generate_analytics, send_notification, web_search]

# ============================================
# Initialize Agent
# ============================================
def initialize_agent():
    """Initialize the agent with configuration from environment"""
    model = AGENT_CONFIG.get("model", "gpt-4o")
    
    # Select appropriate API key and client
    if "gemini" in model.lower():
        api_key = GEMINI_API_KEY
        client = AgentsAsyncOpenAI(api_key=api_key, base_url="https://generativelanguage.googleapis.com/v1beta/openai/")
        model_name = "gemini-2.0-flash-exp"
    elif "grok" in model.lower():
        api_key = GROK_API_KEY
        client = AgentsAsyncOpenAI(api_key=api_key, base_url="https://api.x.ai/v1")
        model_name = "grok-beta"
    else:
        api_key = OPENAI_API_KEY
        client = AgentsAsyncOpenAI(api_key=api_key)
        model_name = "gpt-4o"
    
    if not api_key:
        raise ValueError(f"API key not found for model: {model}")
    
    MODEL = OpenAIChatCompletionsModel(model=model_name, openai_client=client)
    
    # Recreate tools - handle both nested and flat business_context
    business_context = AGENT_CONFIG.get("business_context", {})
    if not isinstance(business_context, dict):
        business_context = {}
    
    domain = business_context.get("domain") or AGENT_CONFIG.get("domain", "generic")
    business_name = business_context.get("business_name") or AGENT_CONFIG.get("business_name", "Business")
    tools = recreate_tools_from_config(domain, business_name)
    
    # Get agent name - try multiple possible keys
    agent_name = AGENT_CONFIG.get("name") or AGENT_CONFIG.get("agent_name", "AI Agent")
    
    # Get instructions
    instructions = AGENT_CONFIG.get("instructions", "You are a helpful AI assistant.")
    
    # Create agent
    agent = Agent(
        name=agent_name,
        instructions=instructions,
        model=MODEL,
        tools=tools
    )
    
    return agent, tools

# Initialize agent on startup
AGENT_INSTANCE, AGENT_TOOLS = initialize_agent()

# ============================================
# API Endpoints
# ============================================
@app.get("/")
async def root():
    """Health check and agent info"""
    # Extract tool names properly
    tool_names = []
    for tool in AGENT_TOOLS:
        if hasattr(tool, '__name__'):
            tool_names.append(tool.__name__)
        elif hasattr(tool, 'name'):
            tool_names.append(tool.name)
        else:
            # Try to extract from string representation
            tool_str = str(tool)
            if "name='" in tool_str:
                try:
                    name_start = tool_str.index("name='") + 6
                    name_end = tool_str.index("'", name_start)
                    tool_names.append(tool_str[name_start:name_end])
                except:
                    tool_names.append(str(tool)[:50])  # Truncate long strings
            else:
                tool_names.append(str(tool)[:50])
    
    return {
        "status": "online",
        "agent_name": AGENT_CONFIG.get("name") or AGENT_CONFIG.get("agent_name") or "GenericAgent",
        "agent_id": AGENT_CONFIG.get("agent_id"),
        "business": AGENT_CONFIG.get("business_context", {}).get("business_name") if isinstance(AGENT_CONFIG.get("business_context"), dict) else None,
        "domain": AGENT_CONFIG.get("business_context", {}).get("domain") if isinstance(AGENT_CONFIG.get("business_context"), dict) else AGENT_CONFIG.get("domain"),
        "tools_count": len(AGENT_TOOLS),
        "tools": tool_names,
        "model": AGENT_CONFIG.get("model"),
        "deployment": "Hugging Face Space"
    }

@app.post("/run", response_model=ChatResponse)
async def run_agent(request: ChatRequest) -> ChatResponse:
    """

    Main endpoint to interact with the agent.

    This is the primary interface for users.

    """
    import time
    
    try:
        # Run the agent
        runner = Runner()
        temp_session = SQLiteSession(":memory:")
        
        response = await runner.run(AGENT_INSTANCE, request.message, session=temp_session)
        final_output = str(response.final_output) if hasattr(response, 'final_output') else str(response)
        
        return ChatResponse(
            status="success",
            agent_name=AGENT_CONFIG.get("name", "GenericAgent"),
            user_message=request.message,
            agent_response=final_output,
            tools_available=[tool.__name__ if hasattr(tool, '__name__') else str(tool) for tool in AGENT_TOOLS],
            timestamp=time.time()
        )
        
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Agent execution error: {str(e)}")

@app.get("/config")
async def get_config():
    """Get agent configuration (without sensitive data)"""
    safe_config = {
        "agent_id": AGENT_CONFIG.get("agent_id"),
        "name": AGENT_CONFIG.get("name"),
        "model": AGENT_CONFIG.get("model"),
        "business_context": AGENT_CONFIG.get("business_context"),
        "tools_count": len(AGENT_TOOLS),
        "deployment_ready": AGENT_CONFIG.get("deployment_ready")
    }
    return safe_config

@app.get("/health")
async def health_check():
    """Health check endpoint"""
    return {"status": "healthy", "agent": AGENT_CONFIG.get("name")}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=7860)