"""Standalone Agent CRUD routes""" from typing import Optional from fastapi import APIRouter, Depends from pydantic import BaseModel from sqlalchemy.orm import Session from luxx.database import get_db from luxx.models import Agent, LLMProvider, User from luxx.routes.auth import get_current_user from luxx.utils.helpers import success_response, error_response router = APIRouter(prefix="/agents", tags=["Agents"]) class AgentCreate(BaseModel): name: str role: str = "" provider_id: Optional[int] = None model: str = "" system_prompt: str = "You are a helpful AI assistant." color: str = "#2563eb" class AgentUpdate(BaseModel): name: Optional[str] = None role: Optional[str] = None provider_id: Optional[int] = None model: Optional[str] = None system_prompt: Optional[str] = None color: Optional[str] = None @router.get("/", response_model=dict) def list_agents( current_user: User = Depends(get_current_user), db: Session = Depends(get_db) ): """List all agents for current user""" agents = db.query(Agent).filter( Agent.user_id == current_user.id ).order_by(Agent.updated_at.desc()).all() return success_response(data=[a.to_dict() for a in agents]) @router.post("/", response_model=dict) def create_agent( data: AgentCreate, current_user: User = Depends(get_current_user), db: Session = Depends(get_db) ): """Create a new agent""" model = data.model provider_id = data.provider_id if provider_id and not model: provider = db.query(LLMProvider).filter( LLMProvider.id == provider_id, LLMProvider.user_id == current_user.id ).first() if provider: model = provider.default_model if not model: default_provider = db.query(LLMProvider).filter( LLMProvider.user_id == current_user.id, LLMProvider.is_default == True ).first() if default_provider: provider_id = default_provider.id model = default_provider.default_model if not model: model = "gpt-4" agent = Agent( user_id=current_user.id, name=data.name, role=data.role, provider_id=provider_id, model=model, system_prompt=data.system_prompt, color=data.color ) db.add(agent) db.commit() db.refresh(agent) return success_response(data=agent.to_dict(), message="Agent created") @router.get("/{agent_id}", response_model=dict) def get_agent( agent_id: int, current_user: User = Depends(get_current_user), db: Session = Depends(get_db) ): """Get agent details""" agent = db.query(Agent).filter( Agent.id == agent_id, Agent.user_id == current_user.id ).first() if not agent: return error_response("Agent not found", 404) return success_response(data=agent.to_dict()) @router.put("/{agent_id}", response_model=dict) def update_agent( agent_id: int, data: AgentUpdate, current_user: User = Depends(get_current_user), db: Session = Depends(get_db) ): """Update an agent""" agent = db.query(Agent).filter( Agent.id == agent_id, Agent.user_id == current_user.id ).first() if not agent: return error_response("Agent not found", 404) update_data = data.dict(exclude_unset=True) for key, value in update_data.items(): setattr(agent, key, value) db.commit() db.refresh(agent) return success_response(data=agent.to_dict(), message="Agent updated") @router.delete("/{agent_id}", response_model=dict) def delete_agent( agent_id: int, current_user: User = Depends(get_current_user), db: Session = Depends(get_db) ): """Delete an agent""" agent = db.query(Agent).filter( Agent.id == agent_id, Agent.user_id == current_user.id ).first() if not agent: return error_response("Agent not found", 404) db.delete(agent) db.commit() return success_response(message="Agent deleted")