CueAPI Python Setup: AI Agent Scheduling Tutorial

Building AI agents that run reliably requires more than just writing good code. You need to know your agent received its job, executed it correctly, and delivered the expected outcome. This CueAPI Python setup tutorial walks you through creating accountable agents that report back when they work.

Platform schedulers like cron fire jobs into the void. Your Python agent might crash, fail silently, or produce no output, and you won't know until a user complains. CueAPI gives you delivery confirmation, execution visibility, and outcome tracking for every agent task.

TL;DR: Use CueAPI's REST API with Python's requests library, create a cue with a webhook endpoint, build an agent handler that reports outcomes, and get guaranteed delivery with retry logic. Your agent will tell you when it works, not just when it runs.

Key Takeaways: - CueAPI REST API works with any HTTP client like requests or httpx for full API access - Cues deliver jobs to any Python application with at-least-once guarantee and 3 retry attempts - Outcome reporting proves your agent did the work with evidence like file paths or API responses - Failed executions trigger email alerts within 5 minutes, not hours later when users complain - Proper outcome tracking eliminates manual monitoring and provides automatic success/failure reporting

What You'll Build

You'll create a Python agent that processes data files every morning at 9 AM Eastern. The agent will receive delivery confirmation, process files, report success with evidence, and alert you immediately if anything fails.

By the end, you'll have a complete agent accountability system that works on any Python environment: local machines, Replit, cloud servers, or container platforms.

Prerequisites

• Python 3.8 or higher
• Basic understanding of HTTP requests and webhook endpoints
• A text editor or Python IDE
• 10 minutes to complete the tutorial

Step 1: Install HTTP Client Library

CueAPI provides a REST API that works with any HTTP client. Install requests for making API calls:

pip install requests

Verify the installation by importing the library:

import requests

print("Requests library installed successfully")

⚠️ Warning: This tutorial requires Python 3.8+. Check your version with python --version before proceeding.

Expected output:

Requests library installed successfully

Step 2: Get Your API Key

Sign up for a CueAPI account at https://dashboard.cueapi.ai/signup to get your API key. The free tier includes 10 cues and 300 executions per month.

Create a new file called config.py and add your API key:

# config.py
CUEAPI_KEY = "cue_sk_your_key_here"
WEBHOOK_URL = "https://your-domain.com/webhook"  # We'll set this up next
BASE_URL = "https://api.cueapi.ai"

Create a helper function for API requests:

# main.py
import requests
from config import CUEAPI_KEY, BASE_URL

def make_api_request(method, endpoint, data=None):
    headers = {
        "Authorization": f"Bearer {CUEAPI_KEY}",
        "Content-Type": "application/json"
    }
    url = f"{BASE_URL}{endpoint}"
    response = requests.request(method, url, json=data, headers=headers)
    response.raise_for_status()
    return response.json()

print("CueAPI client initialized")

📝 Note: Keep your API key secure. Never commit it to version control. Use environment variables in production.

Step 3: Create Your First Cue

A cue defines when your agent runs, where it receives jobs, and how failures are handled. This example creates a daily data processing task.

# create_cue.py
import requests
from config import CUEAPI_KEY, WEBHOOK_URL, BASE_URL

def create_cue():
    cue_data = {
        "name": "daily-data-processor",
        "description": "Process customer data files every morning",
        "schedule": {
            "type": "recurring",
            "cron": "0 9 * * *",  # 9 AM daily
            "timezone": "America/New_York"
        },
        "transport": "webhook",
        "callback": {
            "url": WEBHOOK_URL,
            "method": "POST",
            "headers": {"Authorization": "Bearer your-secret-here"}
        },
        "payload": {
            "task": "process_files",
            "source": "/data/customer_files",
            "format": "csv"
        },
        "retry": {
            "max_attempts": 3,
            "backoff_minutes": [1, 5, 15]
        },
        "on_failure": {
            "email": True,
            "webhook": None,
            "pause": False
        }
    }

    headers = {
        "Authorization": f"Bearer {CUEAPI_KEY}",
        "Content-Type": "application/json"
    }

    response = requests.post(f"{BASE_URL}/v1/cues", json=cue_data, headers=headers)
    response.raise_for_status()
    cue = response.json()
    
    print(f"Created cue: {cue['id']}")
    print(f"Next run: {cue.get('next_run_at', 'Not scheduled')}")
    return cue

if __name__ == "__main__":
    create_cue()

You can also create the same cue using curl:

curl -X POST https://api.cueapi.ai/v1/cues \
  -H "Authorization: Bearer cue_sk_your_key_here" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "daily-data-processor",
    "schedule": {
      "type": "recurring", 
      "cron": "0 9 * * *",
      "timezone": "America/New_York"
    },
    "transport": "webhook",
    "callback": {
      "url": "https://your-domain.com/webhook",
      "method": "POST"
    },
    "payload": {"task": "process_files"},
    "retry": {"max_attempts": 3, "backoff_minutes": [1, 5, 15]}
  }'

