Skip to main content

Documentation Index

Fetch the complete documentation index at: https://atp.hypertext.studio/llms.txt

Use this file to discover all available pages before exploring further.

​
Overview

This tutorial will guide you through the process of creating and sending your first notification using the Agent Triage Protocol (ATP). By the end, you’ll understand how to structure notifications, set appropriate actions, and handle responses.

​
Prerequisites

Before starting this tutorial, make sure you have:
  • Completed the Quickstart Guide and set up your ATP client
  • Registered your service with the ATP server
  • Set up a webhook endpoint for receiving responses (for production use)

​
Understanding Notification Components

A well-structured ATP notification consists of several key components:
  1. Title and Description: Clear, concise information about what the notification is for
  2. Actions: The possible responses a user can take
  3. Deadline: When a response is needed by
  4. Metadata: Additional context for the notification

​
Step 1: Planning Your Notification

Before writing code, consider what you’re asking the user to do. For this tutorial, we’ll create a notification that asks for approval to run a large database query. Think about:
  • What information does the user need to make a decision?
  • What actions should be available to the user?
  • How long should they have to respond?

​
Step 2: Define Your Actions

Actions are the possible responses a user can provide. Each action needs:
  • A unique ID
  • A descriptive label
  • A response type (SIMPLE, TEXT, NUMBER, etc.)
  • Optional constraints or flags
For our example, we’ll create two actions:
  1. “Approve” - A simple yes button
  2. “Deny” - With a text field for explaining why

​
Step 3: Create the Notification

Now let’s create the notification object: 
from atp import TriageNotification, Action, ResponseType
from datetime import datetime, timedelta

# Create a notification
notification = TriageNotification(
    title="Approve Database Query",
    description="""
    The system needs to run a large query that will analyze all customer records.
    This query may take up to 30 minutes and could impact database performance.
    Please approve or deny this operation.
    """,
    actions=[
        Action(
            id="approve",
            label="Approve Query",
            response_type=ResponseType.SIMPLE
        ),
        Action(
            id="deny",
            label="Deny",
            response_type=ResponseType.TEXT,
            constraints={"placeholder": "Reason for denial"}
        ),
        Action(
            id="schedule",
            label="Schedule for Later",
            response_type=ResponseType.DATETIME,
            constraints={"min": datetime.now().isoformat(), "max": (datetime.now() + timedelta(days=7)).isoformat()}
        )
    ],
    deadline=datetime.now() + timedelta(hours=1),
    metadata={
        "query_id": "analyze-customers-2023-q4",
        "estimated_duration": "30min",
        "estimated_rows": "1.2M",
        "requested_by": "data-analytics-team"
    }
)

​
Step 4: Send the Notification

Now let’s send the notification to the ATP server: 
# Send the notification
response = client.send_notification(notification)

# Store the notification ID for checking status later
notification_id = response.notification_id
print(f"Notification sent! ID: {notification_id}")

​
Step 5: Set Up Your Webhook Handler

To receive responses, you need a webhook endpoint. Here’s an example implementation: 
# Using Flask as an example web framework
from flask import Flask, request, jsonify
import hmac, hashlib

app = Flask(__name__)

@app.route('/atp/webhook', methods=['POST'])
def handle_webhook():
    # Verify webhook signature
    payload = request.data
    signature = request.headers.get('X-ATP-Signature')
    expected_signature = hmac.new(
        bytes(client.webhook_secret, 'utf-8'),
        payload,
        hashlib.sha256
    ).hexdigest()
    
    if not hmac.compare_digest(expected_signature, signature):
        return jsonify({'error': 'Invalid signature'}), 403
    
    # Process the webhook payload
    data = request.json
    notification_id = data.get('notification_id')
    action_id = data.get('action_id')
    response_data = data.get('response_data')
    
    # Handle different actions
    if action_id == 'approve':
        # Proceed with running the query
        run_database_query(data.get('metadata').get('query_id'))
        print(f"Query approved and started!")
    elif action_id == 'deny':
        # Log the denial reason
        reason = response_data
        print(f"Query denied. Reason: {reason}")
    elif action_id == 'schedule':
        # Schedule the query for later
        scheduled_time = response_data
        schedule_query(data.get('metadata').get('query_id'), scheduled_time)
        print(f"Query scheduled for: {scheduled_time}")
    
    # Acknowledge receipt of the webhook
    return jsonify({'status': 'success'}), 200

​
Step 6: Checking Notification Status (Optional)

If you want to proactively check the status of your notification: 
# Check notification status
def check_notification_status(notification_id):
    status = client.get_notification_status(notification_id)
    print(f"Notification status: {status.state}")
    print(f"Responded: {status.responded}")
    
    if status.responded:
        print(f"Action selected: {status.action_id}")
        if status.action_id == 'approve':
            print("Query was approved!")
        elif status.action_id == 'deny':
            print(f"Query was denied. Reason: {status.response_data}")
        elif status.action_id == 'schedule':
            print(f"Query was scheduled for: {status.response_data}")
    else:
        print("Still waiting for a response...")
    
    return status

# Check status every 5 minutes
import time
while True:
    status = check_notification_status(notification_id)
    if status.responded:
        break
    print("Waiting 5 minutes before checking again...")
    time.sleep(300)  # 5 minutes

