Dela via


Webhooks för Azure OpenAI i AI Foundry Models

Med Azure OpenAI-webhooks kan dina program ta emot realtidsmeddelanden om API-händelser, till exempel batchavslut eller inkommande samtal. Genom att prenumerera på webhook-händelser kan du automatisera arbetsflöden, utlösa aviseringar och integrera med andra system sömlöst. Den här guiden beskriver hur du konfigurerar en webhook-server, skyddar dina slutpunkter, distribuerar och felsöker vanliga problem.

Förutsättningar

Installera nödvändiga Python-paket:

pip install flask openai websockets requests

Installation av Webhook-server

En webhook-server är ett program som lyssnar efter och bearbetar automatiserade meddelanden (webhooks) som skickas av Azure OpenAI när specifika händelser inträffar.

Skapa webhook-lyssnarprogrammet

Skapa en fil med namnet app.py med följande Flask-program som tar emot och bearbetar webhook-händelser:

from flask import Flask, request, Response
from openai import OpenAI, InvalidWebhookSignatureError
import os
import logging

app = Flask(__name__)

# Configure logging for Azure App Service
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

client = OpenAI(
    # api-key parameter is required, but If you are only using the client for webhooks the key can be a placeholder string 
    api_key=os.environ.get("OPENAI_API_KEY", "placeholder-key-for-webhooks-only"), 
    webhook_secret=os.environ["OPENAI_WEBHOOK_SECRET"] # This will be created later
)

@app.route("/webhook", methods=["POST"])
def webhook():
    """Webhook endpoint to receive and process OpenAI events."""
    try:
        # Unwrap and verify the message using the webhook secret
        event = client.webhooks.unwrap(request.data, request.headers)
        # Process the event based on type
        if event.type == "realtime.call.incoming":
            logger.info(f"Received a realtime.call.incoming message for call_id = {event.data.call_id}")
            # Add your custom logic here:
            # - Log the call details
            # - Trigger notifications
            # - Update databases
            # - Start call processing workflows
        elif event.type == "response.completed":
            logger.info(f"Received a response.completed message")
            # Add your custom logic here
        else:
            logger.info(f"Received unexpected message of type = {event.type}")
        # Always return 200 to acknowledge receipt
        return Response(status=200)
    except InvalidWebhookSignatureError as e:
        logger.error(f"Invalid signature: {e}")
        return Response("Invalid signature", status=400)
    except Exception as e:
        logger.error(f"Error processing webhook: {e}")
        return Response("Internal server error", status=500)

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

Skapa en requirements.txt fil

Skapa en requirements.txt fil i samma katalog som filen app.py :

flask
openai
websockets
requests

Skapa en Azure-webbapp

Distribuera din webhook-server genom att använda az webapp up. Kommandot måste köras från mappen där app.py koden för ditt program finns.

az webapp up --name unique-webhook-handler-name --resource-group myResourceGroup --runtime "PYTHON:3.12"

Det här kommandot kommer att:

  • Skapa en resursgrupp om den inte finns.
  • Skapa en App Service-plan.
  • Skapa en webbapp.
  • Distribuera koden.

Din webhook-URL blir: https://unique-webhook-handler-name.azurewebsites.net/webhook

Skapa webhook-slutpunkt

För Azure OpenAI måste webhook-slutpunkter skapas med hjälp av REST-API:et. Registrera lyssnaren för att ta emot webhook-händelser för specifika händelsetyper.

import requests
import json
import os

AZURE_OPENAI_API_KEY = os.environ.get("AZURE_OPENAI_API_KEY")
WEBHOOK_NAME = "<WEBHOOK-NAME>" # Give your webhook a custom name
WEBHOOK_URL = "<YOUR-URL>/webhook"  # e.g., "https://unique-webhook-handler-name.azurewebsites.net/webhook"

url = "https://<YOUR_RESOURCE_NAME>.openai.azure.com/openai/v1/dashboard/webhook_endpoints"

headers = {
    "api-key": AZURE_OPENAI_API_KEY,
    "Content-Type": "application/json"
}

data = {
    "name": WEBHOOK_NAME,
    "url": WEBHOOK_URL,
    "event_types": ["response.completed"]
}

response = requests.post(url, headers=headers, json=data)

print(json.dumps(response.json(), indent=2))

Viktigt!

Signeringshemligheten visas bara en gång när den skapas. Hemligheter kan lagras på ett säkert sätt med Hjälp av Azure Key Vault.

Konfigurera webhookhemlighet i Azure Web App

Ange webhook-signeringshemligheten som en miljövariabel i azure-webbappen:

az webapp config appsettings set --name unique-webhook-handler-name \
  --resource-group myResourceGroup \
  --settings OPENAI_WEBHOOK_SECRET="<YOUR-SIGNING-SECRET-GOES-HERE>"

När du har angett miljövariabeln startar du om webbappen:

