Building a Durable AI Call Campaign Orchestrator with Telnyx

Building an outbound call campaign is easy. Building one that is durable, rate-limited, and stateful is significantly harder. You need to queue calls, respect rate limits to avoid carrier flags, track the results of each conversation, and notify stakeholders when the campaign completes. Traditionally, this requires stitching together multiple vendors for voice, messaging, and orchestration.

With Telnyx's AI Communications Infrastructure, you get a single API surface for voice, messaging, and stateful workflow orchestration. In this post, we'll walk through the AI Call Campaign Orchestrator, a production-grade sample app built with Python, Flask, and the Telnyx Agent SDK.

What the App Does

The AI Call Campaign Orchestrator is a Flask application that manages outbound call campaigns. You provide a list of phone numbers, and the application:

1. Queues the calls using the Telnyx Agent SDK.
2. Enforces a rate limit (e.g., 10 calls per minute) to ensure compliance and reliability.
3. Uses Call Control to place the calls, answer them, and interact with the recipient.
4. Saves the result of each call to a local SQLite database.
5. Sends an SMS summary to a designated number once the entire campaign is complete.

Instead of managing queues and rate limiters yourself, the Agent SDK handles the orchestration primitives, allowing you to focus on the business logic of the call itself.

How It Works

The application is built around a CampaignAgent that handles the lifecycle of the campaign. When you hit the POST /campaign endpoint with a list of phone numbers, the agent takes over.

1. Queueing and Rate Limiting

The agent uses this.queue() to enqueue all the calls in the campaign. To prevent flooding the network and triggering carrier rate limits, it uses this.schedule() to enforce a rate limit defined by the CALL_RATE_LIMIT_PER_MINUTE environment variable.

# The agent handles the queueing and scheduling internally
@app.route('/campaign', methods=['POST'])
def start_campaign():
    phone_numbers = request.json.get('phone_numbers', [])
    campaign_id = campaign_agent.start_campaign(phone_numbers)
    return jsonify({"campaign_id": campaign_id, "status": "started"}), 202

2. Placing Calls with Call Control

For each call, the application uses the Telnyx Python SDK to create an outbound Call Control call. Call Control is Telnyx's programmatic interface for controlling voice calls.

import telnyx

telnyx.Call.create(
    connection_id=os.environ['TELNYX_CONNECTION_ID'],
    to=phone_number,
    from_=os.environ['TELNYX_PHONE_NUMBER'],
    webhook_url=os.environ['TELNYX_WEBHOOK_URL']
)

3. Handling Webhooks

When the call is answered, Telnyx sends a webhook to the /webhooks/call endpoint. The app uses telnyx.webhooks.unwrap() to verify the Ed25519 signature, ensuring the request is genuinely from Telnyx.

@app.route('/webhooks/call', methods=['POST'])
def webhooks_call():
    payload = telnyx.webhooks.unwrap(
        request.data,
        request.headers.get('telnyx-signature-ed25519'),
        request.headers.get('telnyx-timestamp'),
        os.environ['TELNYX_PUBLIC_KEY']
    )
    # Handle the call event (answer, hangup, etc.)
    return jsonify({"status": "ok"}), 200

4. Tracking Results in SQL

As calls progress, the agent saves the results to a SQLite database. This allows you to query the status of the campaign at any time via the GET /campaign/<campaign_id> endpoint.

-- The agent executes this internally
INSERT INTO results (campaign_id, to_number, status, call_control_id)
VALUES (?, ?, ?, ?);

5. Sending the SMS Summary

Once all calls in the queue are completed, the agent sends an SMS summary using the Telnyx SMS API.

telnyx.Message.create(
    from_=os.environ['TELNYX_SMS_FROM'],
    to=os.environ['TELNYX_SMS_TO'],
    text=f"Campaign {campaign_id} complete. {completed} calls processed."
)

Setup

To run this locally, you need a Telnyx account, a Call Control Application, and an API key.

1. Clone the repository:

``bash git clone https://github.com/team-telnyx/telnyx-code-examples.git cd telnyx-code-examples/ai-call-campaign-orchestrator ``

1. Set up your environment:

``bash python3 -m venv venv source venv/bin/activate pip install -r requirements.txt cp .env.example .env ``

1. Configure your .env file:

You will need to fill in your Telnyx API Key, Connection ID, Phone Number, Public Key, Webhook URL, and SMS details. You also need to set CALL_RATE_LIMIT_PER_MINUTE and PORT.

1. Run the application:

``bash python app.py ``

Your server will start on http://localhost:5000. You can test the health endpoint at /health and start a campaign by sending a POST request to /campaign with a list of phone numbers.

Conclusion

Building durable communication workflows doesn't have to mean stitching together disparate APIs and managing your own queue infrastructure. The AI Call Campaign Orchestrator demonstrates how Telnyx's AI Communications Infrastructure provides the primitives you need—queueing, rate limiting, Call Control, and SMS—in a single, cohesive platform.

Check out the full code on GitHub and start building your own campaign workflows today.