​
Best Practices for Effective Notifications

To ensure your notifications are effective:
  1. Be Clear and Concise
    • Use descriptive titles that quickly convey purpose
    • Keep descriptions focused on what the user needs to know
    • Avoid technical jargon unless necessary
  2. Set Appropriate Deadlines
    • Match the deadline to the urgency of the decision
    • Consider time zones when setting deadlines
  3. Provide Sufficient Context
    • Include relevant data in the metadata field
    • Explain potential impacts of different actions
    • Link to additional information when necessary
  4. Design Thoughtful Actions
    • Limit options to what’s necessary (2-4 actions is typically ideal)
    • Use appropriate response types for the data you need
    • Make action labels clear and actionable (e.g., “Approve Query” not just “Yes”)
  5. Handle Responses Gracefully
    • Acknowledge when a user responds
    • Handle webhook failures and retries
    • Have fallback plans for expired notifications

​
Complete Example

Here’s a complete example that puts everything together: 
from atp import ATPClient, TriageNotification, Action, ResponseType
from datetime import datetime, timedelta
from flask import Flask, request, jsonify
import hmac, hashlib
import time
import threading

# Initialize the ATP client
client = ATPClient(
    api_key="sk_live_service_abc123",
    webhook_url="https://your-service.com/atp/webhook"
)

# Create and send a notification
def send_database_query_notification():
    notification = TriageNotification(
        title="Approve Database Query",
        description="""
        The system needs to run a large query that will analyze all customer records.
        This query may take up to 30 minutes and could impact database performance.
        Please approve or deny this operation.
        """,
        actions=[
            Action(
                id="approve",
                label="Approve Query",
                response_type=ResponseType.SIMPLE
            ),
            Action(
                id="deny",
                label="Deny",
                response_type=ResponseType.TEXT,
                constraints={"placeholder": "Reason for denial"}
            ),
            Action(
                id="schedule",
                label="Schedule for Later",
                response_type=ResponseType.DATETIME,
                constraints={"min": datetime.now().isoformat(), "max": (datetime.now() + timedelta(days=7)).isoformat()}
            )
        ],
        deadline=datetime.now() + timedelta(hours=1),
        metadata={
            "query_id": "analyze-customers-2023-q4",
            "estimated_duration": "30min",
            "estimated_rows": "1.2M",
            "requested_by": "data-analytics-team"
        }
    )
    
    # Send the notification
    response = client.send_notification(notification)
    notification_id = response.notification_id
    print(f"Notification sent! ID: {notification_id}")
    
    # Start a background thread to poll for status
    threading.Thread(target=poll_notification_status, args=(notification_id,)).start()
    
    return notification_id

# Poll for notification status
def poll_notification_status(notification_id):
    while True:
        status = client.get_notification_status(notification_id)
        print(f"Notification status: {status.state}")
        
        if status.responded:
            print(f"Action selected: {status.action_id}")
            if status.action_id == 'approve':
                run_database_query(status.metadata['query_id'])
                print("Query was approved and started!")
            elif status.action_id == 'deny':
                print(f"Query was denied. Reason: {status.response_data}")
            elif status.action_id == 'schedule':
                schedule_query(status.metadata['query_id'], status.response_data)
                print(f"Query was scheduled for: {status.response_data}")
            break
        
        # Check again in 5 minutes
        time.sleep(300)

# Flask web server for webhook
app = Flask(__name__)

@app.route('/atp/webhook', methods=['POST'])
def handle_webhook():
    # Verify webhook signature
    payload = request.data
    signature = request.headers.get('X-ATP-Signature')
    expected_signature = hmac.new(
        bytes(client.webhook_secret, 'utf-8'),
        payload,
        hashlib.sha256
    ).hexdigest()
    
    if not hmac.compare_digest(expected_signature, signature):
        return jsonify({'error': 'Invalid signature'}), 403
    
    # Process the webhook payload
    data = request.json
    notification_id = data.get('notification_id')
    action_id = data.get('action_id')
    response_data = data.get('response_data')
    metadata = data.get('metadata')
    
    # Handle different actions
    if action_id == 'approve':
        run_database_query(metadata['query_id'])
        print(f"Query approved and started!")
    elif action_id == 'deny':
        print(f"Query denied. Reason: {response_data}")
    elif action_id == 'schedule':
        schedule_query(metadata['query_id'], response_data)
        print(f"Query scheduled for: {response_data}")
    
    # Acknowledge receipt of the webhook
    return jsonify({'status': 'success'}), 200

# Example implementation functions
def run_database_query(query_id):
    print(f"Running query {query_id}...")
    # Implementation to run the query

def schedule_query(query_id, scheduled_time):
    print(f"Scheduling query {query_id} for {scheduled_time}...")
    # Implementation to schedule the query

if __name__ == '__main__':
    # Send a notification
    notification_id = send_database_query_notification()
    
    # Start the webhook server
    app.run(host='0.0.0.0', port=3000)

​
Next Steps

Now that you’ve created your first notification, you might want to explore: Congratulations on sending your first ATP notification! You’re now ready to integrate the Agent Triage Protocol into your applications.