AIアプリケーション用のREST APIを構築する方法

Autonomous AI Database Wojtek Pluta 的头像 AI アプリケーション向けの REST API を構築するなら、推論が遅くなったり、リトライが発生したり、実データに接続したりしてもサービスが壊れないようにしたい。薄い FastAPI レイヤーでリクエストを検証・認証し、永続化された予測ジョブを作成して 202 Accepted とステータス確認用 URL を返し、モデルの実行は HTTP リクエストの外側で worker に任せる。本記事では、公開 API をモデルベンダーから切り離す方法、本番経路で Oracle AI Database にジョブ状態を永続化する方法、必要に応じてインタラクティブなレスポンスをストリーミングする方法、テナント単位のアクセス制御を実装する方法を扱う。内容は Oracle AI Developer Hub で継続的にメンテナンスされている proof case に基づいており、SSE プッシュ、検索オーケストレーション、エンタープライズデータアクセスをカバーする。

模式的なスキーマと継続的にメンテナンスされる proof case 本記事のコードは API の境界を説明するためのものだ。記事後半で紹介する継続的にメンテナンスされる proof case では、これらの境界が実際のアプリケーションでどう機能するかを示す。本番環境では Oracle AI Database による永続化、テナント単位のアクセススコープ、独立した worker または永続化キュー、承認済みの ID 基盤、データ保持ポリシー、モニタリング、分散レート制限を使うこと。SQL、VPD、ストリーミング、BackgroundTasks のコード断片は設計の参考であり、そのままコピーしてデプロイできるコードではない。

要点

リソースのライフサイクルを中心に REST API を設計する。ジョブを投入し、202 Accepted を返し、ステータス確認用 URL を用意する。
モデルを呼び出す前に、サイズ制限付きの JSON を Pydantic で検証し、実装から OpenAPI 契約を自動生成する。
モデルベンダーの認証情報はサーバー側にのみ保持し、ベンダー固有の挙動はアダプターの背後に隔離する。
ジョブ状態を永続化し、読み取りは毎回認証済みテナントの範囲に限定し、冪等なリトライに対応する。
推論時間が不定または長時間に及ぶ場合は worker と永続化キューを使う。同期レスポンスは所要時間を予測できる場合に限る。
FastAPI、Oracle AI Database、ORDS、OpenAI Responses API、ストリーミング、キューはそれぞれ独立したアーキテクチャ境界として扱い、役割を分ける。
デプロイ前に、認証、テナント分離、入力拒否、レート制限、ベンダー障害、worker の状態遷移、公開される OpenAPI 契約をテストする。

何を設計し、何を検証するのか

設計するのは小さな FastAPI サービスだ。予測リクエストを受け取り、呼び出し元を認証し、Location レスポンスヘッダー付きの 202 Accepted を返し、HTTP リクエストの外側で処理を行い、最終状態を GET /v1/predictions/{id} で公開する。同じサービスが /docs と /openapi.json を自動生成する。 Developer Hub のセクションにある 3 つの継続的にメンテナンスされる proof case を使い、自分のアプリケーションにとって本当に重要な境界を検証しよう。インタラクティブなストリーミング出力、検索とオーケストレーション、あるいは ID を意識したエンタープライズデータアクセスのいずれかだ。記事の残りでは設計チェックリストを示す。軽量なルーティング、明確なリソース契約、ベンダーアダプター、非同期タスク、ストリーミング出力、ID 認証、ガバナンスの効いた永続状態だ。proof case では、これらの要素が継続的にメンテナンスされるアプリケーションの中でどう連携するかを示す。

AI アプリケーション向け REST API を構築するベストプラクティスとは

明確なリソース契約を使い、モデルを呼び出す前に入力を検証し、ベンダーの認証情報はサーバー側に保持し、同期か非同期かを意識的に選び、ジョブ状態を永続化し、読み取りを認証済みテナントの範囲に限定し、OpenAPI と安全なエラーメッセージを公開する。失敗経路のテストは、モデルの成功レスポンスのテストと同じくらい真剣に行うこと。推奨する実装は、5 つの境界を通じてこれらのプラクティスを実現する。

**HTTP 境界：**FastAPI、Pydantic、OpenAPI、認証、レート制限、リクエスト ID。
タスク境界：POST /v1/predictions が永続リソースを作成し、202 Accepted を返す。
**Worker 境界：**独立したプロセスがタスクを取得し、モデルアダプターを呼び出す。
**データ境界：**Oracle AI Database がリクエストのライフサイクルを保存し、テナントスコープを適用する。
**ベンダー境界：**モデル統合は差し替え可能で、ベンダーの認証情報をクライアントに露出させてはならない。

AI アプリケーション向けの薄い REST API レイヤーをどう設計するか

REST API は薄く保つ。HTTP のエッジで認証と検証を済ませ、job リソースを作成または読み取り、モデル呼び出し、検索、永続状態は専用の境界に任せる。ルートにベンダー固有のロジックや長時間実行される推論を持ち込まない。データ層が改めて実行しなければならない認可ルールも置かない。推論は長時間実行される HTTP 関数ではなく、リソースのライフサイクルとして扱う。API は処理を受け取って記録し、worker が推論を実行し、クライアントは安定したステータス URL から生成されたリソースを読み取る。本番経路では、FastAPI がアプリケーション向けの境界を担い、Oracle AI Database が永続化された prediction job の状態を担う。