az webapp restart --name unique-webhook-handler-name --resource-group myResourceGroup

Testning av webhook-meddelanden

Kontrollera först att webbappen har startats om genom att kontrollera loggströmmen:

az webapp log tail --name unique-webhook-handler-name --resource-group myResourceGroup

När omstarten är klar kan du testa webhookens slutpunkt genom att skicka exempel på REST API-händelser.

import requests
import time
import json

WEBHOOK_URL = "https://<unique-webhook-handler-name>.azurewebsites.net/webhook"
TEST_TIMESTAMP = int(time.time())

headers = {
    "Content-Type": "application/json",
    "Webhook-ID": "test-id",
    "Webhook-Timestamp": str(TEST_TIMESTAMP),
    "Webhook-Signature": "test-signature"
}

payload = {
    "object": "event",
    "id": "test-event",
    "type": "response.completed",
    "created_at": TEST_TIMESTAMP,
    "data": {
        "call_id": "test-call",
        "sip_headers": []
    }
}

try:
    # timeout in seconds (connect timeout, read timeout)
    response = requests.post(WEBHOOK_URL, headers=headers, json=payload, timeout=(5, 30))
    print(f"Status Code: {response.status_code}")
    print(f"Response: {response.text}")
except requests.exceptions.Timeout:
    print("Request timed out!")
except requests.exceptions.RequestException as e:
    print(f"Request failed: {e}")

Anmärkning

Den första testsignaturen godkänns inte. För produktionstestning kan vi utlösa faktiska händelser från Azure OpenAI med giltiga signaturer.

Starta nu loggströmmen för webhookens lyssnarwebbapp om den inte fortfarande körs:

az webapp log tail --name unique-webhook-handler-name --resource-group myResourceGroup

Om du vill testa igen med en giltig signatur gör du ett anrop med svars-API:et med background=True set.

from openai import OpenAI
from azure.identity import DefaultAzureCredential, get_bearer_token_provider

token_provider = get_bearer_token_provider(
    DefaultAzureCredential(), "https://cognitiveservices.azure.com/.default"
)

client = OpenAI(  
  base_url = "https://<YOUR-RESOURCE-NAME>.openai.azure.com/openai/v1/",  
  api_key=token_provider,
)

resp = client.responses.create(
  model="o4-mini",
  input="Write a very long novel about otters in space.",
  background=True,
)

print(resp.status)

Python output:

queued 

Loggström utflöde:

2025-10-24T23:40:23.623889430Z INFO:app:Received a response.completed message

Metodtips för säkerhet

Det är avgörande att säkra slutpunkten för din webhook. Följ dessa rekommendationer:

Signaturverifiering

Kontrollera alltid webhookssignaturen för att säkerställa att begäranden kommer från Azure OpenAI:

try:
    event = client.webhooks.unwrap(request.data, request.headers)
except InvalidWebhookSignatureError:
    return Response("Invalid signature", status=400)

Idempotens egenskap

Webhook-ID Använd rubriken för att förhindra duplicerad bearbetning:

webhook_id = request.headers.get('Webhook-ID')
if webhook_id in processed_webhooks:
    return Response(status=200)  # Already processed

Hantering av tidsgräns

Svara snabbt för att undvika tidsgränser för webhook. Avlasta tung bearbetning till bakgrundstrådar:

def process_call_async(call_data):
    # Your heavy processing here
    pass

# In webhook handler:
threading.Thread(target=process_call_async, args=(event.data,)).start()
return Response(status=200)

Ytterligare rekommendationer

  • Lagra din signeringshemlighet på ett säkert sätt; den visas bara en gång när den skapas.
  • Använd HTTPS för alla webhook-slutpunkter.
  • Skydda och rotera Azure-åtkomsttoken regelbundet.
  • Verifiera inkommande begäranden med hjälp av signeringshemligheten.

Exempel på händelsebearbetning

Här följer några vanliga mönster för bearbetning av webhook-händelser:

Anropsloggning

if event.type == "realtime.call.incoming":
    call_data = {
        'call_id': event.data.call_id,
        'timestamp': event.created_at,
        'from': next((h['value'] for h in event.data.sip_headers if h['name'] == 'From'), None),
        'to': next((h['value'] for h in event.data.sip_headers if h['name'] == 'To'), None)
    }
    # Log to database or file
    log_call(call_data)

Meddelandesystem

if event.type == "realtime.call.incoming":
    # Send notification to monitoring system
    send_notification({
        'type': 'incoming_call',
        'call_id': event.data.call_id,
        'caller': get_caller_from_headers(event.data.sip_headers)
    })

Hantera webhook-slutpunkter

Lista webhook-slutpunkter

curl -X GET https://<YOUR-RESOURCE-NAME>.openai.azure.com/openai/v1/dashboard/webhook_endpoints \
  -H "api-key: $AZURE_OPENAI_API_KEY" \
  -H "Content-Type: application/json"

Uppdatera webhook-slutpunkt

