"""Claude Code Integration API. Provides three integration surfaces: 1. Session Tracking — GET /claude-code/sessions, /sessions/{id}, /projects Reads ~/.claude/projects/**/*.jsonl directly. Extracts token usage, model, cost estimates, and active status without requiring a gateway. 2. Config Scanner — GET /claude-code/config Reads ~/.claude/settings.json, ~/.codex/config.toml, and ~/.codex/rules/. 3. Direct CLI info — GET /claude-code/cli Reports which CLI tools are detected on the host and their versions. (Actual CLI execution goes through the gateway; this endpoint surfaces discovery metadata for the dashboard.) """ from __future__ import annotations import asyncio import shutil from typing import Any from fastapi import APIRouter, Depends, HTTPException, Query, status from app.api.deps import require_org_member from app.core.logging import get_logger from app.schemas.claude_code import ( ClaudeConfigRead, ClaudeProjectRead, ClaudeSessionListResponse, ClaudeSessionRead, ClaudeSessionStatsRead, SessionTokensRead as ClaudeSessionTokensRead, ) from app.services import claude_code_reader as reader from app.services.organizations import OrganizationContext logger = get_logger(__name__) router = APIRouter(prefix="/claude-code", tags=["claude-code"]) ORG_MEMBER_DEP = Depends(require_org_member) def _session_to_read(s: reader.ClaudeSession) -> ClaudeSessionRead: return ClaudeSessionRead( session_id=s.session_id, project_dir=s.project_dir, cwd=s.cwd, title=s.title, models=s.models, tokens=ClaudeSessionTokensRead( input=s.tokens.input, output=s.tokens.output, cache_read=s.tokens.cache_read, cache_write=s.tokens.cache_write, total=s.tokens.total, ), cost_usd=s.cost_usd, message_count=s.message_count, first_message_at=s.first_message_at, last_message_at=s.last_message_at, is_active=s.is_active, entrypoints=s.entrypoints, git_branch=s.git_branch, version=s.version, ) # ── Session Tracking ────────────────────────────────────────────────────────── @router.get( "/sessions", response_model=ClaudeSessionListResponse, summary="List local Claude Code sessions", description=( "Auto-discovers sessions from `~/.claude/projects/**/*.jsonl`. " "Returns token usage, model info, cost estimates, and active status " "for each session without requiring a gateway connection." ), ) async def list_sessions( project: str | None = Query(None, description="Filter by project directory name substring"), active_only: bool = Query(False, description="Return only currently active sessions"), limit: int = Query(100, ge=1, le=500, description="Maximum sessions to return"), ctx: OrganizationContext = ORG_MEMBER_DEP, ) -> ClaudeSessionListResponse: """List Claude Code sessions from local JSONL files.""" sessions = await asyncio.to_thread( reader.list_sessions, project_filter=project, active_only=active_only, limit=limit, ) stats = reader.session_stats(sessions) return ClaudeSessionListResponse( sessions=[_session_to_read(s) for s in sessions], total=len(sessions), stats=ClaudeSessionStatsRead( session_count=stats["session_count"], active_sessions=stats["active_sessions"], total_tokens=stats["total_tokens"], total_cost_usd=stats["total_cost_usd"], models=stats["models"], ), ) @router.get( "/sessions/{session_id}", response_model=ClaudeSessionRead, summary="Get a single Claude Code session", ) async def get_session( session_id: str, ctx: OrganizationContext = ORG_MEMBER_DEP, ) -> ClaudeSessionRead: """Return a single parsed session by its UUID.""" session = await asyncio.to_thread(reader.get_session, session_id) if session is None: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Session not found") return _session_to_read(session) @router.get( "/projects", response_model=list[ClaudeProjectRead], summary="List discovered local Claude Code projects", description="Returns one entry per project directory with aggregated token and cost stats.", ) async def list_projects( ctx: OrganizationContext = ORG_MEMBER_DEP, ) -> list[ClaudeProjectRead]: """Aggregate session stats grouped by project directory.""" projects = await asyncio.to_thread(reader.list_projects) return [ ClaudeProjectRead( project_dir=p["project_dir"], cwd=p.get("cwd"), session_count=p["session_count"], total_tokens=p["total_tokens"], total_cost_usd=p["total_cost_usd"], last_active_at=p.get("last_active_at"), is_active=p["is_active"], ) for p in projects ] # ── Config Scanner ──────────────────────────────────────────────────────────── @router.get( "/config", response_model=ClaudeConfigRead, summary="Read local Claude Code and Codex CLI configuration", description=( "Reads `~/.claude/settings.json`, `~/.codex/config.toml`, and " "`~/.codex/rules/*.rules`. Never returns secrets — only settings " "and trust/rule data." ), ) async def get_config( ctx: OrganizationContext = ORG_MEMBER_DEP, ) -> ClaudeConfigRead: """Return parsed local CLI configuration.""" cfg = await asyncio.to_thread(reader.read_config) return ClaudeConfigRead( claude_settings=cfg.claude_settings, codex_config=cfg.codex_config, codex_rules=cfg.codex_rules, claude_credentials_configured=cfg.claude_credentials_path is not None, codex_credentials_configured=cfg.codex_credentials_path is not None, ) # ── Direct CLI Discovery ────────────────────────────────────────────────────── async def _cli_version(cmd: list[str]) -> str | None: try: proc = await asyncio.create_subprocess_exec( *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) stdout, _ = await asyncio.wait_for(proc.communicate(), timeout=5) line = stdout.decode(errors="replace").strip().splitlines()[0] if stdout else "" return line or None except Exception: return None @router.get( "/cli", summary="Detect installed CLI tools", description=( "Checks which of Claude Code, Codex CLI, and Ollama are installed " "on the host and returns their versions. No gateway required." ), ) async def detect_cli( ctx: OrganizationContext = ORG_MEMBER_DEP, ) -> dict[str, Any]: """Return discovery metadata for locally installed CLI tools.""" claude_path = shutil.which("claude") codex_path = shutil.which("codex") ollama_path = shutil.which("ollama") claude_version = await _cli_version(["claude", "--version"]) if claude_path else None codex_version = await _cli_version(["codex", "--version"]) if codex_path else None ollama_version = await _cli_version(["ollama", "--version"]) if ollama_path else None return { "claude": { "installed": claude_path is not None, "path": claude_path, "version": claude_version, "projects_dir": str(reader._projects_dir()), "projects_dir_exists": reader._projects_dir().exists(), }, "codex": { "installed": codex_path is not None, "path": codex_path, "version": codex_version, }, "ollama": { "installed": ollama_path is not None, "path": ollama_path, "version": ollama_version, }, }