チャットボット REST API は FastAPI でどのエンドポイントを公開すべきか

チャットボット REST API は、会話リソース、メッセージ、prediction job、オプションのストリームを分離すべきだ。こうすればチャット履歴をアドレス指定でき、遅い推論が可観測になり、ライフサイクル全体を 1 つの /chat リクエストの背後に隠すのではなく、リソースごとに認可を適用できる。

チャットボット風の REST API では、公開するエンドポイントが同じリソースライフサイクルに素直に対応しているべきです。

POST /v1/conversations/{conversation_id}/messages —— メッセージを作成し、基盤となる prediction job をスケジュールする。
GET /v1/conversations/{conversation_id} —— 会話のメタデータと現在の状態を返す。
GET /v1/conversations/{conversation_id}/messages —— 認可済みのメッセージ履歴を返す。
GET /v1/predictions/{id} —— 永続化された job をポーリングし、モデルの応答が準備できるまで待つ。
GET /v1/conversations/{conversation_id}/messages/{message_id}/stream —— 任意で Server-Sent Events（SSE）により増分結果をストリーミングする。
DELETE /v1/conversations/{conversation_id} —— 保持ポリシーに従って会話データを削除する。

クライアントから見た流れは一貫しています。FastAPI が認証を済ませてジョブを作成し、worker が Oracle AI Database 内で永続化された job の状態を更新し、クライアントはステータス URL または認可済みのストリームから応答を読み取ります。

このパターンが役立つのは、モデルの推論がリクエストのタイムアウトを超える可能性があるとき、クライアントが信頼できる状態情報を必要とするとき、あるいは AI アプリケーションがガバナンスの効いた業務データのそばにリクエストコンテキストを保持しなければならないときです。

なぜ本番経路に Oracle AI Database を選ぶのか

Oracle AI Database があれば、本番経路は永続化された prediction job の状態とガバナンスの効いたアプリケーションデータを同じ場所に置けます。テナント単位で作業範囲を区切り、信頼できる状態の読み取りと冪等な記録を支え、API が公開する行のすぐ近くでデータアクセス制御を効かせられます。

検証ケース：API パターンを実際の AI 開発者タスクに当てはめる

REST API のパターンは、開発者が実際に届けようとしている仕事につながって初めて役に立ちます。以下の 3 つの Oracle AI Developer Hub リソースは本チュートリアルの代わりではありません。それぞれが、設計上のどこかの境界を確かめるための実用的な検証ケースです。

タスク：ルーティングを agent にせず、対話的な AI 体験を届ける

POST /v1/chat のプロトタイプから、進捗をストリーミングしながら agent harness をブラウザの外に置く対話型アプリへ進みたいときは、Total Recall を使ってください。FastAPI バックエンドとブラウザアプリで構成され、SSE でストリーミングし、そのうえで周辺の関心事——コンテキスト、検索、記憶、ツール、trace——を独立した層として公開します。

API 設計にとって何を証明するか： ストリーミングは HTTP 境界における配送の仕組みであり、安定したリソースモデル、サーバー側の provider 資格情報、永続状態の必要性をなくすものではありません。プロダクトの要件が「今すぐ役に立つ進捗を見せる」ことであって「後でブロッキング応答を返す」ことではないとき、この検証ケースを使ってください。

すぐ動かす： Python 3.11+、必要な embedding モデルを備えた到達可能な Oracle AI Database、承認済みのモデル provider 資格情報を用意し、appbook のセットアップに従って ./run.sh を実行します。成功のシグナルとして記録されているのは、ローカルのブラウザアプリが使えること、FastAPI サービスが使えること、そしてデータベース、harness、reranker のフォールバックまたは可用性、設定済みモデル provider の状態インジケーターがあることです。ライブ trace とストリーミングチャットで、harness をクライアントに入れずにルーティングが増分出力を届けているかを確認してください。

タスク：検索に支えられた AI API に正しい根拠で答えさせる

API がモデルを呼ぶだけでは済まないときは、From RAG to Agents workshop を使ってください。この workshop はデータのロードと検索から始まり、RAG、agent ツール、マルチ agent オーケストレーション、永続的な会話記憶を経て、研究論文アシスタントを組み上げていきます。 API 設計にとって何を意味するか： 検索とオーケストレーションは、ベンダー非依存のサービス境界の後ろに置くべきです。POST /v1/predictions や POST /v1/chat には引き続き検証、アイデンティティ、状態、レスポンス構造を担わせ、検索・ツール呼び出し・モデル実行の組み立ては worker または agent runtime に任せます。開発者のタスクが「根拠ある検索をアプリの機能にしつつ、内部パイプラインを公開 API として晒さない」ことであるとき、この検証ケースを使ってください。 すぐ動かす： Docker、Python、Jupyter をインストールし、workshop の Oracle AI Database コンテナを起動して依存関係を入れ、workshop/notebook_student.ipynb を開きます。学習パスはデータのロードと検索から始まり、RAG、ツール、オーケストレーション、会話記憶へと進みます。確かめるべきは、モデルが回答を出せることだけではなく、どの検索とオーケストレーションの境界がその回答を生んだのかを追えることです。

タスク：アイデンティティとアクセスポリシーを迂回せずに企業データを開放する

