FastAPI × htmx × Azure OpenAI Service で作るチャットアプリ

この記事では、FastAPI + htmx + Azure OpenAI Service を組み合わせて実装した、シンプルかつリアルタイム（ストリーミング）対応のチャットアプリをご紹介します。

「フロントエンドはできるだけ薄くしたい」「社内ツールやプロトタイプを素早く作りたい」
そんな用途を想定した構成です。

以下の記事を拝見して、作成してみました。

「FastAPI + htmxが最強説」- AIエンジニアがモック作るならReactは不要、Streamlitも捨てよう

この記事で分かること

• FastAPI と htmx を組み合わせた最小構成のチャットUI
• Azure OpenAI Service の Chat Completions（streaming）の使い方
• htmx SSE（Server-Sent Events）拡張によるトークン単位のリアルタイム描画

全体構成

.
├── main.py            # FastAPI アプリ本体
├── templates/
│   ├── index.html     # 画面全体
│   └── message.html   # チャット1件分のテンプレート
├── static/
│   └── style.css      # スタイル
└── .env               # Azure OpenAI 設定

フロントエンドは HTML + htmx のみ。

React や Vue は一切使っていません。

コード

main.py

from __future__ import annotations

import html
import os
import threading
import uuid
from typing import Dict, List

from dotenv import load_dotenv
from fastapi import FastAPI, Form, Request
from fastapi.responses import HTMLResponse, StreamingResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from openai import AzureOpenAI

load_dotenv(override=True)

app = FastAPI()
app.mount("/static", StaticFiles(directory="static"), name="static")
templates = Jinja2Templates(directory="templates")

# -----------------------------
# In-memory session store
# -----------------------------
_sessions_lock = threading.Lock()
_sessions: Dict[str, List[dict]] = {}

# stream_id -> session_id mapping
_streams_lock = threading.Lock()
_streams: Dict[str, str] = {}


def _get_session_id(request: Request) -> str:
    sid = request.cookies.get("sid")
    if sid:
        return sid
    return str(uuid.uuid4())


def _get_session_messages(session_id: str) -> List[dict]:
    with _sessions_lock:
        return _sessions.setdefault(session_id, [])


def _render_message(message: dict) -> str:
    template = templates.env.get_template("message.html")
    return template.render(message=message)


def _azure_client() -> AzureOpenAI:
    endpoint = os.getenv("AZURE_OPENAI_ENDPOINT", "").strip()
    api_key = os.getenv("AZURE_OPENAI_API_KEY", "").strip()
    api_version = os.getenv("AZURE_OPENAI_API_VERSION", "2024-02-15-preview").strip()
    if not endpoint or not api_key:
        raise RuntimeError("Azure OpenAI endpoint or API key is missing.")
    return AzureOpenAI(azure_endpoint=endpoint, api_key=api_key, api_version=api_version)


def _build_messages(history: List[dict]) -> List[dict]:
    messages: List[dict] = []
    system_prompt = os.getenv("SYSTEM_PROMPT", "").strip()
    if system_prompt:
        messages.append({"role": "system", "content": system_prompt})
    messages.extend(history)
    return messages


def _chat_completion_stream(history: List[dict]):
    deployment = os.getenv("AZURE_OPENAI_DEPLOYMENT", "").strip()
    if not deployment:
        raise RuntimeError("AZURE_OPENAI_DEPLOYMENT is missing.")
    client = _azure_client()
    # stream=True enables token streaming
    return client.chat.completions.create(
        model=deployment,
        messages=_build_messages(history),
        stream=True,
    )


@app.get("/", response_class=HTMLResponse)
async def index(request: Request) -> HTMLResponse:
    session_id = _get_session_id(request)
    history = _get_session_messages(session_id)
    response = templates.TemplateResponse(
        "index.html",
        {"request": request, "messages": history},
    )
    if not request.cookies.get("sid"):
        response.set_cookie("sid", session_id, httponly=True, samesite="lax")
    return response