Anmärkning

Uppdatera webhook-egenskaper som namn, URL och registrerade händelsetyper. Signeringshemligheten kan inte uppdateras via den här åtgärden.

curl -X POST https://<YOUR-RESOURCE-NAME>.openai.azure.com/openai/v1/dashboard/webhook_endpoints/<webhook_id> \
  -H "api-key: $AZURE_OPENAI_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "<UpdatedName>",
    "url": "<UpdatedURL>",
    "event_types": [
      "response.completed",
      "response.failed",
      "realtime.call.incoming"
    ]
  }'

Ta bort webhook-slutpunkt

Ta bort en webhook-slutpunkt med hjälp av dess webhook-ID:

curl -X DELETE https://<YOUR-RESOURCE-NAME>.openai.azure.com/openai/v1/dashboard/webhook_endpoints/<webhook_id> \
  -H "api-key: $AZURE_OPENAI_API_KEY" \
  -H "Accept: application/json"

Kända problem och felsökning

Problematik Lösning
Ogiltig signatur Kontrollera att din OPENAI_WEBHOOK_SECRET är korrekt och matchar hemligheten som angavs när du skapade den.
Timeout Behåll webhook-bearbetning under 10 sekunder; använda bakgrundsaktiviteter för tunga åtgärder.
Händelser som saknas Se till att slutpunkten är offentligt tillgänglig och använder HTTPS. Övervaka leveransförsök i Azure-loggar.

Hantera webhook-begäranden på en server

När en händelse inträffar som du prenumererar på får du en POST-begäran med följande struktur:

POST https://my-webhook-handler.azurewebsites.net/webhook
User-Agent: OpenAI/1.0 (+https://platform.openai.com/docs/webhooks)
Content-Type: application/json
Webhook-ID: <webhook_eventid>
Webhook-Timestamp: 1750287078
Webhook-Signature: v1,Sample Signature
{
  "object": "event",
  "id": "evt_685343a1381c819085d44c354e1b330e",
  "type": "realtime.call.incoming",
  "created_at": 1750287018,
  "data": {
    "call_id": "some_unique_id",
    "sip_headers": [
      { 
        "name": "From", 
        "value": "sip:+142555512112@sip.example.com" 
      },
      { 
        "name": "To", 
        "value": "sip:+18005551212@sip.example.com" 
      },
      { 
        "name": "Call-ID", 
        "value": "rtc_9d3d08ea002a4909813d207a592957c4"
      }
    ]
  }
}

Rubrik

  • Webhook-ID: Unik identifierare för idempotens – använd detta för att förhindra duplicerad bearbetning
  • Webhook-Timestamp: Unix-tidsstämpel för leveransförsöket
  • Webhook-Signature: Kryptografisk signatur för att verifiera äktheten från OpenAI

Nyttolast

  • object: Alltid "händelse" för webhook-händelser
  • id: Unik händelseidentifierare
  • type: Händelsetyp (till exempel "realtime.call.incoming")
  • created_at: Unix-tidsstämpel när händelsen skapades
  • data: Händelsespecifika data som innehåller samtalsinformation och SIP-huvuden

Webhook-event-referens

Följande händelsetyper är tillgängliga för webhookregistrering:

Kategori Händelsetyp Description
Svarsevenemang response.completed Svaret har slutförts
response.failed Svaret misslyckades
response.cancelled Svaret avbröts
response.incomplete Svaret är ofullständigt
Batch-händelser batch.completed Batchen har slutförts
batch.failed Batchen misslyckades
batch.cancelled Batch avbröts
batch.expired Batchen har gått ut
Realtidshändelser realtime.call.incoming Inkommande samtalshändelse

Exempel på nyttolast

{
  "object": "event",
  "id": "evt_685343a1381c819085d44c354e1b330e",
  "type": "realtime.call.incoming",
  "created_at": 1750287018,
  "data": {
    "call_id": "some_unique_id",
    "sip_headers": [
      { "name": "From", "value": "sip:+142555512112@sip.example.com" },
      { "name": "To", "value": "sip:+18005551212@sip.example.com" },
      { "name": "Call-ID", "value": "rtc_9d3d08ea002a4909813d207a592957c4" }
    ]
  }
}

Rensa resurser

När du inte längre behöver webhook-servern kan du ta bort Azure Web App och dess associerade resurser.

Ta endast bort webbappen

Så här tar du bort webbappen samtidigt som resursgruppen och andra resurser behålls:

az webapp delete --name unique-webhook-handler-name --resource-group myResourceGroup

Ytterligare föreslagna konfigurationer

  • Implementera korrekt loggning och övervakning för webhook-händelser.
  • Lägg till databaslagring för anropsposter.
  • Konfigurera aviseringar för misslyckade webhook-leveranser.
  • Implementera omprövningslogik för underordnade tjänstanrop.
  • Lägg till autentisering för webhookens slutpunkt om det behövs.