api_server v1.0.0

CallBridge

Autonomous outbound phone-negotiation agent that places calls on behalf of a human client, intercepts constraint violations in real time, stalls the receptionist with a natural holding phrase, escalates a Red Decision Card to the human, and speaks an approved counter-offer back — all through a FastAPI backend with WebSocket and REST interfaces.

Primary Purpose

CallBridge solves the problem of high-stakes, constraint-bound phone negotiations that humans find stressful or error-prone — such as booking medical appointments under time and budget limits, placing food orders with specific restrictions, or negotiating service terms. The system acts as a fully autonomous AI caller: it accepts a natural-language goal from the user, parses it into a structured rule-set (budgets, time windows, forbidden disclosures), and conducts the entire outbound conversation on the user's behalf. The real-world value is that it eliminates the cognitive load of live negotiation while guaranteeing that user-defined hard limits are never breached without explicit human sign-off.

Control Flow & Architecture

Requests enter through a FastAPI server (app.py) which maintains an in-memory session store (sessiondb). On session creation, the user's raw prompt is forwarded to engine.extract_rules_from_prompt(), which calls the Groq LLM with a structured system prompt and json_object response format to produce an ExtractedRules Pydantic model. During the call, each receptionist utterance arrives either as text via POST /evaluate-turn or as raw audio via POST /audio-turn — the audio path first passes through engine.transcribe_audio() (AssemblyAI primary, Groq Whisper fallback) before joining the same evaluation path. engine.evaluate_caller_turn() then builds a context window from the last 8 transcript turns plus the decision ledger and sends it to the LLM, which returns a TurnDecision. Depending on that decision, the session transitions to IN_PROGRESS (safe reply), DECISION_REQUIRED (dealbreaker), or COMPLETED (call confirmed). A parallel WebSocket endpoint (/ws/call/{session_id}) mirrors the full state machine for streaming frontends using named JSON events.

Invariants & Edge Cases

The most critical invariant is the dealbreaker interception: if the LLM sets is_dealbreaker=true, the session is immediately frozen at DECISION_REQUIRED and a stalling phrase is returned for the agent to speak aloud — the call cannot proceed until the human explicitly approves a counter-action via /user-action followed by /approve. An approved override is appended to the decision_ledger, and the LLM is explicitly instructed to treat ledger entries as authoritative — preventing previously approved alternatives from being re-flagged as violations. Audio transcription failures degrade gracefully: AssemblyAI errors are caught and logged, falling back to Groq Whisper; if both fail the endpoint returns a 500 with a structured error body rather than silently dropping the turn. TTS playback failures in tts_player.py are swallowed entirely (bare except pass) so a speaker driver issue never crashes the client. Empty or silent audio turns return a canned 'couldn't hear you' draft response without invoking the LLM, avoiding wasted inference on empty input.

CPU only ≥ 2 GB RAM CPU-only. Requires microphone and speakers for the interactive test client. Windows assumed for TTS playback via PowerShell MediaPlayer.

Core Technology Stack

Python >=3.10 FastAPI >=0.110.0 Uvicorn >=0.28.0 Pydantic >=2.6.0 OpenAI SDK (Groq) >=1.14.0 AssemblyAI >=0.23.0 Edge TTS >=6.1.10 SoundDevice + SciPy >=0.4.6 / >=1.12.0 python-dotenv >=1.0.0 websockets >=12.0

Entry Point

Primary File

app.py

Execution Type

asgi_server

Run Command

uvicorn app:app --reload --port 8000

Lifecycle Summary

On startup, FastAPI registers all routes and CORS middleware. A client POSTs to /api/session/create with a natural-language prompt; engine.extract_rules_from_prompt() calls the Groq LLM to produce an ExtractedRules object stored in sessiondb. During the call, each receptionist utterance is evaluated by engine.evaluate_caller_turn() which returns a TurnDecision. Dealbreakers freeze the session at DECISION_REQUIRED; the human picks a counter-action via /user-action, engine.generate_staged_draft() produces a bilingual response, and /approve resumes the call. At call end, /complete calls engine.generate_call_summary() for a structured audit report.

01

Interactive Architecture Flowchart

Click on any flowchart node to jump directly to its matching execution card

Clickable Nodes Active