Expected output:

Created cue: cue_abc123xyz
Next run: 2024-03-25T13:00:00Z

✅ Success: Your cue is now scheduled. CueAPI will deliver the payload to your webhook URL at 9 AM Eastern every day.

Step 4: Build an Agent Handler

Create a webhook endpoint that receives jobs from CueAPI and processes them. This example uses Flask for the web server.

pip install flask requests

# agent.py
from flask import Flask, request, jsonify
import os
import csv
from datetime import datetime
import requests
from config import CUEAPI_KEY, BASE_URL

app = Flask(__name__)

def report_outcome(execution_id, success, result=None, error=None):
    """Report execution outcome to CueAPI"""
    headers = {
        "Authorization": f"Bearer {CUEAPI_KEY}",
        "Content-Type": "application/json"
    }
    
    data = {"success": success}
    if result:
        data["result"] = result
    if error:
        data["error"] = error
    
    response = requests.post(
        f"{BASE_URL}/v1/executions/{execution_id}/outcome",
        json=data,
        headers=headers
    )
    response.raise_for_status()

@app.route('/webhook', methods=['POST'])
def handle_cue():
    # Get execution details from CueAPI headers
    execution_id = request.headers.get('X-CueAPI-Execution-ID')
    cue_id = request.headers.get('X-CueAPI-Cue-ID')
    
    if not execution_id:
        return jsonify({"error": "Missing execution ID"}), 400
    
    # Get the job payload
    payload = request.json
    task = payload.get('task')
    source = payload.get('source', '/data/customer_files')
    
    try:
        if task == 'process_files':
            result = process_customer_files(source)
            
            # Report success
            report_outcome(
                execution_id,
                success=True,
                result=f"Processed {result['record_count']} records"
            )
            
            return jsonify({
                "success": True,
                "records_processed": result['record_count'],
                "output_file": result['output_path']
            })
        else:
            raise ValueError(f"Unknown task: {task}")
            
    except Exception as e:
        # Report failure
        report_outcome(
            execution_id,
            success=False,
            error=str(e)
        )
        return jsonify({"error": str(e)}), 500

def process_customer_files(source_dir):
    """Process CSV files in the source directory"""
    if not os.path.exists(source_dir):
        raise FileNotFoundError(f"Source directory not found: {source_dir}")
    
    csv_files = [f for f in os.listdir(source_dir) if f.endswith('.csv')]
    if not csv_files:
        raise ValueError(f"No CSV files found in {source_dir}")
    
    total_records = 0
    output_path = f"/data/processed/batch_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
    
    # Process files (simplified example)
    for csv_file in csv_files:
        file_path = os.path.join(source_dir, csv_file)
        with open(file_path, 'r') as f:
            reader = csv.DictReader(f)
            records = list(reader)
            total_records += len(records)
    
    # Write processed output
    os.makedirs(os.path.dirname(output_path), exist_ok=True)
    with open(output_path, 'w') as f:
        f.write(f"Processed {total_records} records at {datetime.now()}")
    
    return {
        'record_count': total_records,
        'output_path': output_path,
        'files_processed': len(csv_files)
    }

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000, debug=True)

📝 Note: The agent automatically reports success and failure outcomes using the CueAPI REST endpoints.

Step 5: Test Your Agent

Start your Flask server:

python agent.py

Manually trigger your cue to test the integration:

# test_trigger.py
import requests
from config import CUEAPI_KEY, BASE_URL

def trigger_cue(cue_id):
    headers = {
        "Authorization": f"Bearer {CUEAPI_KEY}",
        "Content-Type": "application/json"
    }
    
    response = requests.post(
        f"{BASE_URL}/v1/cues/{cue_id}/trigger",
        headers=headers
    )
    response.raise_for_status()
    execution = response.json()
    
    print(f"Triggered execution: {execution['id']}")
    print(f"Status: {execution.get('status', 'queued')}")
    return execution

if __name__ == "__main__":
    # Replace with your actual cue ID
    cue_id = "cue_abc123xyz"
    trigger_cue(cue_id)

Expected output:

Triggered execution: exec_def456ghi
Status: queued

Check your Flask server logs. You should see the webhook request and processing output.

✅ Success: Your agent received the job, processed it, and reported back to CueAPI automatically.

Step 6: Add Outcome Reporting

Enhance your agent with detailed outcome reporting:

# Enhanced reporting example
@app.route('/webhook', methods=['POST'])
def handle_cue():
    execution_id = request.headers.get('X-CueAPI-Execution-ID')
    
    if not execution_id:
        return jsonify({"error": "Missing execution ID"}), 400
    
    try:
        payload = request.json
        result = process_customer_files(payload.get('source'))
        
        # Report final outcome
        report_outcome(
            execution_id,
            success=True,
            result={
                "records_processed": result['record_count'],
                "files_processed": result['files_processed'],
                "output_path": result['output_path']
            }
        )
        
        return jsonify({
            "success": True,
            "execution_id": execution_id,
            "timestamp": datetime.now().isoformat(),
            "metrics": {
                "records_processed": result['record_count'],
                "files_processed": result['files_processed'],
                "output_path": result['output_path']
            }
        })
        
    except Exception as e:
        # Report failure
        report_outcome(
            execution_id,
            success=False,
            error=str(e)
        )
        return jsonify({"error": str(e)}), 500

