Spaces:
Configuration error
Configuration error
File size: 3,408 Bytes
e280b6e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 | import uuid
import logging
from datetime import datetime
from typing import List, Optional, Dict, Any
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy import select, desc
from sqlalchemy.ext.asyncio import AsyncSession
from app.database.postgres import get_db, AgentRun, Email, User
from app.database.mongo import mongo_db
from app.auth.security import get_current_user
from app.tasks.email_tasks import fetch_emails_task
from pydantic import BaseModel
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/agent", tags=["agent"])
class AgentRunListItem(BaseModel):
id: str
action_type: str
status: str
started_at: str
completed_at: Optional[str] = None
email_subject: Optional[str] = None
email_id: Optional[str] = None
class AgentLogItem(BaseModel):
step: int
thought: str
action: str
result: str
timestamp: str
@router.post("/trigger-sync")
async def trigger_sync(current_user: User = Depends(get_current_user)):
"""Manually trigger the email synchronization task in the background."""
task = fetch_emails_task.delay()
return {"status": "triggered", "task_id": task.id}
@router.get("/runs", response_model=List[AgentRunListItem])
async def list_agent_runs(
db: AsyncSession = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Retrieve history of agent executions, joining email subjects where applicable."""
# Custom join to fetch email subject
query = (
select(AgentRun, Email.subject)
.outerjoin(Email, AgentRun.email_id == Email.id)
.where(AgentRun.user_id == current_user.id)
.order_by(desc(AgentRun.started_at))
.limit(30)
)
result = await db.execute(query)
rows = result.all()
runs = []
for row in rows:
run_obj, email_subj = row
runs.append(
AgentRunListItem(
id=str(run_obj.id),
action_type=run_obj.action_type,
status=run_obj.status,
started_at=run_obj.started_at.isoformat(),
completed_at=run_obj.completed_at.isoformat() if run_obj.completed_at else None,
email_subject=email_subj,
email_id=str(run_obj.email_id) if run_obj.email_id else None
)
)
return runs
@router.get("/runs/{run_id}/logs", response_model=List[AgentLogItem])
async def get_run_logs(
run_id: str,
current_user: User = Depends(get_current_user)
):
"""Retrieve step-by-step logic and thoughts of the AI agent from MongoDB."""
# Ensure MongoDB client connection
if mongo_db.db is None:
mongo_db.connect()
try:
# Fetch logs for the run, ordered by step
cursor = mongo_db.agent_logs.find({"run_id": run_id}).sort("step", 1)
logs = await cursor.to_list(length=100)
return [
AgentLogItem(
step=log.get("step", 0),
thought=log.get("thought", ""),
action=log.get("action", ""),
result=log.get("result", ""),
timestamp=log.get("timestamp", datetime.utcnow()).isoformat()
)
for log in logs
]
except Exception as e:
logger.error(f"Error fetching agent logs: {e}")
raise HTTPException(status_code=500, detail="Failed to fetch execution logs")
|