graph TD
  subgraph Client["Client Entry Points"]
    A1["test_mic_caller.py\n(interactive CLI)"]
    A2["Frontend / Postman\n(HTTP + WebSocket)"]
  end

  subgraph API["app.py — FastAPI Server"]
    B1["POST /api/session/create"]
    B2["POST /{id}/evaluate-turn"]
    B3["POST /{id}/audio-turn"]
    B4["POST /{id}/user-action"]
    B5["POST /{id}/approve"]
    B6["POST /{id}/complete"]
    B7["WS /ws/call/{id}"]
  end

  subgraph Engine["engine.py — LLM Orchestration"]
    C1["extract_rules_from_prompt()\nGroq JSON mode"]
    C2["evaluate_caller_turn()\nGroq JSON mode"]
    C3["generate_staged_draft()\nGroq JSON mode"]
    C4["generate_call_summary()\nGroq JSON mode"]
    C5["transcribe_audio()\nAssemblyAI / Groq Whisper"]
  end

  subgraph TTS["tts_player.py"]
    E1["speak_text_aloud()\nedge_tts"]
    E2["play_audio_file()\nPowerShell"]
  end

  subgraph State["In-Memory State"]
    F1["sessiondb: Dict"]
  end

  A1 -->|HTTP POST| B1
  A2 -->|HTTP / WS| B1
  A2 --> B7
  B1 --> C1
  C1 --> F1
  B2 --> C2
  B3 --> C5
  C5 --> C2
  C2 --> F1
  B4 --> C3
  C3 --> F1
  B6 --> C4
  B7 --> C2
  B7 --> C3
  B7 --> C4
  A1 --> E1
  E1 --> E2

Scroll to zoom · Drag to pan

02

Step-by-Step Execution Pipeline

Lifecycle stages, module boundaries, IO contracts, and code anchors

Step 1app.py → create_session(request: IncomingUserRequest)

Session Creation & Rule Extraction

Validates the raw prompt is non-empty, generates a UUID session ID (8 chars), calls engine.extract_rules_from_prompt() to parse constraints via LLM, then stores the full session dict in sessiondb.

IN:JSON body: { raw_prompt: str, user_language: str = 'English', mode: str = 'delegate' | 'assist' }

OUT:JSON: { status, session_id, flags: {is_configured, status, mode}, extracted_rules: ExtractedRules }

</>Anchor: app.py:34-69PYTHON

rules = extract_rules_from_prompt(clean_prompt, request.user_language)

Step 2engine.py → extract_rules_from_prompt(prompt: str, user_language: str) -> ExtractedRules

LLM Goal Parsing

Builds a structured system prompt and calls client.chat.completions.create() with response_format={'type': 'json_object'} at temperature=0.1. Parses the returned JSON and validates through the ExtractedRules Pydantic model.

IN:Natural-language instruction string + user's preferred language

OUT:ExtractedRules: { intent, target_business, action_details, opening_phrase, constraints: List[RuleConstraint], forbidden_disclosures, special_notes }

</>Anchor: engine.py:35-69PYTHON

response = client.chat.completions.create(model=get_model(), messages=[...], response_format={"type": "json_object"}, temperature=0.1)

Step 3engine.py → evaluate_caller_turn(caller_text: str, session: dict) -> TurnDecision

Receptionist Turn Evaluation

Assembles a context-rich system prompt including the last 8 history turns, decision ledger of approved overrides, constraint rules, and ASSIST-mode addendum if active. Calls Groq at temperature=0.1 with json_object response format.

IN:Receptionist speech string + full session dict (rules, transcript_history, decision_ledger, mode)

OUT:TurnDecision: { is_dealbreaker, call_completed, violation_parameter, violation_reason, immediate_stalling_phrase, suggested_user_options: List[str], draft_response }

</>Anchor: engine.py:71-149PYTHON

raw_json = json.loads(response.choices[0].message.content)
return TurnDecision(**raw_json)

Step 4engine.py → transcribe_audio(audio_input, filename: str) -> str

Audio Transcription

Primary path: AssemblyAI Transcriber (if ASSEMBLYAI_API_KEY is set). Fallback: Groq Whisper via client.audio.transcriptions.create() using model='whisper-large-v3-turbo'. Result feeds into evaluate_caller_turn().

IN:bytes | file-like object | file path for WAV/MP3 audio; filename string for MIME hint

OUT:Transcribed plain-text string

</>Anchor: engine.py:229-259PYTHON

transcriber = aai.Transcriber()
transcript = transcriber.transcribe(audio_stream)

Step 5engine.py → generate_staged_draft(session: dict, action: UserDecisionAction) -> StagedResponse

Dealbreaker Escalation & Staged Draft

When TurnDecision.is_dealbreaker=True the session enters DECISION_REQUIRED. The human picks an option; generate_staged_draft() sends it to the LLM which produces a bilingual polite counter-offer. The english_response is stored as pending_draft and a ledger entry is appended.

IN:Session dict (rules, language) + UserDecisionAction { chosen_action, custom_instruction }

OUT:StagedResponse: { english_response, translated_response, action_summary }

</>Anchor: app.py:178-200PYTHON

staged = generate_staged_draft(session, action)
session["pending_draft"] = staged.english_response

Step 6tts_player.py → speak_text_aloud(text: str, voice: str)

Approval & TTS Playback

POST /approve clears pending_draft, appends agent turn to transcript_history, sets status to IN_PROGRESS, and returns speak_text. The test client calls speak_text_aloud() which synthesises with edge_tts.Communicate, saves a temp MP3, and plays it via a PowerShell subprocess blocking until playback completes.

IN:Session must be in AWAITING_APPROVAL state with a non-null pending_draft