@app.post("/chat", response_class=HTMLResponse)
async def chat(request: Request, message: str = Form(...)) -> HTMLResponse:
    """
    Returns immediately:
      - renders the user message
      - renders an empty assistant placeholder with a unique stream_id
      - starts SSE connection to /chat/stream/{stream_id}
    """
    session_id = _get_session_id(request)
    history = _get_session_messages(session_id)

    user_text = message.strip()
    if not user_text:
        return HTMLResponse("")

    # Save user message right away
    user_msg = {"role": "user", "content": user_text}
    history.append(user_msg)

    # Create stream for assistant response
    stream_id = str(uuid.uuid4())
    with _streams_lock:
        _streams[stream_id] = session_id

    # IMPORTANT:
    # message.html must render assistant placeholder with:
    #   <div id="assistant-content-{{ message.stream_id }}"></div>
    assistant_placeholder = {"role": "assistant", "content": "", "stream_id": stream_id}

    html_out = (
        _render_message(user_msg)
        + _render_message(assistant_placeholder)
        + f'<div id="sse-{stream_id}" hx-ext="sse" sse-connect="/chat/stream/{stream_id}" sse-swap="message" sse-close="done"></div>'

    )

    response = HTMLResponse(html_out)
    if not request.cookies.get("sid"):
        response.set_cookie("sid", session_id, httponly=True, samesite="lax")
    return response


@app.get("/chat/stream/{stream_id}")
def chat_stream(stream_id: str):
    """
    Streams assistant tokens via SSE.

    Each SSE 'data:' payload is an HTML snippet that uses hx-swap-oob="beforeend"
    to append content into:
      #assistant-content-{stream_id}
    """

    with _streams_lock:
        session_id = _streams.get(stream_id)

    if not session_id:
        # Unknown stream id: return an empty stream
        return StreamingResponse(iter(()), media_type="text/event-stream")

    history = _get_session_messages(session_id)

    def event_gen():
        assistant_accum = ""
        try:
            stream = _chat_completion_stream(history)

            for evt in stream:
                # Azure/OpenAI streaming: delta content appears here
                delta = ""
                try:
                    delta = evt.choices[0].delta.content or ""
                except Exception:
                    delta = ""

                if not delta:
                    continue

                assistant_accum += delta

                # Make HTML-safe; convert newlines to <br/>
                safe = html.escape(delta).replace("\n", "<br/>")

                # OOB append into the placeholder
                payload = (
                    f'<div id="assistant-content-{stream_id}" '
                    f'hx-swap-oob="beforeend">{safe}</div>'
                )
                yield f"data: {payload}\n\n"

            # Persist the full assistant message to history at the end
            with _sessions_lock:
                history.append({"role": "assistant", "content": assistant_accum})

            # 接続要素をDOMから消す（再接続ループが止まる）
            yield f'data: <div id="sse-{stream_id}" hx-swap-oob="delete"></div>\n\n'

            # htmx-sse に「終わった」と通知して閉じさせる
            yield "event: done\ndata: [DONE]\n\n"

        except Exception as exc:  # noqa: BLE001
            safe = html.escape(f"\n\nError: {exc}").replace("\n", "<br/>")
            payload = (
                f'<div id="assistant-content-{stream_id}" '
                f'hx-swap-oob="beforeend">{safe}</div>'
            )
            yield f"data: {payload}\n\n"

            # 接続要素をDOMから消す（再接続ループが止まる）
            yield f'data: <div id="sse-{stream_id}" hx-swap-oob="delete"></div>\n\n'

            # htmx-sse に「終わった」と通知して閉じさせる
            yield "event: done\ndata: [DONE]\n\n"

        finally:
            with _streams_lock:
                _streams.pop(stream_id, None)

    return StreamingResponse(event_gen(), media_type="text/event-stream")

index.html

<!DOCTYPE html>
<html lang="ja">
  <head>
    <meta charset="utf-8" />
    <meta name="viewport" content="width=device-width, initial-scale=1" />
    <title>Azure OpenAI HTMX Chat</title>
    <link rel="stylesheet" href="/static/style.css" />
    <script src="https://unpkg.com/htmx.org@1.9.12"></script>
    <script src="https://unpkg.com/htmx.org@1.9.12/dist/ext/sse.js"></script>
  </head>
  <body>
    <main class="app-shell">
      <header class="app-header">
        <div>
          <p class="eyebrow">FastAPI + HTMX</p>
          <h1>Azure OpenAI Chat</h1>
        </div>
        <p class="subtitle">Minimal chat UI backed by Azure OpenAI.</p>
      </header>
      <section id="chat-log" class="chat-log">
        {% for message in messages %}
          {% include "message.html" %}
        {% endfor %}
      </section>
      <form
        id="chat-form"
        class="chat-form"
        hx-post="/chat"
        hx-target="#chat-log"
        hx-swap="beforeend"
        hx-on="htmx:afterRequest:this.reset()"
      >
        <input
          type="text"
          name="message"
          placeholder="Ask something..."
          autocomplete="off"
          required
          hx-on="keydown:
            if (event.key === 'Enter' && (event.ctrlKey || event.metaKey)) {
              event.preventDefault();
              this.form.requestSubmit();
            }
          "
          />
        <button type="submit">Send</button>
      </form>
    </main>
  </body>