API が異なるデータを見せるべきユーザーにサービスを提供するときは、Enterprise Data Agent workshop を使ってください。notebook と稼働中のアプリを組み合わせ、同じデータベースに対するアイデンティティ認識の行レベル・列レベルポリシー、検索、記憶、ツール、そしてライブチャット UI を実演します。 API 設計にとって何を意味するか： ルートでの認証は必要ですが、十分ではありません。FastAPI で呼び出し元のアイデンティティを導き、承認されたスキーマを通じてそのアイデンティティをデータベースセッションに持ち込み、検索・会話の読み取り・ツールアクセスがすべて同じスコープを適用するようにします。開発者のタスクが「企業データに AI の能力を加えつつ、ガバナンスの効かない第二のアクセス経路を作らない」ことであるとき、この検証ケースを使ってください。 すぐ動かす： workshop の GitHub Codespaces 経路なら設定の手間が最小です。ローカルなら Docker、Python、Node.js の構成に従って notebook と付随アプリを起動します。確かめるべきは、データアクセスがアイデンティティに応じて変わることです。同じアプリと agent ワークフローが、チャットリクエストで渡されたスコープのない識別子を信頼するのではなく、現在のアイデンティティとデータポリシーに縛られるべきです。

証拠の境界： これらのリンクは Developer Hub の特定の commit に固定されているため、コードと設定手順はいつでも確認できます。示しているのは動くアプリにおける API 境界です。Total Recall は FastAPI と SSE を、From RAG to Agents は検索とオーケストレーションを、Enterprise Data Agent はアイデンティティ認識アクセスを示します。あなたがデプロイする構成のレイテンシ、スループット、可用性、セキュリティをベンチマークしたものではありません。公開前に、自分のモデル、データ、アイデンティティプロバイダー、トラフィックでこれらの指標を測ってください。

公開 API を AI ベンダーから独立させるには？

アプリケーション境界で、ベンダー非依存のリクエスト・結果・エラーの契約を定義します。各モデル SDK はアダプターの後ろに置き、アダプターが内部リクエストをベンダーへの呼び出しに翻訳し、結果を API の安定した schema にマッピングし直します。クライアントがベンダーのレスポンスオブジェクト、ベンダー固有のエラーペイロード、資格情報を受け取ることはありません。 Copied to ClipboardError: Could not Copy

from typing import Protocol

class ModelResult(BaseModel):
 text: str
 provider_request_id: str | None = None

class ModelAdapter(Protocol):
 def generate(self, *, prompt: str, temperature: float) -> ModelResult: ...

アダプターはサーバー側の設定で選びます。こうすればベンダーやモデルが変わっても、HTTP 契約、タスク schema、認可、テストは安定したままです。

開発中、FastAPI の `/docs` と `/openapi.json` は何の役に立つのか？

FastAPI はリクエストとレスポンスのモデルから、対話的な /docs ページと機械可読な /openapi.json 契約を生成します。これらを使って、認証付きの開発リクエストを試し、検証ルールを確認し、フロントエンドやプラットフォームチームと正確な契約を共有し、クライアントを生成し、契約テストを追加します。これらが改善するのは開発者のフィードバックです。認証、認可、レート制限、ゲートウェイ制御の代わりにはなりません。

デプロイ前に、/openapi.json が意図して公開しているフィールドとレスポンスだけを露出していることを確認する。/docs を開発環境以外でも使い続けるかどうかを決め、デプロイの運用ポリシーに沿って保護する。

AI 推論は同期と非同期のどちらにすべきか

同期レスポンスを使うのは、完了時間が予測でき、API 契約に沿っている場合だけだ。推論にどれだけ時間がかかるか読めない場合、リトライが必要な場合、あるいは HTTP リクエストから独立して監査する必要がある場合は、非同期のジョブリソースを使う。

AI のレスポンスが予測可能な HTTP の締め切り内に用意できるとは限らない。ジョブリソースがあれば、API は有効な作業を受け取ったことを素早く確認でき、リクエストスレッドが推論で塞がるのを避けられる。クライアント側にも、ポーリングできる安定したリソースを渡せる。

次のリソースを使う。

エンドポイント	用途
`POST /v1/predictions`	プロンプトを検証して予測ジョブを投入し、`202 Accepted` を返す。
`GET /v1/predictions/{id}`	ジョブのキュー待ち、実行中、成功、失敗の状態を読む。
`GET /healthz`	プラットフォームのヘルスチェックが API プロセスの可用性を確認するために使う。

/generateText のような URL を主要なインターフェースとして扱わないこと。予測はライフサイクルを持つリソースであり、API はそのライフサイクルを見えるようにすべきだ。

client → FastAPI → prediction_jobs in Oracle AI Database
 ↓
 durable worker → approved model adapter
 ↓
client ← GET /v1/predictions/{id}

インプロセスのバックグラウンドタスクはローカルのプロトタイプには十分だ。本番では worker プロセスを別に動かし、永続化されたキューか同等のプラットフォームサービスを使う。そうすれば、受け付けた作業が Web プロセスの再起動をまたいで生き残る。

FastAPI のチャットにストリーミングレスポンスを追加するには？

クライアントがインクリメンタルなレスポンス内容から利益を得られるなら、Server-Sent Events を使う。ストリーミングは出力方法の選択にすぎず、永続化されたジョブ状態の代わりにはならない。UI が部分的な内容を受け取る場合でも、正式な結果を永続化し、状態リソースを提供すること。

import asyncio
import json

from fastapi import Request
from fastapi.responses import StreamingResponse

