Inbound Call Webhook with Python and FastAPI

Overview

Build a production-ready FastAPI webhook endpoint to handle inbound calls using the Telnyx Voice API. This tutorial demonstrates the command-event model of Call Control, proper webhook validation, and automated call handling with answer, speak, and hangup actions.

Prerequisites

Python 3.8 or higher.
A Telnyx account with an active API key from the Telnyx Portal.
A Telnyx phone number configured for inbound calls.
A Call Control Application ID (connection_id).
ngrok or similar tool for webhook testing.

Step 1: Setup

Install the required dependencies:

pip install telnyx fastapi uvicorn python-dotenv

Create a project directory and navigate into it:

mkdir telnyx-inbound-webhook
cd telnyx-inbound-webhook

Step 2: Configuration

Create a .env file in your project root to store credentials securely:

TELNYX_API_KEY=YOUR_API_KEY_HERE
TELNYX_CONNECTION_ID=YOUR_CONNECTION_ID_HERE

Replace YOUR_API_KEY_HERE with your actual API key and YOUR_CONNECTION_ID_HERE with your Call Control Application ID from the Telnyx Portal.

Step 3: Implementation

Create main.py and initialize the Telnyx client. Define helper functions to handle call actions with proper error handling:

import os
import telnyx
from dotenv import load_dotenv
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse

load_dotenv()

app = FastAPI(title="Telnyx Inbound Call Webhook")

# Initialize client with the new SDK pattern
client = telnyx.Telnyx(api_key=os.getenv("TELNYX_API_KEY"))


def answer_call(call_control_id: str) -> dict:
    """Answer an incoming call and return response data."""
    response = client.calls.actions.answer(call_control_id)
    return {
        "call_control_id": response.data.call_control_id,
        "status": "answered"
    }


def speak_to_caller(call_control_id: str, message: str) -> dict:
    """Play text-to-speech message to the caller."""
    response = client.calls.actions.speak(
        call_control_id,
        payload=message,
        voice="female",
        language="en-US"
    )
    return {
        "call_control_id": response.data.call_control_id,
        "status": "speaking"
    }


def hangup_call(call_control_id: str) -> dict:
    """Terminate the call."""
    response = client.calls.actions.hangup(call_control_id)
    return {
        "call_control_id": response.data.call_control_id,
        "status": "hangup"
    }

Step 4: Testing

Add the webhook endpoint with comprehensive event handling for production resilience:

@app.post("/webhook/voice")
async def handle_voice_webhook(request: Request):
    """Handle incoming voice webhook events from Telnyx."""
    try:
        payload = await request.json()
        event_type = payload.get("data", {}).get("event_type")
        call_control_id = payload.get("data", {}).get("payload", {}).get("call_control_id")

        if not event_type or not call_control_id:
            raise HTTPException(status_code=400, detail="Invalid webhook payload")

        # Handle different call events
        if event_type == "call.initiated":
            # Answer the incoming call
            result = answer_call(call_control_id)
            return result

        elif event_type == "call.answered":
            # Play welcome message after call is answered
            message = "Hello! Thank you for calling. This call will now end. Goodbye!"
            result = speak_to_caller(call_control_id, message)
            return result

        elif event_type == "call.speak.ended":
            # Hang up after the message finishes playing
            result = hangup_call(call_control_id)
            return result

        elif event_type == "call.hangup":
            # Call ended - log for cleanup
            return {
                "call_control_id": call_control_id,
                "status": "call_ended",
                "message": "Call cleanup completed"
            }

        # Return success for unhandled events
        return {"status": "received", "event_type": event_type}

    except telnyx.AuthenticationError:
        raise HTTPException(status_code=401, detail="Invalid API key")
    except telnyx.RateLimitError:
        raise HTTPException(status_code=429, detail="Rate limit exceeded")
    except telnyx.APIStatusError as e:
        raise HTTPException(status_code=e.status_code, detail=str(e))
    except telnyx.APIConnectionError:
        raise HTTPException(status_code=503, detail="Network error connecting to Telnyx")


@app.get("/health")
async def health_check():
    """Health check endpoint for monitoring."""
    return {"status": "healthy", "service": "telnyx-voice-webhook"}


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Start the server:

python main.py

Expose your local server using ngrok:

ngrok http 8000

Configure your Telnyx phone number to send webhooks to your ngrok URL:

