# Project export: SuperDial

This document was generated by HackStack to give an AI agent context about a hackathon project. Sections are labeled with their provenance; content marked as truncated was cut to keep this document small.

## Project metadata

- Hackathon: Cal Hacks 12.0
- Tagline: SuperDial — your AI voice assistant that becomes your personal caller, handling conversations, bookings, and tasks in your own voice while you focus on what matters.
- Devpost: https://devpost.com/software/dialsense
- GitHub: https://github.com/dliu99/pokesense
- Video: https://www.youtube.com/embed/dpzwmZqkNhc?enablejsapi=1&hl=en_US&rel=0&start=&version=3&wmode=transparent
- Team: 1 GitHub contributor(s) — dliu99 (11 commits)

## Devpost submission (written by the team)

No Devpost description available.

## README (from the GitHub repository)

# MCP Server Template

A minimal [FastMCP](https://github.com/jlowin/fastmcp) server template for Render deployment with streamable HTTP transport.

[![Deploy to Render](https://render.com/images/deploy-to-render-button.svg)](https://render.com/deploy?repo=https://github.com/InteractionCo/mcp-server-template)

## Local Development

### Setup

Fork the repo, then run:

```bash
git clone <your-repo-url>
cd mcp-server-template
uv venv --python 3.13 mcp-server
.\mcp-server\Scripts\activate
uv pip install -r requirements.txt
```

### Test

```bash
python src/server.py
# then in another terminal run:
npx @modelcontextprotocol/inspector
```

Open http://localhost:3000 and connect to `http://localhost:8000/mcp` using "Streamable HTTP" transport (NOTE THE `/mcp`!).

## Deployment

### Option 1: One-Click Deploy
Click the "Deploy to Render" button above.

### Option 2: Manual Deployment
1. Fork this repository
2. Connect your GitHub account to Render
3. Create a new Web Service on Render
4. Connect your forked repository
5. Render will automatically detect the `render.yaml` configuration

Your server will be available at `https://your-service-name.onrender.com/mcp` (NOTE THE `/mcp`!)

## Poke Setup

You can connect your MCP server to Poke at (poke.com/settings/connections)[poke.com/settings/connections].
To test the connection explitly, ask poke somethink like `Tell the subagent to use the "{connection name}" integration's "{tool name}" tool`.
If you run into persistent issues of poke not calling the right MCP (e.g. after you've renamed the connection) you may send `clearhistory` to poke to delete all message history and start fresh.
We're working hard on improving the integration use of Poke :)


## Customization

Add more tools by decorating functions with `@mcp.tool`:

```python
@mcp.tool
def calculate(x: float, y: float, operation: str) -> float:
    """Perform basic arithmetic operations."""
    if operation == "add":
        return x + y
    elif operation == "multiply":
        return x * y
    # ...
```


## Detected evidence (automated analysis)

Indexed codebase: 5 recognized source files, 18 KB.
- Flask (technology) — detected in the code
- Python (language) — detected in the code

## Codebase structure (from repository index)

### Files (9 of 9)

```
.gitignore
README.md
render.yaml
requirements.txt
src/server.py
src/test.py
src/tts_server.py
test_output_long.pcm
test_output_valid.pcm
```

### Dependencies

- requirements.txt: fastmcp@>=2.12.0, fish-audio-sdk, flask@>=3.0.0, google-genai, python-dotenv, requests, uvicorn@>=0.35.0, vapi_server_sdk

### Recent commits (newest first)

- fix: webhook handling
- fixed webhook handling in tts-server
- updated phone id
- idc committing envs
- added ngrok url
- removed bd tool
- search doesnt work in prod?
- call flow improvements, websockets
- comment
- feat: added bright-data tool
- ??
- prompt
- feat: vapi implementation
- api keys
- [readme] add clearhistory docs
- [server] fix 400s, add prompt guidance
- [docs] highlight need for /mcp
- [Example Server] Initial commit

## Key source files (fetched from GitHub, selected and truncated for size)

### requirements.txt

```
fastmcp>=2.12.0
uvicorn>=0.35.0
vapi_server_sdk
google-genai
python-dotenv
requests
fish-audio-sdk
flask>=3.0.0
```

### src/server.py

```python
from fastmcp.client.transports import NodeStdioTransport, PythonStdioTransport, SSETransport, StreamableHttpTransport
import os
import random
import fastmcp
from fastmcp import FastMCP
from dotenv import load_dotenv
import requests
from vapi import Vapi
import time
import json
from google import genai
from google.genai import types

load_dotenv()
POKE_API_KEY = os.getenv('POKE_API_KEY')
BRIGHT_DATA_API_KEY = os.getenv('BRIGHT_DATA_API_KEY')
GEMINI_API_KEY = os.getenv('GEMINI_API_KEY')
BRIGHT_DATA_MCP_URL = os.getenv('BRIGHT_DATA_MCP_URL')
ass_id = "7341414f-3916-49a6-9501-69de1d5690c7"
#"c28fcf1f-6496-407e-a142-928acc714892"#"554cfbbc-f0d0-4e1b-aa88-b460ede9c553"
phone_id = "79421ad1-ee29-4dc1-a6f2-cd83a922486f"
TTS_SERVER_URL = os.getenv("TTS_SERVER_URL", "https://f2ed034e2ba5.ngrok-free.app")

vapi_client = Vapi(token="513d7c4b-27d5-4eb0-9c3d-4c61a2bcf647")
mcp = FastMCP("superdial mcp")
client = genai.Client(
    api_key=GEMINI_API_KEY,
)

def get_call_status_from_webhook(call_id: str, timeout_seconds: int = 400):
    """
    Poll the webhook status endpoint for call updates.
    This queries the tts_server for call status received via webhook.
    """
    start_time = time.time()
    poll_interval = 1  # Check every 1 second
    
    while time.time() - start_time < timeout_seconds:
        try:
            response = requests.get(f"{TTS_SERVER_URL}/api/calls/{call_id}/status", timeout=10)
            if response.status_code == 200:
                data = response.json()
                status = data.get("status")
                print(f"Webhook status check: {status}")
                
                if status == "ended":
                    print(f"Call {call_id} ended via webhook")
                    return {
                        "status": status,
                        "analysis": data.get("analysis"),
                        "result": data.get("result"),
                        "updated_at": data.get("updated_at")
                    }
            elif response.status_code == 404:
                print(f"Call {call_id} not yet recorded by webhook")
        except Exception as e:
            print(f"Error checking webhook status: {e}")
        
        time.sleep(poll_interval)
    
    # Timeout - return last known status from Vapi
    print(f"Webhook status check timeout after {timeout_seconds}s")
    return None
# cut b/c issues in prod?
'''
@mcp.tool(description="Search the web for information. Use this to get names & phone numbers of businesses to reserve or book something. Example: 'search_web(query='barbers in sf near palace of fine arts')'")
async def search_web(query: str) -> dict:
    # bright data mcp works, but not bright data api? sorry for this convoluted setup
    
    try:
        remote_client = fastmcp.Client(BRIGHT_DATA_MCP_URL)
        async with remote_client:

            result = await remote_client.call_tool(
                "search_engine",
                {
                    "query": query,
                    "engine": "google",  #bing or yandex
                }
            )
            
            search_results = result.content if hasattr(result, 'content') else result
            print(f"Search results for '{query}':")
            print(search_results)
            
            return {
                "status": "success",
                "data": search_results,
                "query": query,
            }
    except Exception as exc:
        print(f"Failed to search using Bright Data MCP server: {exc}")
        return {
            "status": "error",
            "error": str(exc),
        }
'''
@mcp.tool(description="Make a call to a phone number (format: +1XXXXXXXXXX). Provide background information about your intentions (i.e. to book a haircut at 4pm) to the voice assistant.")
def make_call(phone_number: str, target_name: str, my_name: str, call_info_notes_for_agent: str) -> dict:
    #generate first msg w/ gemini flash
    model = "gemini-flash-lite-latest"
    contents = [
        types.Content(
            role="user",
            parts=[
                types.Part.from_text(text=
                f"""Write a two sentence first message for a voice assistant in the format of "{random.choice(['What\'s good', 'Yooo', 'Sup bro'])} {target_name}! I'm {my_name}, calling {target_name} to (REASON)."
Use the following to complete the (REASON) blank:
Call information notes: {call_info_notes_for_agent}
Target name: {target_name}
My name: {my_name}"""),
            ],
        ),
    ]
    generate_content_config = types.GenerateContentConfig(
        thinking_config = types.ThinkingConfig(
            thinking_budget=0,
        ),
        response_mime_type="application/json",
        response_schema=genai.types.Schema(
            type = genai.types.Type.OBJECT,
            required = ["message"],
            properties = {
                "message": genai.types.Schema(
                    type = genai.types.Type.STRING,
                ),
            },
        ),
    )
    first_msg = ""
    for chunk in client.models.generate_content_stream(
        model=model,
        contents=contents,
        config=generate_content_config,
    ):
        first_msg += chunk.text
    
    print("getting")
    ass = vapi_client.assistants.get(id=ass_id)

    print("\nASKDJSKJD\n")
    try:
        res = vapi_client.calls.create(
            phone_number_id=phone_id,
            customer={"number": "+15109497606"}, #change to phone number in prod
            assistant_id=ass_id,
            assistant_overrides={
                "firstMessage": first_msg,
                "variableValues": {
                    "target_name": target_name,
                    "my_name": my_name,
                }
            }
        )
        print(f"Call created: {res.status}")
        call_id = res.id

        if call_info_notes_for_agent:
            try:
                vapi_client.calls.update(
                    call_id,
                    {"messages": [{"role": "syst
[truncated — 1507 more characters]
```

### render.yaml

```yaml
services:
  - type: web
    name: fastmcp-server
    runtime: python
    buildCommand: pip install -r requirements.txt
    startCommand: python src/server.py
    plan: free
    autoDeploy: false
    envVars:
      - key: ENVIRONMENT
        value: production

```

### src/test.py

```python
from fastmcp import Client
from server import mcp
import asyncio
import os
import sys
import uuid

# Add current directory to path to import tts_server module
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from tts_server import synthesize_audio


async def test():
    client = Client(mcp)


    async with client:
        result1 = await client.call_tool("search_web", {
            "query": "barbers in sf near palace of fine arts"
        })
        print(result1)

        '''result = await client.call_tool("make_call", {
            "phone_number": "+15109497606",
            "name": "Devin",
            "call_info_notes_for_agent": "Help Devin book a haircut (low taper fade) for tomorrow at noon"
        })
        print(result)'''
        
       

if __name__ == "__main__":
    asyncio.run(test())
```

### src/tts_server.py

```python
from __future__ import annotations

import os
import time
import uuid
from threading import Lock
from typing import Optional
import json

from dotenv import load_dotenv
from flask import Flask, Response, jsonify, request
from fish_audio_sdk import Session, TTSRequest
from fish_audio_sdk.exceptions import HttpCodeErr


load_dotenv()

VALID_SAMPLE_RATES = {8000, 16000, 22050, 24000}
DEFAULT_LATENCY = os.getenv("FISH_LATENCY_MODE", "balanced")

FISH_API_KEY = os.getenv("FISH_API_SECRET") or os.getenv("FISH_AUDIO_API_KEY")
FISH_REFERENCE_ID = os.getenv("FISH_REFERENCE_ID")
SERVER_SHARED_SECRET = os.getenv("TTS_SERVER_SECRET")


app = Flask(__name__)


_session_lock = Lock()
_fish_session: Optional[Session] = None

# Call state management for webhook integration
_calls_lock = Lock()
_calls_state: dict[str, dict] = {}  # Stores call data: {call_id: {"status": "...", "data": {...}}}


def get_fish_session() -> Session:
    """Initialise and cache a Fish Audio session."""

    global _fish_session

    if not FISH_API_KEY:
        raise RuntimeError(
            "Fish Audio API key is not configured. Set FISH_API_SECRET or FISH_AUDIO_API_KEY."
        )

    if _fish_session is None:
        with _session_lock:
            if _fish_session is None:
                _fish_session = Session(FISH_API_KEY)

    return _fish_session


def error_response(message: str, status_code: int = 400):
    """Return a JSON error response."""

    return jsonify({"error": message}), status_code


def parse_sample_rate(raw_value) -> Optional[int]:
    """Attempt to coerce the sample rate to an integer."""

    if raw_value is None:
        return None

    if isinstance(raw_value, int):
        return raw_value

    if isinstance(raw_value, str) and raw_value.isdigit():
        return int(raw_value)

    return None


def synthesize_audio(
    text: str,
    sample_rate: int,
    reference_id: Optional[str] = None,
    latency: str = DEFAULT_LATENCY,
) -> bytes:
    """Generate PCM16 mono audio via Fish Audio."""

    session = get_fish_session()

    tts_request = TTSRequest(
        text=text,
        format="pcm",
        sample_rate=sample_rate,
        reference_id=reference_id,
        latency=latency,
    )

    audio = bytearray()

    try:
        for chunk in session.tts(tts_request):
            if chunk:
                audio.extend(chunk)
    except HttpCodeErr:
        raise

    audio_bytes = bytes(audio)

    if len(audio_bytes) % 2 == 1:
        audio_bytes = audio_bytes[:-1]

    return audio_bytes


@app.route("/api/webhooks/call", methods=["POST"])
def call_webhook():
    """Webhook endpoint for Vapi call status updates."""
    try:
        request_body = request.json
        message = request_body.get("message", {})
        message_type = message.get("type")
        status = message.get("status")
        call_id = message.get("call", {}).get("id")
        
        if not call_id:
            print("Warning: Missing call_id in webhook")
            return {"error": "Missing call_id"}, 400
        
        # Only process status-update messages
        if message_type != "status-update":
            print(f"Webhook received: Type={message_type}, Call={call_id} (skipped - not status-update)")
            return {"received": True}, 200
        
        print(f"Webhook received: Type={message_type}, Call={call_id}, Status={status}")
        
        # Store the call state
        with _calls_lock:
            _calls_state[call_id] = {}
            
            _calls_state[call_id]["status"] = status
            _calls_state[call_id]["updated_at"] = time.time()
        
        # Handle ended status
        if status == "ended":
            print(f"Call {call_id} ended - analysis will be retrieved via Vapi API")
        else:
            print(f"Call {call_id} status: {status}")
        
        return {"received": True}, 200
    
    except Exception as e:
        print(f"Error processing webhook: {e}")
        return {"error": str(e)}, 500


@app.route("/api/calls/<call_id>/status", methods=["GET"])
def get_call_status(call_id: str):
    """Check the status of a specific call."""
    with _calls_lock:
        if call_id not in _calls_state:
            return {"error": "Call not found"}, 404
        
        call_data = _calls_state[call_id]
        return jsonify({
            "call_id": call_id,
            "status": call_data.get("status"),
            "analysis": call_data.get("analysis"),
            "result": call_data.get("result"),
            "updated_at": call_data.get("updated_at")
        }), 200


@app.route("/health", methods=["GET"])
def health_check():
    """Simple health endpoint for monitoring."""

    return jsonify({"status": "ok"})


@app.route("/api/synthesize", methods=["POST"])
def synthesize_endpoint():
    request_id = str(uuid.uuid4())
    started_at = time.time()

    if SERVER_SHARED_SECRET:
        provided_secret = request.headers.get("X-Server-Secret")
        if provided_secret != SERVER_SHARED_SECRET:
            return error_response("Unauthorized", 401)

    payload = request.get_json(silent=True)
    if not payload or "message" not in payload:
        return error_response("Missing message object", 400)

    message = payload["message"]
    if message.get("type") != "voice-request":
        return error_response("Invalid message type", 400)

    text = message.get("text")
    if not text or not isinstance(text, str) or not text.strip():
        return error_response("Invalid or missing text", 400)

    sample_rate = parse_sample_rate(message.get("sampleRate"))
    if sample_rate not in VALID_SAMPLE_RATES:
        return (
            jsonify(
                {
                    "error": "Unsupported sample rate",
                    "supportedSampleRates": sorted(VALID_SAMPLE_RATES),
                }
            ),
            400,
        )

    requested_reference_id = message.get("referenceId") or message.get(
        "voice", {}
    ).get("referenceId")
[truncated — 1710 more characters]
```