@app.get("/v1/conversations/{conversation_id}/messages/{message_id}/stream")
async def stream_message(request: Request, conversation_id: str, message_id: str):
 async def events():
 try:
 async for chunk in worker.stream(message_id):
 if await request.is_disconnected():
 return
 yield "event: message\\ndata: " + json.dumps({"delta": chunk}) + "\\n\\n"
 yield "event: done\\ndata: {}\\n\\n"
 except asyncio.CancelledError:
 return
 except Exception:
 yield 'event: error\\ndata: {"code":"stream_failed"}\\n\\n'

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

イベントの種類とペイロードの構造は、公開契約の一部として定義する。各チャンクは JSON にシリアライズし、モデルの生の出力をそのまま SSE フレームに連結しないこと。そのうえで、クライアントの切断を検出し、キャンセルを処理し、プロバイダーのエラーはサーバー側にとどめ、最終的に永続化する結果を確定して後から取得できるようにする。

FastAPI で AI REST API を構築するには？

境界のある Pydantic リクエスト・レスポンスモデルを定義し、未知のフィールドを拒否し、FastAPI の依存性で認証し、明確なステータスコードを返し、OpenAPI 契約は FastAPI に実装から生成させる。

次の2つのコードは、リクエストモデルと投入の境界を示している。自分の ID プロバイダー、ジョブストア、worker ランタイムに合わせて適応させるパターンとして読んでほしい。メンテナンスされている検証用例は、周辺アプリケーションの挙動を示している。

ステップ1：境界のあるリクエスト schema を定義する

Pydantic で未知のフィールドを拒否し、リクエストが高コストなモデル呼び出しに到達する前に、入力を制限する。

class PredictionRequest(BaseModel):
 model_config = ConfigDict(extra="forbid")

 prompt: Annotated[str, Field(min_length=5, max_length=8_000)]
 temperature: Annotated[float, Field(default=0.2, ge=0.0, le=1.0)]

これらの制限はアプリケーション側の判断であり、汎用のデフォルト値ではない。モデルのコンテキストウィンドウ、不正利用ポリシー、想定される payload の形に応じて設定する。

ステップ2：タスクを投入して `202 Accepted` を返す

エンドポイントはまず予測タスクを作成し、次に実行をスケジュールする。Location ヘッダーと JSON の状態 URL を返す。

from fastapi import BackgroundTasks, Depends, Header, Request, status
from fastapi.responses import JSONResponse

@app.post("/v1/predictions")
async def create_prediction(
 payload: PredictionRequest,
 background_tasks: BackgroundTasks,
 request: Request,
 identity: VerifiedIdentity = Depends(require_identity),
 idempotency_key: str | None = Header(default=None, alias="Idempotency-Key"),
):
 job_id = str(uuid4())
 status_url = f"/v1/predictions/{job_id}"
 repository.create(
 job_id=job_id,
 tenant_id=identity.tenant_id,
 client_id=identity.user_id,
 prompt=payload.prompt,
 temperature=payload.temperature,
 request_id=request.state.request_id,
 idempotency_key=idempotency_key,
 )
 background_tasks.add_task(worker.process_once)
 return JSONResponse(
 status_code=status.HTTP_202_ACCEPTED,
 headers={"Location": status_url},
 content={"id": job_id, "status": "queued", "status_url": status_url},
 )

API がタスクを受理したがまだ完了していない場合は、202 Accepted を使う。2秒のような固定の時間しきい値を選ばないこと。結果がリクエスト契約の中で確実に完了できないなら、job リソースを使う。

この BackgroundTasks の例はあくまで示意であり、本番のランタイムのパターンではない。重く、リトライが必要で、プロセスの再起動をまたいで生き残る必要があるタスクには、独立した worker と永続化されたキューを使う。FastAPI もバックグラウンドタスクのガイドで同じ区別をしている。

本番環境で Oracle AI Database を使って AI タスクを永続化するには？

タスクのリクエスト、テナントのスコープ、ライフサイクル状態、安全な結果またはエラーコード、リクエスト相関 ID、冪等性フィンガープリントを Oracle AI Database に保存する。トランザクションは短く保ち、コネクションプール経由でバインド変数を使う。

ステップ3：予測状態を Oracle AI Database に永続化する

サービスはリクエスト、クライアントのスコープ、状態、結果、安全なエラーコード、リクエスト ID を prediction_jobs テーブルに記録する。こうして API の状態がアプリケーションデータの近くに置かれ、そのデータがモデルリクエストの形を決めたり、誰が結果を取得できるかを決めたりする。

CREATE TABLE prediction_jobs (
 id VARCHAR2(36) PRIMARY KEY,
 tenant_id VARCHAR2(255) NOT NULL,
 client_id VARCHAR2(255) NOT NULL,
 status VARCHAR2(16) NOT NULL,
 prompt CLOB NOT NULL,
 temperature NUMBER(3,2) NOT NULL,
 generated_text CLOB,
 error_code VARCHAR2(128),
 request_id VARCHAR2(64) NOT NULL,
 idempotency_key VARCHAR2(255),
 request_fingerprint VARCHAR2(64),
 created_at TIMESTAMP WITH TIME ZONE NOT NULL,
 updated_at TIMESTAMP WITH TIME ZONE NOT NULL,
 CONSTRAINT prediction_jobs_idempotency_uq
 UNIQUE (tenant_id, idempotency_key)
);