Step 7: Handle Failures

CueAPI automatically retries failed deliveries, but your agent should handle business logic failures gracefully.

# Enhanced error handling
@app.route('/webhook', methods=['POST'])
def handle_cue():
    execution_id = request.headers.get('X-CueAPI-Execution-ID')
    
    try:
        payload = request.json
        task = payload.get('task')
        
        if task == 'process_files':
            try:
                result = process_customer_files(payload.get('source'))
                
                report_outcome(execution_id, success=True, result=result)
                return jsonify({"success": True, "result": result})
                
            except FileNotFoundError as e:
                report_outcome(execution_id, success=False, error=str(e))
                raise
                
            except ValueError as e:
                report_outcome(execution_id, success=False, error=str(e))
                raise
                
        else:
            error_msg = f"Unknown task type: {task}"
            report_outcome(execution_id, success=False, error=error_msg)
            raise ValueError(error_msg)
            
    except Exception as e:
        return jsonify({"error": str(e)}), 500

⚠️ Warning: Always report outcomes to CueAPI so it knows whether your agent succeeded or failed.

Before vs After: Platform Scheduler vs CueAPI

Before (Cron/Platform Scheduler):

# ~/.crontab
0 9 * * * /usr/bin/python3 /path/to/agent.py

# If it fails, you find out when users complain
# No delivery confirmation
# No outcome tracking  
# No retry logic
# No alerting

After (CueAPI):

# Scheduled with guaranteed delivery
# Automatic retries on failure
# Outcome tracking with evidence
# Email alerts within minutes
# Works on any platform - no crontab needed

cue_data = {
    "name": "daily-processor",
    "schedule": {"type": "recurring", "cron": "0 9 * * *", "timezone": "UTC"},
    "transport": "webhook",
    "callback": {"url": "https://your-agent.com/webhook", "method": "POST"},
    "retry": {"max_attempts": 3, "backoff_minutes": [1, 5, 15]}
}

The key difference: platform schedulers fire jobs into the void. CueAPI makes your agents accountable for the work you gave them.

Troubleshooting Common Issues

Webhook not receiving requests:

• Check that your server is publicly accessible
• Verify the webhook URL in your cue configuration
• Test connectivity with curl -X POST your-webhook-url

Outcome reporting failing:

• Verify your API key has correct permissions
• Check that execution_id is passed correctly from headers
• Ensure you're using the correct endpoint format

SSL/HTTPS errors:

• CueAPI requires HTTPS webhooks in production
• Use ngrok for local testing: ngrok http 5000
• Update your cue's callback URL to the ngrok URL

Import errors:

• Verify requests installation: pip show requests
• Check Python version compatibility (3.8+ required)
• Try reinstalling: pip uninstall requests && pip install requests

Next Steps

Now that your agent reports outcomes, you can build more sophisticated workflows:

1. Add conditional logic based on execution results
2. Chain multiple agents using CueAPI's webhook outcomes
3. Monitor trends in agent performance over time
4. Scale to multiple agents with different schedules and payloads

Your agent now reports back when it works. You'll know within minutes, not days, if something goes wrong. Silent failures become visible failures you can actually fix.

Make your agents accountable. Know they worked. Get on with building.

Try it yourself. Free tier available at https://dashboard.cueapi.ai/signup.

Frequently Asked Questions

Does the CueAPI REST API work with async frameworks like FastAPI?

Yes, the REST API works with any Python web framework. For FastAPI, use the same webhook pattern but with async handlers. Use httpx or aiohttp for async HTTP requests to CueAPI endpoints.

Can I use CueAPI without exposing a public webhook endpoint?

Yes, use the worker transport instead of webhook transport. Your agent polls for jobs rather than receiving webhook deliveries. Perfect for agents behind firewalls or on local machines.

What happens if my agent takes longer than expected?

CueAPI will wait for your outcome report. Your agent should report outcomes promptly after processing completes.

How do I schedule one-time tasks instead of recurring cues?

Set the schedule type to "once" with an "at" timestamp instead of "recurring" with a cron expression. Perfect for scheduled maintenance tasks or delayed processing jobs.

Can I modify cue schedules after creation?

CueAPI allows you to create new cues with updated configurations. Changes take effect immediately for the next scheduled execution.

Sources

• CueAPI API Reference: Complete REST API documentation: https://docs.cueapi.ai/api-reference/overview
• Flask Documentation: Python web framework for building webhook endpoints
• Cron Expression Format: Standard cron syntax for scheduling recurring tasks

About the author: Govind Kavaturi is co-founder of Vector, a portfolio of AI-native products. He believes the next phase of the internet is built for agents, not humans.