Build serverless APIs with Azure Functions triggers and bindings

func init ai-backend --python          # Python v2 model is the default
cd ai-backend
func new --name classify --template "HTTP trigger"
func start                              # http://localhost:7071/api/classify

curl -X POST http://localhost:7071/api/classify \
  -H "Content-Type: application/json" -d '{"text": "Invoice total 420 USD"}'

# Push a test message to the queue the worker listens on
az servicebus queue message send -g <resource-group> --namespace-name <namespace> \
  -n document-jobs --body '{"job_id": "1", "document_url": "https://..."}'

import azure.functions as func
import json, logging, uuid
from datetime import datetime, timezone

app = func.FunctionApp()

# 1. HTTP API that accepts work and returns 202 (async request-reply)
@app.route(route="process-document", methods=["POST"], auth_level=func.AuthLevel.FUNCTION)
@app.service_bus_queue_output(arg_name="queue_msg", queue_name="document-jobs",
                              connection="ServiceBusConnection")
def accept_document(req: func.HttpRequest, queue_msg: func.Out[str]) -> func.HttpResponse:
    job_id = str(uuid.uuid4())
    queue_msg.set(json.dumps({"job_id": job_id, "document_url": req.get_json()["document_url"]}))
    return func.HttpResponse(json.dumps({"job_id": job_id, "status_url": f"/api/job-status/{job_id}"}),
                             status_code=202, mimetype="application/json")

# 2. Worker: Service Bus trigger + Cosmos DB output binding
@app.service_bus_queue_trigger(arg_name="msg", queue_name="document-jobs",
                               connection="ServiceBusConnection")
@app.cosmos_db_output(arg_name="doc", database_name="ai-results",
                      container_name="classifications", connection="CosmosDBConnection")
def process_document(msg: func.ServiceBusMessage, doc: func.Out[func.Document]) -> None:
    job = json.loads(msg.get_body().decode("utf-8"))
    logging.info("Processing %s (delivery %s)", job["job_id"], msg.delivery_count)
    result = {"category": "invoice", "score": 0.93}   # call your model or AI service here
    doc.set(func.Document.from_dict({
        "id": job["job_id"], **result,
        "processed_at": datetime.now(timezone.utc).isoformat(),
    }))

# 3. Embedding job from the original guide: Service Bus trigger + Blob output
@app.service_bus_queue_trigger(arg_name="msg", queue_name="embed-jobs",
                               connection="ServiceBusConnection")
@app.blob_output(arg_name="out", path="results/{rand-guid}.json", connection="AzureWebJobsStorage")
def process_embedding(msg: func.ServiceBusMessage, out: func.Out[str]):
    payload = json.loads(msg.get_body().decode("utf-8"))
    logging.info("Processing document %s", payload["documentId"])
    out.set(json.dumps({"documentId": payload["documentId"], "status": "embedded"}))

# 4. Event Grid push handler (validation handshake handled by the runtime)
@app.event_grid_trigger(arg_name="event")
def on_blob_created(event: func.EventGridEvent):
    logging.info("%s %s", event.event_type, event.get_json().get("url"))

# 5. Timer: refresh embeddings every night at 02:00 UTC
@app.timer_trigger(schedule="0 0 2 * * *", arg_name="timer", run_on_startup=False)
def nightly_refresh(timer: func.TimerRequest) -> None:
    logging.info("Past due: %s", timer.past_due)

{
  "version": "2.0",
  "extensionBundle": {
    "id": "Microsoft.Azure.Functions.ExtensionBundle",
    "version": "[4.*, 5.0.0)"
  },
  "extensions": {
    "serviceBus": { "maxConcurrentCalls": 1, "maxAutoLockRenewalDuration": "00:05:00" }
  }
}