データベース呼び出しにはバインド変数を使う。本番の実装では oracledb.create_pool を使い、接続を取得するのはタスクの作成、クレーム、完了といった短いトランザクションの中だけにする。

ステップ4：worker にモデル推論を処理させる

worker はキュー待ちのタスクを1つクレームし、モデルアダプターを呼び出し、結果または安全なエラーコードを記録する。

def process_once(self) -> bool:
 job = self._repository.claim_next()
 if not job:
 return False
 try:
 result = self._model.generate(prompt=job.prompt, temperature=job.temperature)
 self._repository.mark_succeeded(job.id, result)
 except Exception:
 self._repository.mark_failed(job.id, "model_inference_failed")
 return True

プロバイダーの生のエラーをそのまま公開レスポンスに返さないこと。内部で障害を分析するために残すのは、マスク済みで相関可能な診断情報だけにする request_id。

OpenAI SDK で最小限の FastAPI チャットインターフェースを構築するには？

最小限の同期プロトタイプなら、メッセージを検証し、サーバー側で OpenAI SDK を呼び出し、独自に定義した安定した JSON 構造を返せばよい。このルートはシンプルに保つ。ここで示すのはリクエストの境界であり、長期間動かせる本番向けのジョブシステムではない。

已复制到剪贴板复制失败

import os
from fastapi import FastAPI
from openai import AsyncOpenAI
from pydantic import BaseModel, Field

app = FastAPI()
client = AsyncOpenAI(api_key=os.environ["OPENAI_API_KEY"])

class ChatRequest(BaseModel):
 message: str = Field(min_length=1, max_length=8_000)

@app.post("/v1/chat")
async def create_chat(payload: ChatRequest):
 response = await client.responses.create(
 model=os.environ["OPENAI_MODEL"],
 input=payload.message,
 )
 return {"output_text": response.output_text}

推論が可変でコストが高く、監査が必要なら、SDK の直接呼び出しを前述の 202 予測ジョブパターンに置き換える。このルートをローカルのプロトタイプ以外で使う前に、セキュリティの章で説明した認証付き Depends パターンを追加する。プロバイダーの生のレスポンスオブジェクトを返してはいけないし、プロバイダーの資格情報をクライアントに露出してもいけない。

REST API を OpenAI Responses API に接続するには？

モデルプロバイダーはサーバー側のアダプターの裏に隠す。シークレット管理サービスか環境変数から OPENAI_API_KEY を読み込み、OPENAI_MODEL を明示的に設定し、worker 内で Responses API を呼び出し、問題調査のためにプロバイダーのリクエスト ID を記録し、自前 API の安定したレスポンス契約だけを返す。

OpenAI API key はどう安全に保存・読み込みすべきか？

OpenAI API key は承認済みのシークレット管理サービスに保管し、実行時にサーバーまたは worker へ注入する。環境変数はプロセス単位の受け渡し方法として適しているが、キーの値はソースコード、Docker イメージ、notebook、ブラウザの JavaScript、コミット済みの .env ファイルではなく、管理されたシークレットストアから取得すべきだ。

環境とワークロード ID ごとに別々のキーを使う。
そのキーを読める人とコンポーネントを限定する。
必須のキーがなければ起動を失敗させる。クライアントのリクエストからキーを受け取ってはいけない。
ログ、trace、例外、ヘルスチェックのレスポンス、設定のエクスポートにキーを出力しない。
組織の資格情報ポリシーに従ってキーをローテーションし、漏洩が疑われる場合は直ちにローテーションする。

キーはサーバー側で読み込み、そのまま provider クライアントに渡す：

Copied to ClipboardError: Could not Copy

import os
from openai import OpenAI

api_key = os.environ["OPENAI_API_KEY"]
openai_client = OpenAI(api_key=api_key)

ここに示したのは provider 境界の書き方の例であり、そのままダウンロードして使える実装や設定ではない。現在の OpenAI API リファレンスは Responses をモデルへのリクエストを直接発行する API として説明し、サーバー側の bearer 資格情報を要求し、本番で使う前にリクエスト ID を記録してレート制限を確認することを推奨している。

provider のキーをブラウザの JavaScript に入れてはいけないし、リクエストボディからキーを受け取ってもいけない。provider の生のエラーではなく、安定した内部エラーを返す。API 自身の呼び出し元認証とモデル provider の認証は、別々の信頼境界だ。

AI REST API の認証と保護はどう行うか？

高コストな処理に入る前に認証を済ませ、検証済みの資格情報からテナント ID を解決し、ジョブの読み取りと更新のたびに認可チェックを課し、リクエストレートを制限し、リクエストボディのサイズを制限し、冪等なリトライに対応し、provider のエラーをマスクし、prompt と生成結果の保持ポリシーを定義する。

セッションはどう保存し、ユーザー帰属の検証をどう実装するか？

各セッションにサーバー側で導出した tenant_id と owner_id を持たせて保存し、そのセッション識別子をメッセージと予測ジョブに引き継ぐ。読み取り、更新、ストリーミング、削除のすべてに認証済みの帰属スコープを付ける。セッション ID を知っていることは認可を意味しない。

Copied to ClipboardError: Could not Copy

SELECT conversation_id, status, updated_at
FROM ai_conversations
WHERE conversation_id = :conversation_id
 AND tenant_id = :authenticated_tenant_id
 AND owner_id = :authenticated_user_id;

