Timed voice interviews in the browser

{
  "name": "Screening interviewer",
  "system_prompt": "# Role\nYou are Asha, a friendly screening interviewer for {{company_name}}. …",
  "first_message": "Hi {{candidate_name}}, thanks for joining. This screening takes about ten minutes. Shall we start?",
  "voice": { "voice_id": "neha" },
  "language": "en-IN",
  "max_duration_secs": 600,
  "end_call": {
    "enabled": true,
    "message": "Thank you for your time today. The hiring team will be in touch. Goodbye!",
    "confirm": true
  },
  "analysis": {
    "enabled": true,
    "prompt": "Score the candidate from the transcript only. communication: clarity and structure of answers. role_fit: relevant experience for the role discussed in the interview. Use 1 (weak) to 5 (strong). List at most three strengths and three concerns, each quoting or paraphrasing what the candidate said.",
    "success_criteria": "The candidate answered at least three of the four interview areas.",
    "output_schema": {
      "type": "object",
      "properties": {
        "communication": { "type": "integer", "minimum": 1, "maximum": 5 },
        "role_fit": { "type": "integer", "minimum": 1, "maximum": 5 },
        "strengths": { "type": "array", "items": { "type": "string" }, "maxItems": 3 },
        "concerns": { "type": "array", "items": { "type": "string" }, "maxItems": 3 },
        "recommendation": { "type": "string", "enum": ["advance", "human_review"] }
      },
      "required": ["communication", "role_fit", "recommendation"]
    }
  }
}

# pip install fastapi uvicorn httpx
import os
import uuid

import httpx
from fastapi import FastAPI, HTTPException, Request

API = "https://sandbox.voice.miraiminds.co"
AGENT_ID = os.environ["MIRAI_AGENT_ID"]
WEBHOOK_URL = os.environ["PUBLIC_WEBHOOK_URL"]  # https://your-app.example/mirai/webhook

app = FastAPI()
mirai = httpx.AsyncClient(
    base_url=API,
    headers={"Authorization": f"Bearer {os.environ['MIRAI_API_KEY']}"},
    timeout=30,
)

# Stand-ins for your own code: sign-in and storage.
from myapp import current_candidate, db


@app.post("/api/interviews")
async def start_interview(request: Request):
    candidate = await current_candidate(request)
    interview = await db.interviews.create(candidate_id=candidate.id, role=candidate.role)

    r = await mirai.post(
        "/v2/calls",
        headers={"Idempotency-Key": f"interview-{interview.id}"},
        json={
            "agent_id": AGENT_ID,
            "channel": "web",
            "variables": {
                "candidate_name": candidate.first_name,
                "role_title": candidate.role,
                "company_name": "Acme",
                "focus_areas": "Their experience with customer escalations "
                               "and how they prioritise a busy queue.",
            },
            "first_message": (
                f"Hi {candidate.first_name}, thanks for making time for your "
                f"{candidate.role} screening. It takes about ten minutes. "
                "Shall we begin?"
            ),
            "metadata": {"interview_id": interview.id, "candidate_id": candidate.id},
            "max_duration_secs": 600,
            "timing": {
                "wrap_up_secs_before_end": 120,
                "wrap_up_instruction": "Time is nearly up. Do not start a new area. "
                                       "Let the candidate finish, then close.",
                "close_secs_before_end": 30,
                "close_message": "That's all the time we have. Thank you, and the "
                                 "hiring team will be in touch. Goodbye!",
                "user_silence_close_secs": 60,
                "user_silence_message": "It seems you've stepped away, so I'll end "
                                        "the interview here. Goodbye!",
            },
            "recording_enabled": False,
            "final_results": True,
            "webhook_url": WEBHOOK_URL,
        },
    )
    call = r.json()
    if r.status_code != 202:
        raise HTTPException(502, call["error"]["code"])

    await db.interviews.update(interview.id, mirai_call_id=call["id"], status="started")
    # The page gets its two links and your own ID. Never the API key.
    return {
        "interview_id": interview.id,
        "ws_url": call["ws_url"],
        "events_url": call["events_url"],
    }

from myapp import current_moderator  # stand-in: only your staff may steer


async def control(interview_id: str, body: dict, action_key: str) -> dict:
    interview = await db.interviews.get(interview_id)
    r = await mirai.post(
        f"/v2/calls/{interview.mirai_call_id}/control",
        headers={"Idempotency-Key": f"{interview_id}-{action_key}"},
        json=body,
    )
    out = r.json()
    if r.status_code == 409:  # call_not_active: not connected yet, or already over
        return {"status": "not_active"}
    if r.status_code not in (200, 202):
        raise HTTPException(502, out["error"]["code"])
    return {"status": out["status"]}  # applied | pending | rejected


@app.post("/api/interviews/{interview_id}/steer")
async def steer(interview_id: str, request: Request):
    await current_moderator(request)
    text = (await request.json())["text"][:1000]
    return await control(interview_id, {"type": "instruction", "text": text},
                         action_key=f"steer-{uuid.uuid4().hex}")


@app.post("/api/interviews/{interview_id}/end")
async def end(interview_id: str, request: Request):
    await current_moderator(request)
    return await control(
        interview_id,
        {"type": "close", "message": "Thank you for your time today. Goodbye!"},
        action_key="close",  # the same key every time: a double click closes once
    )

import json

from fastapi import Response

from myapp import queue, verify_signature  # verify_signature: see the Webhooks page

WEBHOOK_SECRET = os.environ["MIRAI_WEBHOOK_SECRET"]  # whsec_…


@app.post("/mirai/webhook")
async def mirai_webhook(request: Request):
    raw = await request.body()
    if not verify_signature(raw, request.headers.get("X-Mirai-Signature", ""), WEBHOOK_SECRET):
        return Response(status_code=401)
    event = json.loads(raw)
    if not await db.events.first_time(event["id"]):  # retries keep the same id
        return Response(status_code=200)

    call = event["data"]["call"]
    interview_id = (call.get("metadata") or {}).get("interview_id")

    if event["type"] in ("call.completed", "call.failed", "call.aborted"):
        await db.interviews.update(interview_id, status="ended",
                                   ended_reason=call["ended_reason"])
    elif event["type"] == "call.processed":
        # Answer fast; do the work in the background.
        await queue.enqueue("store_interview_result", interview_id=interview_id,
                            call_id=call["id"], event=event)
    return Response(status_code=200)


async def store_interview_result(interview_id: str, call_id: str, event: dict):
    transcript = event["data"].get("transcript") or {}
    status = transcript.get("status")
    if status == "fetch_required":  # longer than the inline limit
        r = await mirai.get(f"/v2/calls/{call_id}/transcript")
        turns, status = r.json()["turns"], "ready"
    else:
        turns = transcript.get("turns") or []

    analysis = event["data"]["call"].get("analysis") or {}
    await db.interviews.update(
        interview_id,
        transcript_status=status,  # ready | empty | failed
        transcript=turns,          # [{role, text, start_ms, end_ms}]
        summary=analysis.get("summary"),
        usable=analysis.get("success"),
        scores=analysis.get("data"),  # the shape of your output_schema
    )