https://your-ngrok-url.ngrok.io/webhook/voice

Test by calling your Telnyx phone number. You should hear the welcome message before the call ends.

Complete Code

#!/usr/bin/env python3
"""Production-ready FastAPI webhook for handling inbound calls via Telnyx."""

import os
import telnyx
from dotenv import load_dotenv
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse

load_dotenv()

app = FastAPI(title="Telnyx Inbound Call Webhook")

# Initialize client with the new SDK pattern
client = telnyx.Telnyx(api_key=os.getenv("TELNYX_API_KEY"))


def answer_call(call_control_id: str) -> dict:
    """Answer an incoming call and return response data."""
    response = client.calls.actions.answer(call_control_id)
    return {
        "call_control_id": response.data.call_control_id,
        "status": "answered"
    }


def speak_to_caller(call_control_id: str, message: str) -> dict:
    """Play text-to-speech message to the caller."""
    response = client.calls.actions.speak(
        call_control_id,
        payload=message,
        voice="female",
        language="en-US"
    )
    return {
        "call_control_id": response.data.call_control_id,
        "status": "speaking"
    }


def hangup_call(call_control_id: str) -> dict:
    """Terminate the call."""
    response = client.calls.actions.hangup(call_control_id)
    return {
        "call_control_id": response.data.call_control_id,
        "status": "hangup"
    }


@app.post("/webhook/voice")
async def handle_voice_webhook(request: Request):
    """Handle incoming voice webhook events from Telnyx."""
    try:
        payload = await request.json()
        event_type = payload.get("data", {}).get("event_type")
        call_control_id = payload.get("data", {}).get("payload", {}).get("call_control_id")

        if not event_type or not call_control_id:
            raise HTTPException(status_code=400, detail="Invalid webhook payload")

        # Handle different call events
        if event_type == "call.initiated":
            # Answer the incoming call
            result = answer_call(call_control_id)
            return result

        elif event_type == "call.answered":
            # Play welcome message after call is answered
            message = "Hello! Thank you for calling. This call will now end. Goodbye!"
            result = speak_to_caller(call_control_id, message)
            return result

        elif event_type == "call.speak.ended":
            # Hang up after the message finishes playing
            result = hangup_call(call_control_id)
            return result

        elif event_type == "call.hangup":
            # Call ended - log for cleanup
            return {
                "call_control_id": call_control_id,
                "status": "call_ended",
                "message": "Call cleanup completed"
            }

        # Return success for unhandled events
        return {"status": "received", "event_type": event_type}

    except telnyx.AuthenticationError:
        raise HTTPException(status_code=401, detail="Invalid API key")
    except telnyx.RateLimitError:
        raise HTTPException(status_code=429, detail="Rate limit exceeded")
    except telnyx.APIStatusError as e:
        raise HTTPException(status_code=e.status_code, detail=str(e))
    except telnyx.APIConnectionError:
        raise HTTPException(status_code=503, detail="Network error connecting to Telnyx")


@app.get("/health")
async def health_check():
    """Health check endpoint for monitoring."""
    return {"status": "healthy", "service": "telnyx-voice-webhook"}


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Troubleshooting

Issue	Problem	Solution
Webhook Not Receiving Events	Your FastAPI server is running but Telnyx webhooks are not being received when calls are made to your phone number.	Verify your ngrok tunnel is active and the webhook URL in your Telnyx Call Control Application matches your ngrok URL exactly (including `/webhook/voice` path). Check that your phone number is associated with the correct Call Control Application in the Telnyx Portal. Test the webhook URL directly with a GET request to confirm it's accessible.
Authentication Error (401)	The webhook endpoint returns HTTP 401 errors when processing Telnyx events.	Verify your `TELNYX_API_KEY` in the `.env` file matches the key shown in the Telnyx Portal. Ensure there are no trailing spaces or quotes around the API key. If the key was regenerated recently, update your environment file and restart the FastAPI server with `python main.py`.
Call Actions Failing	Call control actions like answer, speak, or hangup return errors or don't execute properly.	Confirm your `TELNYX_CONNECTION_ID` environment variable contains your Call Control Application ID, not a phone number or other identifier. Verify the Call Control Application is properly configured in the Telnyx Portal with the correct webhook URL. Check that the `call_control_id` from the webhook payload is being passed correctly to the action functions.