Appearance

Flask — full setup ​

Complete Flask integration: route decorator, OpenAI, requests wrapper, shutdown.

Install ​

bash
pip install amlexia flask openai requests

Environment ​

env
AMLEXIA_SDK_KEY=am_...
AMLEXIA_INGEST_URL=https://ingest.amlexia.com
AMLEXIA_ENVIRONMENT=production
FLASK_APP=app:app
OPENAI_API_KEY=sk-...

Application (app.py) ​

python
import os
import time
import atexit

from flask import Flask, g, jsonify, request
from openai import OpenAI
import requests

from amlexia import AmlexiaClient
from amlexia.flask_integration import amlexia_track
from amlexia.http_wrap import wrap_requests_session
from amlexia.openai_helper import track_openai_completion

client = AmlexiaClient.from_env()
http = wrap_requests_session(client, requests.Session())
openai = OpenAI(http_client=http)

app = Flask(__name__)
atexit.register(client.shutdown)


@app.get("/health")
@amlexia_track(client)
def health():
    return jsonify({"ok": True})


@app.get("/api/users/<int:user_id>")
@amlexia_track(client)
def get_user(user_id: int):
    return jsonify({"id": user_id})


@app.post("/webhooks/stripe")
@amlexia_track(client)
def stripe_webhook():
    return jsonify({"received": True})


@app.post("/api/chat")
@amlexia_track(client)
def chat():
    body = request.get_json(force=True)
    start = time.time()
    try:
        completion = openai.chat.completions.create(
            model="gpt-4o-mini",
            messages=body["messages"],
        )
        track_openai_completion(
            client,
            model=completion.model,
            status_code=200,
            latency_ms=int((time.time() - start) * 1000),
            usage=completion.usage,
        )
        return jsonify(completion.model_dump())
    except Exception as e:
        g.amlexia_status = 500
        client.track(
            endpoint="POST /api/chat",
            method="POST",
            status_code=500,
            latency_ms=int((time.time() - start) * 1000),
            provider="openai",
            error_message=str(e)[:500],
        )
        raise


if __name__ == "__main__":
    app.run(port=int(os.environ.get("PORT", 5000)), debug=False)

@amlexia_track behavior ​

  • Wraps the view; measures latency
  • Normalizes path (/users/42 → /users/:id)
  • On exception: status 500 + re-raises
  • Set g.amlexia_status before return to override status (e.g. 201)

Webhook flag (optional) ​

For clearer provider charts:

python
client.track(
    endpoint="POST /webhooks/stripe",
    method="POST",
    status_code=200,
    latency_ms=latency_ms,
    provider="stripe",
    is_webhook=True,
)

Verify ​

bash
amlexia health
flask run
curl http://127.0.0.1:5000/health