OUT:REST: { status: 'APPROVED', speak_text }. Audio: MP3 synthesised via edge-tts and played via Windows PowerShell MediaPlayer.

</>Anchor: tts_player.py:44-63PYTHON

communicate = edge_tts.Communicate(text, voice)
await communicate.save(temp_file)
await asyncio.to_thread(play_audio_file, temp_file)

Step 7engine.py → generate_call_summary(session: dict) -> FinalSummary

Post-Call Audit Summary

Formats the complete conversation as a plain-text block and sends it with task constraints and ledger to the LLM at temperature=0.1. The model extracts confirmed agreements, open items, and produces an executive headline.

IN:Full session dict with transcript_history, decision_ledger, and rules

OUT:FinalSummary: { outcome_headline, outcome_subtext, confirmed_items: List[str], unresolved_items: List[str] }

</>Anchor: engine.py:191-227PYTHON

raw_json = json.loads(response.choices[0].message.content)
return FinalSummary(**raw_json)

Core Data Models

IncomingUserRequest models.py:4

Field	Type	Description
raw_prompt	str	Natural-language goal description from the human client.
user_language	Optional[str]	Preferred language for translated responses. Defaults to 'English'.
mode	Optional[str]	'delegate' — AI negotiates autonomously; 'assist' — pauses for human approval on every commitment.

ExtractedRules models.py:26

Field	Type	Description
intent	str	Primary action, e.g. 'schedule_appointment'.
target_business	str	Business being called, e.g. 'Dr. Smith Dental Clinic'.
action_details	Optional[str]	Specifics of the request.
opening_phrase	Optional[str]	Natural first sentence CallBridge speaks when the receptionist picks up.
constraints	List[RuleConstraint]	Parsed budget/time/quantity limits with parameter, operator, value, and unit.
forbidden_disclosures	List[str]	Information the agent must never share (credit card, SSN, etc.).
special_notes	Optional[str]	Additional freeform notes from the user's prompt.

TurnDecision models.py:52

Field	Type	Description
is_dealbreaker	bool	True when the receptionist violates a constraint or requests forbidden data.
call_completed	bool	True when the booking/order is confirmed or call ends.
violation_parameter	Optional[str]	The violated constraint name, e.g. 'price'.
violation_reason	Optional[str]	Human-readable Red Card reason.
immediate_stalling_phrase	Optional[str]	Phrase spoken immediately to keep receptionist waiting.
suggested_user_options	Optional[List[str]]	2-4 actionable choices surfaced on the Human-in-the-Loop UI card.
draft_response	Optional[str]	Next agent utterance (safe turn) or initial counter-offer text (dealbreaker).

StagedResponse models.py:97

Field	Type	Description
english_response	str	Exact English phrase agent will speak after approval.
translated_response	str	Same phrase in the user's preferred language for verification.
action_summary	str	One-line ledger entry, e.g. 'Counter-offer drafted for Sunday appointment'.

FinalSummary models.py:108

Field	Type	Description
outcome_headline	str	Bold executive headline of call result.
outcome_subtext	str	One-sentence logistics/payment status description.
confirmed_items	List[str]	Facts explicitly agreed upon during the conversation.
unresolved_items	List[str]	Open questions or ambiguities the call did not resolve.

UserDecisionAction models.py:88

Field	Type	Description
chosen_action	str	Option selected by the human from the Red Decision Card.
custom_instruction	Optional[str]	Free-text instruction if the user chose 'Type my own response'.

File System Directory

File	Role	LoC
./ All application source lives flat in the project root — no subdirectory layering.
app.py	FastAPI application — route handlers, session state machine, CORS middleware	345
engine.py	LLM orchestration — Groq client, all prompt-building and inference calls, AssemblyAI + Whisper transcription	260
models.py	Pydantic v2 data models for all request/response contracts	125
tts_player.py	Edge TTS synthesis and PowerShell-based audio playback	68
test_mic_caller.py	Interactive CLI test harness — mic recording, HTTP orchestration, human-in-the-loop decision prompts	261
list_models.py	Standalone debug script to enumerate available Groq models	24

Setup & Verification Playbook

Prerequisites

›Python >= 3.10
›pip
›Git
›Windows (TTS playback uses PowerShell MediaPlayer)
›Groq API key
›AssemblyAI API key (optional)

Step 1 Clone Repository

git clone https://github.com/Shariq-duet/CallBridge.git && cd CallBridge

Step 2 Create & Activate Virtual Environment

python -m venv venv && venv\Scripts\activate

Step 3 Install Dependencies

pip install --upgrade pip && pip install -r requirements.txt

Step 4 Configure Environment Variables

copy .env.example .env
# Edit .env: set GROQ_API_KEY, GROQ_MODEL, ASSEMBLYAI_API_KEY

Step 5 Start Backend Server

uvicorn app:app --reload --port 8000

Step 6 Verify API is Live

curl http://127.0.0.1:8000/api/models

Step 7 Run Interactive Test Client

python test_mic_caller.py

Step 8 Explore Swagger Docs

start http://127.0.0.1:8000/docs