Webhooks & Events

from rubric import Rubric

client = Rubric()

# Create webhook endpoint
webhook = client.webhooks.create(
    name="production-notifications",

    # Your endpoint
    url="https://your-api.example.com/webhooks/rubric",

    # Events to receive
    events=[
        "evaluation.completed",
        "review.completed",
        "alert.triggered",
        "safety.critical"
    ],

    # Secret for signature verification
    secret="whsec_your_signing_secret",

    # Optional filters
    filters={
        "projects": ["patient-triage", "symptom-checker"],
        "severity": ["high", "critical"]
    },

    # Retry configuration
    retry_config={
        "max_retries": 5,
        "backoff": "exponential",
        "initial_delay_ms": 1000
    }
)

print(f"Webhook created: {webhook.id}")
print(f"Signing secret: {webhook.secret}")

{
  "id": "evt_abc123",
  "type": "evaluation.completed",
  "created_at": "2024-03-15T10:30:00Z",

  "data": {
    "evaluation_id": "eval_xyz789",
    "project_id": "patient-triage",
    "name": "Daily Evaluation - March 15",
    "status": "completed",

    "summary": {
      "total_samples": 500,
      "triage_accuracy": 0.87,
      "safety_score": 0.95,
      "samples_for_review": 23
    },

    "completed_at": "2024-03-15T10:30:00Z"
  },

  "metadata": {
    "model_version": "triage-v2.4",
    "dataset": "ds_daily_march15"
  }
}

from flask import Flask, request, abort
import hmac
import hashlib

app = Flask(__name__)
WEBHOOK_SECRET = "whsec_your_signing_secret"

@app.route('/webhooks/rubric', methods=['POST'])
def handle_webhook():
    # Get signature from header
    signature = request.headers.get('X-Rubric-Signature')
    timestamp = request.headers.get('X-Rubric-Timestamp')

    if not signature or not timestamp:
        abort(401)

    # Verify signature
    payload = request.get_data(as_text=True)
    signed_payload = f"{timestamp}.{payload}"

    expected_signature = hmac.new(
        WEBHOOK_SECRET.encode(),
        signed_payload.encode(),
        hashlib.sha256
    ).hexdigest()

    if not hmac.compare_digest(signature, f"sha256={expected_signature}"):
        abort(401)

    # Verify timestamp is recent (within 5 minutes)
    import time
    if abs(time.time() - int(timestamp)) > 300:
        abort(401)

    # Process webhook
    event = request.json
    handle_event(event)

    return {'status': 'ok'}

def handle_event(event):
    event_type = event['type']

    if event_type == 'evaluation.completed':
        handle_evaluation_completed(event['data'])
    elif event_type == 'safety.critical':
        handle_safety_alert(event['data'])
    # ... handle other events

def handle_evaluation_completed(data):
    """Handle completed evaluation."""
    evaluation_id = data['evaluation_id']
    summary = data['summary']

    # Check if metrics meet thresholds
    if summary['triage_accuracy'] < 0.80:
        send_slack_alert(
            channel="#ml-alerts",
            message=f"Low triage accuracy: {summary['triage_accuracy']:.1%}"
        )

    # Update dashboard
    update_metrics_dashboard(evaluation_id, summary)

    # Trigger downstream workflows
    if data.get('samples_for_review', 0) > 0:
        notify_reviewers(evaluation_id)

def handle_safety_alert(data):
    """Handle critical safety event - immediate action required."""
    sample_id = data['sample_id']
    issue = data['issue']

    # Page on-call immediately
    pagerduty.trigger_incident(
        title=f"Critical Safety Issue - {issue['type']}",
        details=issue,
        severity="critical"
    )

    # Create incident ticket
    create_incident_ticket(sample_id, issue)

    # Log for audit
    log_safety_event(sample_id, issue)

def handle_review_completed(data):
    """Handle completed human review."""
    sample_id = data['sample_id']
    review = data['review']

    # Check for disagreement with AI
    if review['ai_correct'] == False:
        # Route for training data
        add_to_training_queue(sample_id, review)

    # Update metrics
    update_reviewer_metrics(review['reviewer_id'], review)

from rubric import Rubric

client = Rubric()

# Configure Slack integration directly
client.webhooks.create(
    name="slack-notifications",

    # Slack incoming webhook URL
    url="https://hooks.slack.com/services/T00000000/B00000000/XXXXXXXX",

    # Events to send to Slack
    events=[
        "evaluation.completed",
        "alert.triggered"
    ],

    # Format for Slack
    format="slack",

    # Custom message templates
    templates={
        "evaluation.completed": {
            "text": "Evaluation completed: {{data.name}}",
            "blocks": [
                {
                    "type": "section",
                    "text": {
                        "type": "mrkdwn",
                        "text": "*{{data.name}}* completed\n"
                               "Accuracy: {{data.summary.triage_accuracy | percent}}\n"
                               "Safety: {{data.summary.safety_score | percent}}"
                    }
                }
            ]
        }
    }
)

from rubric import Rubric

client = Rubric()

# Send test event
test_result = client.webhooks.test(
    webhook_id="wh_abc123",

    # Event type to simulate
    event_type="evaluation.completed",

    # Sample payload
    payload={
        "evaluation_id": "eval_test",
        "summary": {
            "triage_accuracy": 0.85,
            "safety_score": 0.92
        }
    }
)

print(f"Test result: {test_result.status}")
print(f"Response code: {test_result.response_code}")
print(f"Response body: {test_result.response_body}")

from rubric import Rubric

client = Rubric()

# List webhooks
webhooks = client.webhooks.list()
for wh in webhooks:
    print(f"{wh.name}: {wh.url} ({wh.status})")

# View delivery history
deliveries = client.webhooks.deliveries(
    webhook_id="wh_abc123",
    limit=100
)

for delivery in deliveries:
    status = "OK" if delivery.success else "FAIL"
    print(f"{status} {delivery.event_type} - {delivery.response_code}")

# Retry failed deliveries
client.webhooks.retry_failed(
    webhook_id="wh_abc123",
    since="2024-03-15T00:00:00Z"
)

# Disable webhook
client.webhooks.disable("wh_abc123")

# Delete webhook
client.webhooks.delete("wh_abc123")