メッセージ履歴を読み込むときや生成内容を返すときにも、同じスコープを課す。多層防御として、Oracle AI Database で承認済みのポリシー（たとえば VPD）を使ってテナント述語を強制し、あるユーザーが別のユーザーのセッションを読み取ったり、ストリーミングで取得したり、更新したり、削除したりできないことをテストする。

チャットボット風のエンドポイントでは、セッションとメッセージを所有者またはテナント ID 付きで保存し、コンテキスト、履歴、生成内容を返すすべてのエンドポイントでこの帰属関係を強制検証する。prompt injection、安全でない生成出力、想定外のツール呼び出し、機密データの漏洩、コスト枯渇も考慮に入れる。これらのリスクは、厳格な schema 検証、ツールのホワイトリスト、最小権限のデータアクセス、タイムアウト、token 予算、マスキング、影響の大きい操作への人手による承認の導入で緩和する。

FastAPI でユーザー認証 `Depends` を追加するには？

推論ジョブを作成したりプロバイダーを呼び出したりする前に、検証済みの identity オブジェクトを注入する。検証に通った JWT、OAuth セッション、または同等の資格情報から tenant_id と user_id を導出する。リクエストボディの権威あるフィールドとして受け取ってはいけない。

Copied to ClipboardError: Could not Copy

from fastapi import Depends, HTTPException, status
from pydantic import BaseModel

class VerifiedIdentity(BaseModel):
 tenant_id: str
 user_id: str

async def require_identity() -> VerifiedIdentity:
 identity = await verify_request_credentials()
 if identity is None:
 raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
 return identity

@app.post("/v1/chat")
async def chat(request: ChatRequest,
 identity: VerifiedIdentity = Depends(require_identity)):
 # Scope reads and writes with identity.tenant_id and identity.user_id.
 return await create_scoped_job(request, identity)

この依存関係はあくまで例だ。verify_request_credentials() はそのデプロイで承認された認証機構に置き換え、同じ identity をデータベースポリシーでも強制する。

認証と認可： 呼び出し元に認証を要求し、各ジョブをテナントまたはユーザーに限定する。不透明な予測 ID は認可ではない。
レート制限： プロセスローカルの防御はデモにしか使えない。マルチインスタンス構成では、API ゲートウェイ、Redis ベースのサービス、または同等のプラットフォーム制御で分散レート制限を実施する。
入力制御： プロンプトの長さを制限し、未知の JSON フィールドを拒否し、メディアアップロードはテキストリクエストとは別に検証する。
冪等性： ネットワーク障害後に POST をリトライしうるクライアントのために、冪等キーを追加する。
可観測性： リクエスト ID を返し、ジョブ ID とテナントに安全な相関データを記録し、キューの遅延、推論時間、エラー、放棄を計測する。
保持ポリシー： prompt と生成レスポンスがいつ期限切れになるか、いつ削除しなければならないかを明確にする。デフォルトで無期限に保持しない。

Oracle AI Database のパターン：テナントスコープとリクエストの関連付け

API はデータベースセッションを開く前に、まず呼び出し元を認証しなければならない。以下はアプリケーションコンテキストと Oracle Virtual Private Database (VPD) および DBMS_APPLICATION_INFO に基づく簡略化した Oracle 疑似コードである。そのままコピーして使える本番用ポリシーではない。信頼できるコンテキスト設定パッケージを用意し、対象のデプロイに対して VPD のライセンスとセキュリティ要件を確認すること。VPD を実装する DBMS_RLS は Oracle AI Database の Enterprise Edition でのみ提供される。このパターンを採用する前に、エディションとデプロイの統制を確認すること。

已复制到剪贴板错误：无法复制

-- The API derives these values from authenticated identity; do not accept a
-- tenant ID or client identity as an untrusted request-body field.
BEGIN
 trusted_ai_api_context.set_identity(
 tenant_id => :authenticated_tenant_id,
 client_id => :authenticated_client_id
 );

 -- Safe correlation only: do not place prompts, tokens, or secrets here.
 DBMS_APPLICATION_INFO.SET_MODULE(
 module_name => 'ai_predictions_api',
 action_name => 'create_prediction'
 );
 DBMS_APPLICATION_INFO.SET_CLIENT_INFO(:request_id);
END;
/

-- A VPD policy function can add this predicate whenever PREDICTION_JOBS is
-- queried or changed. The application query stays simple; the database
-- enforces the current tenant scope.
CREATE OR REPLACE FUNCTION prediction_tenant_predicate(
 object_schema VARCHAR2,
 object_name VARCHAR2
) RETURN VARCHAR2 AS
BEGIN
 RETURN q'[tenant_id = SYS_CONTEXT('AI_API_CTX', 'TENANT_ID')]';
END;
/

BEGIN
 DBMS_RLS.ADD_POLICY(
 object_schema => 'APP_SCHEMA', -- replace with the schema that owns PREDICTION_JOBS
 object_name => 'PREDICTION_JOBS',
 policy_name => 'PREDICTION_TENANT_SCOPE',
 policy_function => 'PREDICTION_TENANT_PREDICATE',
 statement_types => 'SELECT,INSERT,UPDATE,DELETE',
 update_check => TRUE
 );
END;
/

-- Idempotency belongs in durable state, not only in the API process.
-- The PREDICTION_JOBS table definition above already enforces it with
-- UNIQUE (tenant_id, idempotency_key); do not add the constraint twice.

