from __future__ import annotations import asyncio import logging from pathlib import Path from dotenv import load_dotenv from fastapi import FastAPI, File, HTTPException, Query, UploadFile from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import StreamingResponse from agent.runtime import ADKRuntime from core.config import get_settings from core.db import SessionLocal, init_db from core.schemas import ChatRequest, ToolStatusResponse, UploadResponse from core.utils import ( build_c1_response, chunk_text_for_stream, extract_json_object, format_sse, looks_like_candidate_search, parse_action_command, wrap_text_as_c1, ) from mcp_server.server import mcp_app from services.candidate_service import CandidateService from services.ingestion_service import IngestionService from services.reference_data_service import ingest_reference_resumes, sync_reference_resumes_to_local from services.search_service import QdrantSearchService load_dotenv() logger = logging.getLogger(__name__) settings = get_settings() app = FastAPI(title=settings.app_name) app.mount("/mcp-server", mcp_app) app.add_middleware( CORSMiddleware, allow_origins=settings.cors_origin_list, allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) runtime = ADKRuntime() ingestion_service = IngestionService() search_service = QdrantSearchService() candidate_service = CandidateService() # Tracks the most recent resume uploads per session so the agent can answer # "tell me about the PDF" without the user pasting the candidate id back in. _session_recent_uploads: dict[str, dict[str, str]] = {} def _remember_upload(session_id: str | None, candidate_id: str, file_name: str) -> None: if not session_id: return _session_recent_uploads[session_id] = { "candidate_id": candidate_id, "file_name": file_name, } def _build_session_upload_hint(session_id: str) -> str: info = _session_recent_uploads.get(session_id) if not info: return "" return ( "\n\n[Recent upload context: the user just uploaded the resume " f"`{info['file_name']}` and it has been ingested as candidate id " f"`{info['candidate_id']}`. If their question refers to 'this pdf', " "'the resume', 'the file', or similar, it means this candidate. " "Call candidate_metadata_query with that candidate_id to fetch the " "stored summary, or use open_profile via the standard action.]" ) @app.on_event("startup") def startup_event() -> None: init_db() Path(settings.uploads_dir).mkdir(parents=True, exist_ok=True) @app.get("/health") def health() -> dict: return {"ok": True, "service": settings.app_name} @app.get("/mcp-health") def mcp_health() -> dict: return { "ok": True, "mcp_endpoint": settings.mcp_server_base_url, "tools": [ "ingest_resume_pdf", "semantic_candidate_search", "candidate_metadata_query", "compute_job_match_score", "get_policy_info", "manage_application_status", "manage_interview_records", "generate_job_posting", "bulk_ingest_reference_resumes", "email_compose", ], } @app.get("/api/tooling/status", response_model=ToolStatusResponse) def tooling_status() -> ToolStatusResponse: return ToolStatusResponse( gemini_configured=bool(settings.gemini_api_key), qdrant_configured=settings.has_qdrant_config, r2_configured=settings.has_s3_config, mcp_url=settings.mcp_server_base_url, ) @app.get("/api/diag/email") def diag_email() -> dict: """Diagnostic: which email transport will the runtime try?""" if settings.brevo_api_key: primary = "brevo" elif settings.resend_api_key: primary = "resend" elif settings.has_smtp_config and settings.smtp_from: primary = "smtp" else: primary = "none" return { "brevo_api_key_present": bool(settings.brevo_api_key), "resend_api_key_present": bool(settings.resend_api_key), "smtp_host_present": bool(settings.smtp_host), "smtp_from_present": bool(settings.smtp_from), "smtp_from": settings.smtp_from, "transport_will_use": primary, } @app.post("/api/upload", response_model=UploadResponse) async def upload_resume( file: UploadFile = File(...), session_id: str | None = Query(default=None), ) -> UploadResponse: file_name = file.filename or "resume.pdf" if not file_name.lower().endswith(".pdf"): raise HTTPException(status_code=400, detail="Only PDF files are supported in this endpoint.") file_bytes = await file.read() if not file_bytes: raise HTTPException(status_code=400, detail="Uploaded file is empty.") save_path = Path(settings.uploads_dir) / file_name save_path.write_bytes(file_bytes) with SessionLocal() as db: result = ingestion_service.ingest_pdf( db, file_name=file_name, file_bytes=file_bytes, ) _remember_upload(session_id, result.candidate_id, result.file_name) return UploadResponse( candidate_id=result.candidate_id, file_name=result.file_name, chunks_indexed=result.chunks_indexed, r2_url=result.r2_url, message="Resume ingested successfully.", ) @app.post("/api/bootstrap/reference-resumes") def bootstrap_reference_resumes( limit: int = Query(default=20, ge=1, le=200), source_dir: str | None = Query(default=None), force_reingest: bool = Query(default=False), ) -> dict: summary = ingest_reference_resumes( limit=limit, source_dir=source_dir, force_reingest=force_reingest, ) return { "status": "success", "source_dir": summary.source_dir, "requested_limit": summary.requested_limit, "files_seen": summary.files_seen, "ingested": summary.ingested, "skipped": summary.skipped, "failed": summary.failed, "failures": summary.failures, } @app.post("/api/bootstrap/sync-reference-resumes") def sync_reference_resumes( max_files: int = Query(default=120, ge=1, le=1000), source_dir: str | None = Query(default=None), ) -> dict: result = sync_reference_resumes_to_local(max_files=max_files, source_dir=source_dir) return { "status": "success", **result, } def _collect_tool_ui(tool_calls: list[dict]) -> dict: """Merge UI payloads from MCP tool responses (cards, summary, chart, markdown, tabs).""" merged: dict = {} cards: list = [] for call in tool_calls: response = call.get("response") or {} ui = response.get("ui") if isinstance(response, dict) else None if not isinstance(ui, dict): continue for key in ("summary", "c1_response", "chart", "table", "markdown", "tabs"): value = ui.get(key) if value and not merged.get(key): merged[key] = value ui_cards = ui.get("cards") if isinstance(ui_cards, list): cards.extend(ui_cards) if cards: merged["cards"] = cards return merged def _handle_action_command(command: str, target_id: str) -> tuple[str, dict | None]: """Resolve card actions like ``open_profile:`` without invoking the LLM.""" if command == "open_profile" or command == "details": with SessionLocal() as db: candidate = candidate_service.get_candidate(db, target_id) if candidate is None: return ( f"I could not find a candidate with id `{target_id}`.", None, ) payload_json = candidate_service.build_profile_payload(candidate) return payload_json["summary"], payload_json if command == "shortlist": with SessionLocal() as db: candidate = candidate_service.get_candidate(db, target_id) if candidate is None: return f"Could not shortlist: candidate `{target_id}` not found.", None payload_json = candidate_service.build_profile_payload(candidate) message = ( f"Shortlisted **{candidate.full_name or candidate.external_id}** " f"(id `{candidate.external_id}`). Track them in the application stages tool." ) payload_json["summary"] = message return message, payload_json if command in {"publish_job_posting", "edit_job_posting"}: verb = "publish" if command == "publish_job_posting" else "edit" return ( f"Job posting `{target_id}` queued to {verb}. " "Open the job postings file under backend/data to confirm.", None, ) return "", None @app.post("/api/chat") async def chat_stream(payload: ChatRequest) -> StreamingResponse: async def event_generator(): try: with SessionLocal() as db: candidate_service.audit_query( db, session_id=payload.session_id, query=payload.message, top_k=payload.top_k, ) # If the user's message contains a demonstrative pointer to the most recent # upload ("this pdf", "the candidate", "their resume", etc.) and we have a # recorded upload for this session, short-circuit to the profile lookup so # the agent never has to guess at the candidate id. action = parse_action_command(payload.message) if action is None: lowered = payload.message.lower() pdf_phrases = ( "this pdf", "the pdf", "this resume", "the resume", "this file", "the file", "uploaded pdf", "uploaded resume", "uploaded file", "this candidate", "the candidate", "this person", "the person", "this applicant", "the applicant", "this profile", "the profile", "their resume", "their profile", "their experience", "his resume", "her resume", "his profile", "her profile", "about him", "about her", "about them", ) recent = _session_recent_uploads.get(payload.session_id) if payload.session_id else None if recent and any(phrase in lowered for phrase in pdf_phrases): action = ("open_profile", recent["candidate_id"]) if action is not None: command, target_id = action yield format_sse("status", {"message": f"Resolving {command} action"}) final_text, action_payload = _handle_action_command(command, target_id) for token in chunk_text_for_stream(final_text): yield format_sse("token", {"delta": token}) if action_payload is None: action_payload = {} # Drop any c1_response — the Thesys SDK rejects hand-rolled XML and # shows a red 'Error while generating response' message. The card + # markdown pipeline already renders the response correctly. action_payload.pop("c1_response", None) action_payload.pop("c1Response", None) if not action_payload.get("summary"): action_payload["summary"] = final_text yield format_sse("genui", action_payload) yield format_sse("done", {"message": final_text}) return yield format_sse("status", {"message": "Running ADK recruitment agent"}) agent_message = payload.message upload_hint = _build_session_upload_hint(payload.session_id) if upload_hint: agent_message = payload.message + upload_hint turn = await runtime.run_turn_detailed( message=agent_message, user_id=payload.user_id, session_id=payload.session_id, ) final_text = turn.text or "I could not produce a response. Please retry with more detail." for token in chunk_text_for_stream(final_text): yield format_sse("token", {"delta": token}) payload_json = extract_json_object(final_text) # Promote MCP tool UI payloads (e.g. job posting, interview invite) so the # frontend can render rich Thesys C1 / cards even when the agent only emits # plain markdown text. tool_ui = _collect_tool_ui(turn.tool_calls) if tool_ui: if payload_json is None: payload_json = {} for key, value in tool_ui.items(): if value and not payload_json.get(key): payload_json[key] = value # Run the candidate-search fallback when the user's message clearly asks for # a search AND we don't already have cards. The semantic search tool always # returns a `ui` block (even with empty cards), so we can't just check # `payload_json is None` — we have to look at whether actual cards were # produced. existing_cards = (payload_json or {}).get("cards") if isinstance(payload_json, dict) else None cards_empty = not existing_cards if cards_empty and looks_like_candidate_search(payload.message): fallback_payload: dict | None = None rows = search_service.semantic_search(query=payload.message, top_k=payload.top_k) if rows: fallback_payload = search_service.build_genui_payload(payload.message, rows) else: with SessionLocal() as db: local_matches = candidate_service.search_candidates( db, query=payload.message, limit=payload.top_k, ) if local_matches: fallback_payload = candidate_service.build_genui_payload( payload.message, local_matches, ) if fallback_payload: if payload_json is None: payload_json = fallback_payload else: for key, value in fallback_payload.items(): # Overwrite anything from the agent's empty tool response. if value: payload_json[key] = value # Replace the agent's "no results" final text with our fallback summary # so the bubble matches the cards we surface. fb_summary = fallback_payload.get("summary") if isinstance(fb_summary, str) and fb_summary.strip(): final_text = fb_summary if payload_json is None: payload_json = {} # Strip any c1_response — Thesys C1Component shows 'Error while generating # response' when handed XML it didn't generate itself. The markdown + cards # + chart pipeline below already provides the interactive UI. payload_json.pop("c1_response", None) payload_json.pop("c1Response", None) if payload_json and not payload_json.get("summary"): payload_json["summary"] = final_text yield format_sse("genui", payload_json) done_message = final_text summary = payload_json.get("summary") if isinstance(payload_json, dict) else None if isinstance(summary, str) and summary.strip(): done_message = summary.strip() yield format_sse("done", {"message": done_message}) except asyncio.CancelledError as exc: # pragma: no cover logger.exception("Chat stream cancelled") yield format_sse("error", {"message": f"Agent run cancelled: {exc}"}) except Exception as exc: # pragma: no cover logger.exception("Chat stream failed") yield format_sse("error", {"message": str(exc)}) return StreamingResponse(event_generator(), media_type="text/event-stream") @app.get("/api/candidates/{candidate_id}") def get_candidate_detail(candidate_id: str) -> dict: with SessionLocal() as db: candidate = candidate_service.get_candidate(db, candidate_id) if candidate is None: raise HTTPException(status_code=404, detail="Candidate not found.") payload_json = candidate_service.build_profile_payload(candidate) return { "status": "success", "candidate_id": candidate.external_id, "profile": payload_json, } if __name__ == "__main__": import uvicorn uvicorn.run("main:app", host=settings.host, port=settings.port, reload=False)