</html>

message.html

<div class="message {{ message.role }}">
  <div class="bubble">
    {% if message.role == "assistant" and message.get("stream_id") %}
      <p class="message-text" id="assistant-content-{{ message.stream_id }}"></p>
    {% else %}
      <p class="message-text">{{ message.content }}</p>
    {% endif %}
  </div>
</div>

style.css

@import url("https://fonts.googleapis.com/css2?family=Space+Grotesk:wght@400;600&family=Source+Code+Pro:wght@400&display=swap");

:root {
  --bg: #0f141c;
  --bg-accent: #1b2533;
  --panel: rgba(255, 255, 255, 0.04);
  --text: #f8fafc;
  --muted: #b4bcc8;
  --user: #6ee7b7;
  --assistant: #93c5fd;
  --accent: #f97316;
}

* {
  box-sizing: border-box;
}

body {
  margin: 0;
  min-height: 100vh;
  color: var(--text);
  background: radial-gradient(circle at top, #1f2a44 0%, #0f141c 60%, #0a0f14 100%);
  font-family: "Space Grotesk", "Trebuchet MS", sans-serif;
  display: flex;
  align-items: stretch;
  justify-content: center;
  padding: 32px 16px 48px;
}

.app-shell {
  width: min(900px, 100%);
  display: flex;
  flex-direction: column;
  gap: 24px;
  background: var(--panel);
  border: 1px solid rgba(255, 255, 255, 0.08);
  border-radius: 24px;
  padding: 28px;
  backdrop-filter: blur(14px);
  box-shadow: 0 24px 60px rgba(0, 0, 0, 0.35);
  animation: rise 0.6s ease-out;
}

.app-header {
  display: flex;
  align-items: flex-end;
  justify-content: space-between;
  gap: 24px;
  border-bottom: 1px solid rgba(255, 255, 255, 0.08);
  padding-bottom: 16px;
}

.eyebrow {
  text-transform: uppercase;
  letter-spacing: 0.2em;
  font-size: 11px;
  color: var(--muted);
  margin: 0 0 8px;
}

h1 {
  margin: 0;
  font-size: clamp(28px, 4vw, 42px);
  letter-spacing: 0.02em;
}

.subtitle {
  margin: 0;
  color: var(--muted);
  max-width: 280px;
}

.chat-log {
  display: flex;
  flex-direction: column;
  gap: 16px;
  min-height: 240px;
  padding-right: 8px;
}

.message {
  display: flex;
  align-items: flex-start;
}

.message.user {
  justify-content: flex-end;
}

.message.assistant {
  justify-content: flex-start;
}

.bubble {
  max-width: min(640px, 90%);
  padding: 14px 18px;
  border-radius: 18px;
  font-size: 15px;
  line-height: 1.5;
  background: rgba(255, 255, 255, 0.08);
  border: 1px solid rgba(255, 255, 255, 0.08);
  box-shadow: 0 10px 30px rgba(0, 0, 0, 0.25);
  animation: fadeIn 0.4s ease-out;
}

.message.user .bubble {
  border-top-right-radius: 6px;
  border-color: rgba(110, 231, 183, 0.4);
}

.message.user .bubble p {
  color: var(--user);
}

.message.assistant .bubble {
  border-top-left-radius: 6px;
  border-color: rgba(147, 197, 253, 0.4);
}

.message.assistant .bubble p {
  color: var(--assistant);
}

.chat-form {
  display: grid;
  grid-template-columns: 1fr auto;
  gap: 12px;
}

.chat-form input {
  width: 100%;
  padding: 14px 16px;
  border-radius: 14px;
  border: 1px solid rgba(255, 255, 255, 0.12);
  background: rgba(15, 23, 42, 0.6);
  color: var(--text);
  font-size: 15px;
  font-family: "Source Code Pro", "Courier New", monospace;
}

.chat-form button {
  padding: 14px 22px;
  border-radius: 14px;
  border: none;
  background: var(--accent);
  color: #0b1118;
  font-weight: 600;
  letter-spacing: 0.02em;
  cursor: pointer;
  transition: transform 0.2s ease, box-shadow 0.2s ease;
}

.chat-form button:hover {
  transform: translateY(-1px);
  box-shadow: 0 12px 24px rgba(249, 115, 22, 0.3);
}

@keyframes fadeIn {
  from {
    opacity: 0;
    transform: translateY(8px);
  }
  to {
    opacity: 1;
    transform: translateY(0);
  }
}

@keyframes rise {
  from {
    opacity: 0;
    transform: translateY(18px) scale(0.98);
  }
  to {
    opacity: 1;
    transform: translateY(0) scale(1);
  }
}

@media (max-width: 720px) {
  .app-shell {
    padding: 22px;
  }

  .app-header {
    flex-direction: column;
    align-items: flex-start;
  }

  .subtitle {
    max-width: none;
  }

  .chat-form {
    grid-template-columns: 1fr;
  }
}

使っている技術と役割

技術	役割
FastAPI	API / SSE サーバー
Jinja2	HTML テンプレート
htmx	フォーム送信・DOM更新
htmx-sse	SSE を DOM に直接反映
Azure OpenAI Service	ChatGPT API

セッション管理

_sessions: Dict[str, List[dict]] = {}

• Cookie に sid（UUID）を保存
• sid ごとに 会話履歴（role, content）をメモリに保持

def _get_session_id(request: Request) -> str:
    sid = request.cookies.get("sid")
    return sid or str(uuid.uuid4())

DB は使わず、プロトタイプ用途に割り切った実装です。

トップページ /

@app.get("/", response_class=HTMLResponse)
async def index(request: Request):

• Cookie から session_id を取得
• これまでの会話履歴をテンプレートに渡す

<section id="chat-log" class="chat-log">
  {% for message in messages %}
    {% include "message.html" %}
  {% endfor %}
</section>

サーバーサイドレンダリング + htmx なので、ページリロードしても会話が復元されます。

メッセージ送信 /chat

@app.post("/chat")
async def chat(request: Request, message: str = Form(...)):

ここが htmx チャットの中核です。

やっていること

1. ユーザー発言を即座に保存
2. Assistant 用の stream_id を発行
3. ユーザー発言 + 空の Assistant プレースホルダを HTML として即返却
4. SSE 接続を開始

<form
  hx-post="/chat"
  hx-target="#chat-log"
  hx-swap="beforeend"
>

→ ページ遷移なしで DOM の末尾に追加されます。

message.html の工夫

{% if message.role == "assistant" and message.get("stream_id") %}
  <p id="assistant-content-{{ message.stream_id }}"></p>
{% else %}
  <p>{{ message.content }}</p>
{% endif %}

• Assistant は最初「空」
• stream_id を使って 後からトークンを流し込む

SSE によるストリーミング /chat/stream/{stream_id}

@app.get("/chat/stream/{stream_id}")
def chat_stream(stream_id: str):

ここで Azure OpenAI の stream=True を使います。

return client.chat.completions.create(
    model=deployment,
    messages=_build_messages(history),
    stream=True,
)

トークンを受け取るたびに

payload = (
  f'<div id="assistant-content-{stream_id}" '
  f'hx-swap-oob="beforeend">{safe}</div>'
)
yield f"data: {payload}\n\n"

• hx-swap-oob="beforeend"
• 指定IDの要素に直接追記

これにより、

ChatGPT の文字が 1 文字ずつ表示される

体験が実現します。

ストリーム終了処理

yield "event: done\ndata: [DONE]\n\n"

さらに、

<div id="sse-{stream_id}" hx-swap-oob="delete"></div>

• SSE 接続用の要素を削除
• 再接続ループを防止

地味ですが重要なポイントです。

JavaScript がほぼ不要な理由

• フォーム送信 → htmx
• DOM 更新 → htmx
• ストリーミング → SSE + htmx-sse

自前 JS は 0 行。

「サーバーが HTML を返す」

という設計です。

この構成が向いているケース

• 社内向け AI チャット
• PoC / プロトタイプ
• 管理画面・業務ツール
• フロントエンドを最小にしたいチーム

逆に、

• 大規模 SPA
• 高度な状態管理

には React 等の方が向いているかと思います。

まとめ

• FastAPI + htmx だけで 体験の良い AI チャットが作れる
• SSE と hx-swap-oob がストリーミングの鍵
• Azure OpenAI Service との相性も良好

「まず動くものを最速で作る」用途では、使えるかと思います。