DBMS_APPLICATION_INFO を使って安全なリクエスト ID をデータベース操作に関連付ける。その CLIENT_INFO フィールドは、鍵や機密性の高いプロンプトデータを置く場所ではない。VPD を使うシステムでは、ポリシー関数を保護対象のテーブルに紐付け、Oracle が設定済みのステートメント種別に述語を適用する。SQL の部分はあくまでアーキテクチャ上の指針として扱い、対象のエディション、スキーマ設計、セッションプールの挙動、ID モデルについてレビューを行うこと。

FastAPI、ORDS、ストリーミング、キューのどれを使うべきか

アプリケーション向けの検証とモデルオーケストレーションには FastAPI、ガバナンスの効いたデータベース自身のリソースには ORDS、逐次的な対話出力には SSE、本番環境のワーカー配信には永続化キューを使う。これらは併用できるが、それぞれ異なる境界の問題を解決するものだ。

要件	最適な境界	例またはリソース
短時間で予測可能な推論	同期 REST レスポンス	入力上限とタイムアウトを設定した直接ルート
長時間実行される推論	`202` ジョブリソースとワーカー	前述のジョブリソースとワーカーのパターン
トークン単位の対話	SSE などのストリーミングプロトコル	Total Recall appbook（発展的な学習パスとして）
データベース自身の状態リソース	ORDS	Oracle REST Data Services ドキュメント
検索拡張アプリケーション	独立した検索 API	From RAG to Agents ワークショップ

データベース自身のリソースにガバナンスの効いた REST インターフェースが必要な場合、ORDS は有用だ。ただしモデルオーケストレーションの公開 API 境界として FastAPI を置き換える必要はない。このパターンでは、FastAPI がリクエスト検証とモデルのライフサイクルを担い、既存のデータベース側アプリケーションが本当に必要とする場合に限り、ORDS がジョブ状態データに対して範囲を絞った読み取り専用の統合を提供する。

AI REST API のテストとデプロイ方法

デプロイ前に、公開コントラクト、認証、テナント分離、冪等性、レート制限、ワーカーの状態遷移、プロバイダー障害、OpenAPI 出力をテストする。本番環境ではワーカーを別プロセスで動かし、プロバイダーとデータベースの資格情報はソース管理の外に置く。

まず既存の Total Recall appbook で FastAPI/SSE の境界処理を把握し、次に From RAG to Agents workshop でデータベースベースのオーケストレーションパターンを学ぶ。
境界を定めた Pydantic スキーマと、認証済みの ID 依存関係を使ってリソースコントラクトを実装する。
プロバイダー呼び出しを公開する前に、永続化されたタスク状態、テナント分離、冪等性、ワーカーを追加する。
レビュー済みの repository 実装を通じて、Oracle AI Database の永続化とコネクションプールを接続する。
デプロイ前に、コントラクト、認証、テナント分離、ストリーミング、ワーカー、プロバイダーアダプターのテストをすべて実行する。

OpenAPI コントラクトは /openapi.json に置く。これを使ってクライアント SDK を生成し、コントラクトテストを実行し、統合中にリクエスト/レスポンススキーマを照合する。開発中は FastAPI の /docs ページでエンドポイントと検証済みモデルを確認できるが、どちらのエンドポイントもセキュリティ制御ではない。認可、テナント分離、レート制限は、API ロジックとゲートウェイポリシーで引き続き強制しなければならない。

AI REST API のトラブルシューティング

401 unauthorized： API がレビュー済みの資格情報から ID を取得していること、およびクライアントが必要な資格情報を提示していることを確認する。
422 Unprocessable Entity： プロンプトの長さ、temperature の範囲、JSON のフィールド名、Content-Type: application/json リクエストヘッダーを確認する。
409 idempotency_conflict： 論理リクエストボディが完全に同一の場合にのみ Idempotency-Key を再利用する。リクエスト内容が変わったら新しい key を生成する。
ポーリング時に 404 not_found が発生： 状態クエリは、そのタスクを作成したのと同じ認証済みテナントの範囲に限定する。
再起動後にタスクが消える： 受け付けたタスクを永続化された状態に書き込み、独立したワーカーまたはキューで処理する。
/docs は読み込めるのにリクエストが失敗する： OpenAPI はコントラクトを記述するもので、認証や設定を提供するものではない。リクエストの資格情報と JSON エラーコードを確認する。

本番環境チェックリスト

FAQ

AI アプリケーション向けの REST API はどう作るか？

FastAPI でリクエストを検証・認証し、推論タスクを永続化して 202 Accepted とステータス URL を返し、worker がモデル provider を呼び出す。タスクの状態は Oracle AI Database のような永続データベースに保存し、読み取りは認証済みテナントの範囲に限定する。JSON のステータスとエラー応答は安定した形式で返す。

AI REST API のベストプラクティスとは？

入力の境界を定めた検証、provider 認証をサーバー側で完結させること、レート制限、永続化された非同期タスク、テナントを意識した認可、冪等性、構造化されたエラー、リクエストの相関、保持期間の制御、そして失敗経路のテスト——推論が成功するケースだけをテストするのでは不十分だ。

OpenAI ベースの REST API の認証はどうするか？

サービス境界で API 呼び出し元を認証し、モデル provider への認証はサーバーまたは worker 内で別途行う。provider のキーはシークレットマネージャーか保護された環境変数に置き、モデルを明示的に指定し、安全な provider リクエスト ID を記録する。キーや provider の生のエラーをクライアントに返してはならない。

