claude-code-tracker/app/api/tool_calls.py

"""
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)}"
        )