shekkari21/agent-from-scratch

1"""FastAPI web application for the Agent Framework."""2 3import os4import sys5import uuid6import shutil7from pathlib import Path8from typing import Optional, List9from datetime import datetime10 11# Add parent directory to path12sys.path.insert(0, str(Path(__file__).parent.parent))13 14from fastapi import FastAPI, UploadFile, File, Form, HTTPException15from fastapi.staticfiles import StaticFiles16from fastapi.responses import HTMLResponse, FileResponse17from fastapi.middleware.cors import CORSMiddleware18from pydantic import BaseModel19from dotenv import load_dotenv20 21from agent_framework import (22    Agent, LlmClient, InMemorySessionManager, 23    display_trace, ExecutionContext, format_trace24)25from agent_tools import calculator, search_web, read_file, list_files, unzip_file, read_media_file26 27# Load environment variables28load_dotenv()29 30app = FastAPI(title="Agent Chat", description="AI Agent with Tools")31 32# Enable CORS33app.add_middleware(34    CORSMiddleware,35    allow_origins=["*"],36    allow_credentials=True,37    allow_methods=["*"],38    allow_headers=["*"],39)40 41# Global session manager (shared across requests)42session_manager = InMemorySessionManager()43 44# Upload directory for files45UPLOAD_DIR = Path(__file__).parent / "uploads"46UPLOAD_DIR.mkdir(exist_ok=True)47 48# Available tools49TOOLS = [calculator, search_web, read_file, list_files, unzip_file, read_media_file]50 51# Create agent  52def create_agent(use_session: bool = True) -> Agent:53    """Create an agent instance."""54    55    # Include the actual upload directory path in instructions56    upload_path = str(UPLOAD_DIR.absolute())57    58    instructions = f"""You are a helpful AI assistant with access to various tools.59 60You can:61- Perform calculations using the calculator62- Search the web for current information63- Read excel files using the read_file tool64- List files in directories using the list_files tool65- Extract zip files using the unzip_file tool66- Read pdf using read_media_file67 68IMPORTANT - Uploaded files location:69Files uploaded by users are stored at: {upload_path}70To see uploaded files, use: list_files("{upload_path}")71To read a file, use: read_file("{upload_path}/filename.ext")72 73Always be helpful and use your tools when needed to provide accurate answers."""74 75    return Agent(76        model=LlmClient(model="gpt-4o-mini"),77        tools=TOOLS,78        instructions=instructions,79        max_steps=10,80        session_manager=session_manager if use_session else None81    )82 83 84# Pydantic models for API85class ChatRequest(BaseModel):86    message: str87    session_id: Optional[str] = None88    use_session: bool = True89 90 91class ChatResponse(BaseModel):92    response: str93    session_id: str94    events_count: int95    tools_used: List[str]96    trace_text: str = ""  # Simple text-based trace like display_trace97 98 99class ToolInfo(BaseModel):100    name: str101    description: str102 103 104class SessionInfo(BaseModel):105    session_id: str106    events_count: int107    created_at: str108 109 110# API Endpoints111@app.get("/")112async def root():113    """Serve the chat interface."""114    return FileResponse(Path(__file__).parent / "static" / "index.html")115 116 117@app.get("/api/tools")118async def get_tools() -> List[ToolInfo]:119    """Get list of available tools."""120    return [121        ToolInfo(122            name=tool.name,123            description=tool.description[:100] + "..." if len(tool.description) > 100 else tool.description124        )125        for tool in TOOLS126    ]127 128 129@app.post("/api/chat")130async def chat(request: ChatRequest) -> ChatResponse:131    """Send a message to the agent."""132    133    # Generate or use provided session ID134    session_id = request.session_id or str(uuid.uuid4())135    136    # Create agent137    agent = create_agent(use_session=request.use_session)138    139    try:140        # Run the agent141        if request.use_session:142            result = await agent.run(request.message, session_id=session_id)143        else:144            result = await agent.run(request.message)145        146        # Extract tools used147        tools_used = []148        for event in result.context.events:149            for item in event.content:150                if hasattr(item, 'name') and item.type == "tool_call":151                    if item.name not in tools_used:152                        tools_used.append(item.name)153        154        # Use your format_trace function directly!155        trace_text = format_trace(result.context)156        157        return ChatResponse(158            response=str(result.output) if result.output else "I couldn't generate a response.",159            session_id=session_id,160            events_count=len(result.context.events),161            tools_used=tools_used,162            trace_text=trace_text163        )164    except Exception as e:165        raise HTTPException(status_code=500, detail=str(e))166 167 168@app.post("/api/upload")169async def upload_file(file: UploadFile = File(...)):170    """Upload a file for the agent to access."""171    172    # Save file to uploads directory173    file_path = UPLOAD_DIR / file.filename174    175    try:176        with open(file_path, "wb") as buffer:177            shutil.copyfileobj(file.file, buffer)178        179        return {180            "filename": file.filename,181            "path": str(file_path),182            "size": file_path.stat().st_size,183            "message": f"File uploaded successfully. You can reference it at: {file_path}"184        }185    except Exception as e:186        raise HTTPException(status_code=500, detail=str(e))187 188 189@app.get("/api/uploads")190async def list_uploads():191    """List uploaded files."""192    files = []193    for f in UPLOAD_DIR.iterdir():194        if f.is_file() and not f.name.startswith('.'):195            files.append({196                "name": f.name,197                "path": str(f),198                "size": f.stat().st_size199            })200    return files201 202 203@app.delete("/api/uploads/{filename}")204async def delete_upload(filename: str):205    """Delete an uploaded file."""206    file_path = UPLOAD_DIR / filename207    if file_path.exists():208        file_path.unlink()209        return {"message": f"Deleted {filename}"}210    raise HTTPException(status_code=404, detail="File not found")211 212 213@app.get("/api/sessions")214async def list_sessions() -> List[SessionInfo]:215    """List all active sessions."""216    sessions = []217    for sid, session in session_manager._sessions.items():218        sessions.append(SessionInfo(219            session_id=sid,220            events_count=len(session.events),221            created_at=session.created_at.isoformat()222        ))223    return sessions224 225 226@app.delete("/api/sessions/{session_id}")227async def delete_session(session_id: str):228    """Delete a session to clear conversation history."""229    if session_id in session_manager._sessions:230        del session_manager._sessions[session_id]231        return {"message": f"Session {session_id} cleared"}232    raise HTTPException(status_code=404, detail="Session not found")233 234 235# Mount static files236static_dir = Path(__file__).parent / "static"237static_dir.mkdir(exist_ok=True)238app.mount("/static", StaticFiles(directory=static_dir), name="static")239 240 241if __name__ == "__main__":242    import uvicorn243    port = int(os.getenv("PORT", 7860))244    uvicorn.run(app, host="0.0.0.0", port=port)245 246