AI アプリケーションの API には FastAPI と ORDS のどちらを使うべきか？

アプリケーション向けの検証、認証、オーケストレーション、モデルのライフサイクル管理には FastAPI を使う。データベースが所有するリソースにガバナンスの効いた REST インターフェースが必要な場合は ORDS を使う。両者は共存できる。ORDS が FastAPI に取って代わり、対外的なモデルオーケストレーションの境界を担う必要はない。

AI REST API は同期的に結果を返すべきか？

予測可能で短時間で終わる操作ならそれでよい。リクエストのウィンドウ内で確実に完了しない可能性がある場合、あるいは永続的なリトライ、監査、ステータス照会が必要な場合は、タスクリソースと 202 Accepted を使う。

FastAPI は本番環境の AI アプリケーションを支えられるか？

FastAPI は優れた HTTP フレームワークだが、あくまで一層にすぎない。本番環境には永続ストレージ、worker またはキュー、認証、レート制限、可観測性、保持期間の制御が別途必要になる。

AI REST API 向けに、メンテナンスが続いている検証用サンプルはどれを動かすべきか？

FastAPI と SSE なら Total Recall、検索とオーケストレーションなら From RAG to Agents workshop、アイデンティティを意識したガバナンス下のデータアクセスなら Enterprise Data Agent workshop を動かす。優先的に検証したい境界に合うものを選ぶ。

Oracle AI Database は AI API にどう役立つか？

タスクの状態とアプリケーションデータを同じガバナンス下のデータプラットフォームに永続化できる。アプリケーション API は引き続きリクエストの検証とオーケストレーションを担い、その背後で Oracle AI Database が永続化された状態とデータアクセス層を提供する。

次のステップ

メンテナンスが続いているサンプルプロジェクトを選び、本ガイドの API 契約を当てはめる：

Total Recall：FastAPI、SSE、trace、そして agent harness の境界。
From RAG to Agents：検索、ツール、オーケストレーション、会話メモリ。
Enterprise Data Agent：アイデンティティを伴う、ガバナンス下のエンタープライズデータアクセス。

ドキュメント

著者

Wojtek Pluta のアバター

Wojtek Pluta

テクニカルプロダクトマーケティングディレクター

Portable Oracle Select AI Agent Teams：開発から本番まで

Mark Hornick | 読了時間 2 分次の記事

Oracle AI Database で制御された Cursor MCP ワークフローを構築する

Srinidhi Sathyamurthy | 読了時間 2 分

要点

何を設計し、何を検証するのか

AI アプリケーション向け REST API を構築するベストプラクティスとは

AI アプリケーション向けの薄い REST API レイヤーをどう設計するか

チャットボット REST API は FastAPI でどのエンドポイントを公開すべきか

なぜ本番経路に Oracle AI Database を選ぶのか

検証ケース：API パターンを実際の AI 開発者タスクに当てはめる

タスク：ルーティングを agent にせず、対話的な AI 体験を届ける

タスク：検索に支えられた AI API に正しい根拠で答えさせる

タスク：アイデンティティとアクセスポリシーを迂回せずに企業データを開放する

公開 API を AI ベンダーから独立させるには？

開発中、FastAPI の /docs と /openapi.json は何の役に立つのか？

AI 推論は同期と非同期のどちらにすべきか

FastAPI のチャットにストリーミングレスポンスを追加するには？

FastAPI で AI REST API を構築するには？

ステップ1：境界のあるリクエスト schema を定義する

ステップ2：タスクを投入して 202 Accepted を返す

本番環境で Oracle AI Database を使って AI タスクを永続化するには？

ステップ3：予測状態を Oracle AI Database に永続化する

ステップ4：worker にモデル推論を処理させる

OpenAI SDK で最小限の FastAPI チャットインターフェースを構築するには？

REST API を OpenAI Responses API に接続するには？

OpenAI API key はどう安全に保存・読み込みすべきか？

AI REST API の認証と保護はどう行うか？

セッションはどう保存し、ユーザー帰属の検証をどう実装するか？

FastAPI でユーザー認証 Depends を追加するには？

Oracle AI Database のパターン：テナントスコープとリクエストの関連付け

FastAPI、ORDS、ストリーミング、キューのどれを使うべきか

AI REST API のテストとデプロイ方法

AI REST API のトラブルシューティング

本番環境チェックリスト

FAQ

AI アプリケーション向けの REST API はどう作るか？

AI REST API のベストプラクティスとは？

OpenAI ベースの REST API の認証はどうするか？

AI アプリケーションの API には FastAPI と ORDS のどちらを使うべきか？

AI REST API は同期的に結果を返すべきか？

FastAPI は本番環境の AI アプリケーションを支えられるか？

AI REST API 向けに、メンテナンスが続いている検証用サンプルはどれを動かすべきか？

Oracle AI Database は AI API にどう役立つか？

次のステップ

ドキュメント

著者

Wojtek Pluta

テクニカルプロダクトマーケティングディレクター

Portable Oracle Select AI Agent Teams：開発から本番まで

Oracle AI Database で制御された Cursor MCP ワークフローを構築する

開発中、FastAPI の `/docs` と `/openapi.json` は何の役に立つのか？

ステップ2：タスクを投入して `202 Accepted` を返す

FastAPI でユーザー認証 `Depends` を追加するには？