""" Tool call tracking API endpoints. """ import json from datetime import datetime from typing import List, Optional from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select, func, desc from sqlalchemy.orm import selectinload from pydantic import BaseModel, Field from app.database.connection import get_db from app.models.session import Session from app.models.tool_call import ToolCall router = APIRouter() class ToolCallRequest(BaseModel): """Request schema for recording a tool call.""" session_id: Optional[str] = Field(None, description="Session ID (auto-detected if not provided)") tool_name: str = Field(..., description="Name of the tool being called") parameters: Optional[dict] = Field(None, description="Tool parameters as JSON object") result_status: Optional[str] = Field("success", description="Result status: success, error, timeout") error_message: Optional[str] = Field(None, description="Error message if failed") execution_time_ms: Optional[int] = Field(None, description="Execution time in milliseconds") timestamp: Optional[datetime] = Field(None, description="Timestamp of the tool call") class ToolCallResponse(BaseModel): """Response schema for tool call operations.""" id: int session_id: str tool_name: str result_status: str timestamp: datetime message: str class ToolUsageStats(BaseModel): """Statistics about tool usage.""" tool_name: str total_calls: int success_calls: int error_calls: int avg_execution_time_ms: Optional[float] success_rate: float async def get_current_session_id() -> Optional[str]: """Get the current session ID from the temporary file.""" try: with open("/tmp/claude-session-id", "r") as f: return f.read().strip() except (OSError, FileNotFoundError): return None @router.post("/tool-calls", response_model=ToolCallResponse, status_code=status.HTTP_201_CREATED) async def record_tool_call( request: ToolCallRequest, db: AsyncSession = Depends(get_db) ): """ Record a tool call made during a Claude Code session. This endpoint is called by Claude Code hooks when tools are used. It automatically detects the current session if not provided. """ try: # Get session ID session_id = request.session_id if not session_id: session_id = await get_current_session_id() if not session_id: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="No active session found. Please provide session_id or ensure a session is running." ) # Verify session exists result = await db.execute( select(Session).where(Session.id == int(session_id)) ) session = result.scalars().first() if not session: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Session {session_id} not found" ) # Create tool call record tool_call = ToolCall( session_id=str(session_id), tool_name=request.tool_name, parameters=json.dumps(request.parameters) if request.parameters else None, result_status=request.result_status or "success", error_message=request.error_message, execution_time_ms=request.execution_time_ms, timestamp=request.timestamp or datetime.utcnow() ) db.add(tool_call) await db.commit() await db.refresh(tool_call) return ToolCallResponse( id=tool_call.id, session_id=tool_call.session_id, tool_name=tool_call.tool_name, result_status=tool_call.result_status, timestamp=tool_call.timestamp, message=f"Tool call '{request.tool_name}' recorded successfully" ) except HTTPException: raise except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to record tool call: {str(e)}" ) @router.get("/tool-calls/session/{session_id}", response_model=List[dict]) async def get_session_tool_calls( session_id: str, db: AsyncSession = Depends(get_db) ): """Get all tool calls for a specific session.""" try: result = await db.execute( select(ToolCall) .where(ToolCall.session_id == session_id) .order_by(desc(ToolCall.timestamp)) ) tool_calls = result.scalars().all() return [ { "id": tc.id, "tool_name": tc.tool_name, "parameters": json.loads(tc.parameters) if tc.parameters else None, "result_status": tc.result_status, "error_message": tc.error_message, "execution_time_ms": tc.execution_time_ms, "timestamp": tc.timestamp } for tc in tool_calls ] except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to get tool calls: {str(e)}" ) @router.get("/tool-calls/stats", response_model=List[ToolUsageStats]) async def get_tool_usage_stats( project_id: Optional[int] = None, start_date: Optional[datetime] = None, end_date: Optional[datetime] = None, db: AsyncSession = Depends(get_db) ): """Get tool usage statistics with optional filtering.""" try: # Base query query = select( ToolCall.tool_name, func.count(ToolCall.id).label("total_calls"), func.sum(func.case([(ToolCall.result_status == "success", 1)], else_=0)).label("success_calls"), func.sum(func.case([(ToolCall.result_status == "error", 1)], else_=0)).label("error_calls"), func.avg(ToolCall.execution_time_ms).label("avg_execution_time_ms") ) # Apply filters if project_id: query = query.join(Session).where(Session.project_id == project_id) if start_date: query = query.where(ToolCall.timestamp >= start_date) if end_date: query = query.where(ToolCall.timestamp <= end_date) # Group by tool name query = query.group_by(ToolCall.tool_name).order_by(desc("total_calls")) result = await db.execute(query) rows = result.all() stats = [] for row in rows: success_rate = (row.success_calls / row.total_calls * 100) if row.total_calls > 0 else 0 stats.append(ToolUsageStats( tool_name=row.tool_name, total_calls=row.total_calls, success_calls=row.success_calls, error_calls=row.error_calls, avg_execution_time_ms=float(row.avg_execution_time_ms) if row.avg_execution_time_ms else None, success_rate=round(success_rate, 2) )) return stats except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to get tool usage stats: {str(e)}" ) @router.get("/tool-calls/popular", response_model=List[dict]) async def get_popular_tools( limit: int = 10, project_id: Optional[int] = None, db: AsyncSession = Depends(get_db) ): """Get the most frequently used tools.""" try: query = select( ToolCall.tool_name, func.count(ToolCall.id).label("usage_count"), func.max(ToolCall.timestamp).label("last_used") ) if project_id: query = query.join(Session).where(Session.project_id == project_id) query = query.group_by(ToolCall.tool_name).order_by(desc("usage_count")).limit(limit) result = await db.execute(query) rows = result.all() return [ { "tool_name": row.tool_name, "usage_count": row.usage_count, "last_used": row.last_used } for row in rows ] except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to get popular tools: {str(e)}" )