AgentraX/agent-saas-typescript-project-template
0
1"""2AgentForge - Hugging Face Space Template3This is a generic, reusable agent runner that reads configuration from environment variables.4"""5import os6import json7from fastapi import FastAPI, Request, HTTPException8from fastapi.middleware.cors import CORSMiddleware9from pydantic import BaseModel, Field10from typing import Optional, List, Dict, Any11from agents import Agent, AsyncOpenAI as AgentsAsyncOpenAI, OpenAIChatCompletionsModel, function_tool, Runner, SQLiteSession12import aiosmtplib13from email.message import EmailMessage14# ============================================15# Load Agent Configuration from Environment16# ============================================17AGENT_CONFIG_STR = os.getenv("AGENT_CONFIG")18if not AGENT_CONFIG_STR:19 raise ValueError("AGENT_CONFIG environment variable is required")20 21# Parse the config - handle both nested and flat structures22try:23 raw_config = json.loads(AGENT_CONFIG_STR)24except json.JSONDecodeError as e:25 raise ValueError(f"Failed to parse AGENT_CONFIG as JSON: {e}")26 27# Handle nested structure (from full API response)28if isinstance(raw_config, dict):29 # Check if it's the full response structure with result.agent_build30 if "result" in raw_config and "agent_build" in raw_config.get("result", {}):31 AGENT_CONFIG = raw_config["result"]["agent_build"]32 # Check if it's nested under a different key33 elif "agent_build" in raw_config:34 AGENT_CONFIG = raw_config["agent_build"]35 # Otherwise assume it's already the flat agent_build structure36 else:37 AGENT_CONFIG = raw_config38else:39 AGENT_CONFIG = raw_config40 41# Validate that we have the required fields42if not isinstance(AGENT_CONFIG, dict):43 raise ValueError(f"AGENT_CONFIG must be a dictionary, got {type(AGENT_CONFIG)}")44 45# Log config keys for debugging (in production, this helps identify issues)46print(f"Loaded AGENT_CONFIG with keys: {list(AGENT_CONFIG.keys())[:10]}...") # Print first 10 keys47 48# API Keys from environment49OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")50GEMINI_API_KEY = os.getenv("GEMINI_API_KEY")51GROK_API_KEY = os.getenv("GROK_API_KEY")52 53# ============================================54# FastAPI App Setup55# ============================================56app = FastAPI(57 title=f"{AGENT_CONFIG.get('name', 'Agent')} API",58 description=f"Deployed agent for {AGENT_CONFIG.get('business_context', {}).get('business_name', 'Business')}",59 version="1.0.0"60)61 62app.add_middleware(63 CORSMiddleware,64 allow_origins=["*"],65 allow_credentials=True,66 allow_methods=["*"],67 allow_headers=["*"],68)69 70# ============================================71# Request/Response Models72# ============================================73class ChatRequest(BaseModel):74 message: str = Field(..., description="User message to the agent")75 session_id: Optional[str] = Field(default="default", description="Session ID for conversation tracking")76 77class ChatResponse(BaseModel):78 status: str79 agent_name: Optional[str] = None # May be missing in config80 user_message: str81 agent_response: str82 tools_available: List[str]83 timestamp: float84 85# ============================================86# Dynamic Tool Recreation87# ============================================88def recreate_tools_from_config(domain: str, business_name: str):89 """90 Recreate tools based on domain.91 This mirrors the DynamicToolFactory logic from agent_architect.py92 """93 94 if domain == "pharmacy":95 @function_tool96 async def manage_prescription(action: str, prescription_id: str = None, patient_id: str = None, medication: str = None) -> dict:97 """Manage prescriptions - check, refill, or create"""98 from datetime import datetime99 return {"prescription_id": prescription_id or f"RX-{datetime.now().strftime('%Y%m%d%H%M')}", 100 "action": action, "status": "Processed", "refills": 3}101 102 @function_tool103 async def check_drug_inventory(medication_name: str) -> dict:104 """Check medication stock and expiry"""105 return {"medication": medication_name, "in_stock": True, "quantity": 250, "expiry": "2026-06-15"}106 107 @function_tool108 async def get_patient_info(patient_id: str) -> dict:109 """Retrieve patient records and allergies"""110 return {"patient_id": patient_id, "allergies": ["Penicillin"], "medications": ["Metformin"]}111 112 @function_tool113 def web_search(query: str) -> dict:114 """Perform a web search for current information"""115 return {"query": query, "results": "Web search functionality - integrate with real API"}116 117 return [manage_prescription, check_drug_inventory, get_patient_info, web_search]118 119 elif domain == "ecommerce":120 @function_tool121 async def search_products(query: str, category: str = None) -> dict:122 """Search product catalog"""123 return {"query": query, "results": [{"id": "P001", "name": query, "price": 49.99, "stock": 50}]}124 125 @function_tool126 async def track_order(order_id: str) -> dict:127 """Track order status and delivery"""128 return {"order_id": order_id, "status": "In Transit", "eta": "2025-11-20", "location": "Distribution Center"}129 130 @function_tool131 async def manage_cart(action: str, product_id: str = None, quantity: int = 1) -> dict:132 """Add, remove, or view cart items"""133 return {"action": action, "product_id": product_id, "cart_total": 149.99, "items": 3}134 135 @function_tool136 def web_search(query: str) -> dict:137 """Perform a web search for current information"""138 return {"query": query, "results": "Web search functionality"}139 140 return [search_products, track_order, manage_cart, web_search]141 142 elif domain == "weather":143 @function_tool144 async def get_forecast(location: str, days: int = 7) -> dict:145 """Get weather forecast"""146 return {"location": location, "days": days, "forecast": [{"date": "2025-12-12", "high": 22, "low": 15, "condition": "partly cloudy"}]}147 148 @function_tool149 async def severe_weather_alert(location: str) -> dict:150 """Check for severe weather alerts"""151 return {"location": location, "alerts": [], "severity": "none", "preparedness_tips": ["Normal precautions"]}152 153 @function_tool154 async def historical_weather_comparison(location: str, date: str) -> dict:155 """Compare current weather to historical data"""156 return {"location": location, "date": date, "current_temp": 20, "historical_avg": 18, "difference": 2, "percentile": 65}157 158 @function_tool159 def web_search(query: str) -> dict:160 """Perform a web search for current information"""161 return {"query": query, "results": "Web search functionality"}162 163 return [get_forecast, severe_weather_alert, historical_weather_comparison, web_search]164 165 elif domain == "email_marketing":166 @function_tool167 async def draft_cold_email(recipient_name: str, company: str, pain_point: str, solution_offer: str) -> dict:168 """Draft a personalized cold email based on research and pain points"""169 return {170 "subject": f"Question regarding {company}'s {pain_point} strategy",171 "body": f"Hi {recipient_name},\n\nI noticed {company} might be facing challenges with {pain_point}. Our solution for {solution_offer} has helped similar companies...\n\nBest regards,\nAgent",172 "status": "drafted",173 "quality_score": 0.95174 }175 176 @function_tool177 async def verify_email_format(email: str) -> dict:178 """Verify if an email address is valid and formatted correctly"""179 is_valid = "@" in email and "." in email.split("@")[-1]180 return {"email": email, "is_valid": is_valid, "suggestion": None if is_valid else "Check format"}181 182 @function_tool183 async def send_email(to: str, subject: str, body: str, is_html: bool = True) -> dict:184 """Actually send an email using SMTP configurations from environment."""185 host = os.getenv("SMTP_HOST")186 port = int(os.getenv("SMTP_PORT", "587"))187 username = os.getenv("SMTP_USER")188 password = os.getenv("SMTP_PASSWORD")189 from_email = os.getenv("SMTP_FROM_EMAIL", username)190 191 if not all([host, username, password]):192 return {193 "status": "error",194 "message": "SMTP credentials (SMTP_HOST, SMTP_USER, SMTP_PASSWORD) are not configured in environment."195 }196 197 message = EmailMessage()198 message["From"] = from_email199 message["To"] = to200 message["Subject"] = subject201 if is_html:202 message.set_content(body, subtype="html")203 else:204 message.set_content(body)205 206 try:207 await aiosmtplib.send(208 message,209 hostname=host,210 port=port,211 username=username,212 password=password,213 use_tls=(port == 465),214 start_tls=(port == 587),215 )216 return {"to": to, "subject": subject, "status": "sent", "timestamp": "2024-02-09T12:00:00"}217 except Exception as e:218 return {"status": "error", "message": str(e)}219 220 @function_tool221 def web_search(query: str) -> dict:222 """Perform a web search for prospect research"""223 return {"query": query, "results": f"Research data for {query}"}224 225 return [draft_cold_email, verify_email_format, send_email, web_search]226 227 # Add more domains as needed...228 else: # generic229 @function_tool230 async def generate_analytics(metric: str, time_range: str) -> dict:231 """Generate business analytics"""232 return {"metric": metric, "time_range": time_range, "value": 12500, "trend": "+15%", "insights": f"{metric} growing"}233 234 @function_tool235 async def send_notification(recipient: str, message: str, channel: str = "email") -> dict:236 """Send notifications"""237 if channel == "email":238 host = os.getenv("SMTP_HOST")239 if host:240 # Implementation similar to send_email241 return {"recipient": recipient, "message": "Notification sent via actual email", "status": "Sent"}242 return {"recipient": recipient, "message": message, "channel": channel, "status": "Sent"}243 244 @function_tool245 async def send_email(to: str, subject: str, body: str, is_html: bool = True) -> dict:246 """Actually send an email using SMTP configurations from environment."""247 host = os.getenv("SMTP_HOST")248 port = int(os.getenv("SMTP_PORT", "587"))249 username = os.getenv("SMTP_USER")250 password = os.getenv("SMTP_PASSWORD")251 from_email = os.getenv("SMTP_FROM_EMAIL", username)252 253 if not all([host, username, password]):254 return {255 "status": "error",256 "message": "SMTP credentials (SMTP_HOST, SMTP_USER, SMTP_PASSWORD) are not configured in environment."257 }258 259 message = EmailMessage()260 message["From"] = from_email261 message["To"] = to262 message["Subject"] = subject263 if is_html:264 message.set_content(body, subtype="html")265 else:266 message.set_content(body)267 268 try:269 await aiosmtplib.send(270 message,271 hostname=host,272 port=port,273 username=username,274 password=password,275 use_tls=(port == 465),276 start_tls=(port == 587),277 )278 return {"to": to, "subject": subject, "status": "sent", "timestamp": "2024-02-09T12:00:00"}279 except Exception as e:280 return {"status": "error", "message": str(e)}281 282 @function_tool283 def web_search(query: str) -> dict:284 """Perform a web search for current information"""285 return {"query": query, "results": "Web search functionality"}286 287 return [generate_analytics, send_notification, send_email, web_search]288 289# ============================================290# Initialize Agent291# ============================================292def initialize_agent():293 """Initialize the agent with configuration from environment"""294 model = AGENT_CONFIG.get("model", "gpt-4o")295 296 # Select appropriate API key and client297 if "gemini" in model.lower():298 api_key = GEMINI_API_KEY299 client = AgentsAsyncOpenAI(api_key=api_key, base_url="https://generativelanguage.googleapis.com/v1beta/openai/")300 model_name = "gemini-2.0-flash-exp"301 elif "grok" in model.lower():302 api_key = GROK_API_KEY303 client = AgentsAsyncOpenAI(api_key=api_key, base_url="https://api.x.ai/v1")304 model_name = "grok-beta"305 else:306 api_key = OPENAI_API_KEY307 client = AgentsAsyncOpenAI(api_key=api_key)308 model_name = "gpt-4o"309 310 if not api_key:311 raise ValueError(f"API key not found for model: {model}")312 313 MODEL = OpenAIChatCompletionsModel(model=model_name, openai_client=client)314 315 # Recreate tools - handle both nested and flat business_context316 business_context = AGENT_CONFIG.get("business_context", {})317 if not isinstance(business_context, dict):318 business_context = {}319 320 domain = business_context.get("domain") or AGENT_CONFIG.get("domain", "generic")321 business_name = business_context.get("business_name") or AGENT_CONFIG.get("business_name", "Business")322 tools = recreate_tools_from_config(domain, business_name)323 324 # Get agent name - try multiple possible keys325 agent_name = AGENT_CONFIG.get("name") or AGENT_CONFIG.get("agent_name", "AI Agent")326 327 # Get instructions328 instructions = AGENT_CONFIG.get("instructions", "You are a helpful AI assistant.")329 330 # Create agent331 agent = Agent(332 name=agent_name,333 instructions=instructions,334 model=MODEL,335 tools=tools336 )337 338 return agent, tools339 340# Initialize agent on startup341AGENT_INSTANCE, AGENT_TOOLS = initialize_agent()342 343# ============================================344# API Endpoints345# ============================================346@app.get("/")347async def root():348 """Health check and agent info"""349 # Extract tool names properly350 tool_names = []351 for tool in AGENT_TOOLS:352 if hasattr(tool, '__name__'):353 tool_names.append(tool.__name__)354 elif hasattr(tool, 'name'):355 tool_names.append(tool.name)356 else:357 # Try to extract from string representation358 tool_str = str(tool)359 if "name='" in tool_str:360 try:361 name_start = tool_str.index("name='") + 6362 name_end = tool_str.index("'", name_start)363 tool_names.append(tool_str[name_start:name_end])364 except:365 tool_names.append(str(tool)[:50]) # Truncate long strings366 else:367 tool_names.append(str(tool)[:50])368 369 return {370 "status": "online",371 "agent_name": AGENT_CONFIG.get("name") or AGENT_CONFIG.get("agent_name") or "GenericAgent",372 "agent_id": AGENT_CONFIG.get("agent_id"),373 "business": AGENT_CONFIG.get("business_context", {}).get("business_name") if isinstance(AGENT_CONFIG.get("business_context"), dict) else None,374 "domain": AGENT_CONFIG.get("business_context", {}).get("domain") if isinstance(AGENT_CONFIG.get("business_context"), dict) else AGENT_CONFIG.get("domain"),375 "tools_count": len(AGENT_TOOLS),376 "tools": tool_names,377 "model": AGENT_CONFIG.get("model"),378 "deployment": "Hugging Face Space"379 }380 381@app.post("/run", response_model=ChatResponse)382async def run_agent(request: ChatRequest) -> ChatResponse:383 """384 Main endpoint to interact with the agent.385 This is the primary interface for users.386 """387 import time388 389 try:390 # Run the agent391 runner = Runner()392 temp_session = SQLiteSession(":memory:")393 394 response = await runner.run(AGENT_INSTANCE, request.message, session=temp_session)395 final_output = str(response.final_output) if hasattr(response, 'final_output') else str(response)396 397 return ChatResponse(398 status="success",399 agent_name=AGENT_CONFIG.get("name", "GenericAgent"),400 user_message=request.message,401 agent_response=final_output,402 tools_available=[tool.__name__ if hasattr(tool, '__name__') else str(tool) for tool in AGENT_TOOLS],403 timestamp=time.time()404 )405 406 except Exception as e:407 raise HTTPException(status_code=500, detail=f"Agent execution error: {str(e)}")408 409@app.get("/config")410async def get_config():411 """Get agent configuration (without sensitive data)"""412 safe_config = {413 "agent_id": AGENT_CONFIG.get("agent_id"),414 "name": AGENT_CONFIG.get("name"),415 "model": AGENT_CONFIG.get("model"),416 "business_context": AGENT_CONFIG.get("business_context"),417 "tools_count": len(AGENT_TOOLS),418 "deployment_ready": AGENT_CONFIG.get("deployment_ready")419 }420 return safe_config421 422@app.get("/health")423async def health_check():424 """Health check endpoint"""425 return {"status": "healthy", "agent": AGENT_CONFIG.get("name") or AGENT_CONFIG.get("agent_name")}426 427@app.get("/debug/config")428async def debug_config():429 """Debug endpoint to see what config is loaded (without sensitive data)"""430 safe_config = {431 "has_config": bool(AGENT_CONFIG),432 "config_keys": list(AGENT_CONFIG.keys()) if isinstance(AGENT_CONFIG, dict) else [],433 "agent_name": AGENT_CONFIG.get("name") or AGENT_CONFIG.get("agent_name"),434 "agent_id": AGENT_CONFIG.get("agent_id"),435 "model": AGENT_CONFIG.get("model"),436 "has_business_context": "business_context" in AGENT_CONFIG,437 "business_context_type": type(AGENT_CONFIG.get("business_context")).__name__,438 "domain": AGENT_CONFIG.get("business_context", {}).get("domain") if isinstance(AGENT_CONFIG.get("business_context"), dict) else AGENT_CONFIG.get("domain"),439 "business_name": AGENT_CONFIG.get("business_context", {}).get("business_name") if isinstance(AGENT_CONFIG.get("business_context"), dict) else None,440 "tools_count_from_config": len(AGENT_CONFIG.get("tools", [])),441 "tools_count_loaded": len(AGENT_TOOLS),442 }443 return safe_config444 445if __name__ == "__main__":446 import uvicorn447 uvicorn.run(app, host="0.0.0.0", port=7860)448