PythonでOpenAI APIプロキシを構築する：まず自分で書いてから、すべてのモデルにつなげる

TL;DR

• OpenAI APIプロキシとは、あなたのアプリケーションとモデルAPIの間に置く小さなサーバーです。実際のキーを保管し、API自体が提供しない機能を補います。顧客ごとのキー、レート制限、ログ、キーローテーションなどです。以下に約50行で動くFastAPI版を示します。
• HiAPIのテキストAPIはOpenAI互換です。POST /v1/chat/completions と POST /v1/responses は https://api.hiapi.ai/v1 にあります。公式の openai SDKやあなたのプロキシをこのbase URLに向ければ、1つのキーでDeepSeek、Claude、GPT、Kimi、GLM、Grokのテキストモデルを呼び出せます。
• 画像と動画は事情が異なります。数秒から数分かかるため、HiAPIの非同期 POST /v1/tasks 上で動作します。こうしたリクエストを同期プロキシに押し込むべきではありません。

ステップ1：最小限のパススループロキシ

プロキシはクライアントから送られた内部キーを検証し、実際の上流キーに置き換えて、レスポンスをストリーミングで返します。

pip install fastapi uvicorn httpx

# proxy.py
import os
import httpx
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import StreamingResponse
from starlette.background import BackgroundTask

UPSTREAM = os.environ.get("UPSTREAM_BASE", "https://api.openai.com/v1")
UPSTREAM_KEY = os.environ["UPSTREAM_KEY"]
INTERNAL_KEYS = set(os.environ["INTERNAL_KEYS"].split(","))  # comma-separated

app = FastAPI()
client = httpx.AsyncClient(timeout=120.0)

def auth(request: Request) -> str:
    header = request.headers.get("authorization", "")
    key = header.removeprefix("Bearer ").strip()
    if key not in INTERNAL_KEYS:
        raise HTTPException(401, "invalid internal key")
    return key

@app.post("/v1/{path:path}")
async def forward(path: str, request: Request):
    tenant = auth(request)
    body = await request.body()
    upstream_req = client.build_request(
        "POST", f"{UPSTREAM}/{path}", content=body,
        headers={"Authorization": f"Bearer {UPSTREAM_KEY}",
                 "Content-Type": request.headers.get("content-type", "application/json")},
    )
    upstream = await client.send(upstream_req, stream=True)
    # Stream back so long answers and SSE don't time out; close the upstream stream when done
    return StreamingResponse(upstream.aiter_raw(), status_code=upstream.status_code,
                             media_type=upstream.headers.get("content-type"),
                             background=BackgroundTask(upstream.aclose))

実行します：

UPSTREAM_KEY=your-upstream-key INTERNAL_KEYS=tenant-a,tenant-b \
uvicorn proxy:app --host 0.0.0.0 --port 8000

クライアントは通常どおりOpenAI SDKを使い、base URLにプロキシのアドレスを設定します：

from openai import OpenAI

client = OpenAI(api_key="tenant-a", base_url="http://localhost:8000/v1")
resp = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Hello"}],
)
print(resp.choices[0].message.content)

会話、ストリーミング、埋め込みはすべて同じハンドラーを通ります。プロキシはリクエストボディの内部を一切解析しないためです。

ステップ2：上流APIが提供しない機能を補う

テナントごとのレート制限。これがないと、1人の活発な顧客があなたのクォータ全体を消費してしまいます。

import time
from collections import defaultdict

_window: dict[str, list[float]] = defaultdict(list)
LIMIT = 60  # requests per 60 s

def check_rate_limit(tenant: str):
    now = time.time()
    _window[tenant] = [t for t in _window[tenant] if now - t < 60]
    if len(_window[tenant]) >= LIMIT:
        raise HTTPException(429, "rate limit exceeded")
    _window[tenant].append(now)

auth() の直後にこれを呼び出します。サーバープロセスが複数ある場合は、カウンターをRedisに移してください（INCR と EXPIRE）。

ログ。テナント、パス、ステータスコード、レイテンシ、トークン数を記録します。Authorization ヘッダーは決して記録しないでください。

キーローテーション。UPSTREAM_KEY をシークレット管理サービスから読み込み、定期的に更新します。上流キーが裏で差し替わっても、クライアントは引き続きそれぞれのテナントキーを使えます。

コストルール。リクエストボディを解析し、ポリシーを超えるリクエストを拒否します。たとえば無料プランで高価なモデルを使う場合や、max_tokens があなたの上限を超える場合です。

ステップ3：複数のモデルベンダーを同じプロキシの背後に置く

プロキシを1社のベンダーだけに向けるのは簡単です。Anthropic、DeepSeek、Moonshotを追加するとなると、認証ヘッダー、リクエスト構造、エラー形式がそれぞれ異なるため、薄いプロキシは翻訳レイヤーに変わってしまいます。

HiAPIのテキストAPIは、すべてのテキストモデルに対してOpenAIのリクエスト構造を採用しています。そのため上記のプロキシは変更なしで動作します。環境変数を2つ設定します：

UPSTREAM_BASE=https://api.hiapi.ai/v1 UPSTREAM_KEY=your-hiapi-key INTERNAL_KEYS=tenant-a,tenant-b \
uvicorn proxy:app --host 0.0.0.0 --port 8000

あるいはプロキシを省略して、SDKを直接使うこともできます：

import os
from openai import OpenAI

client = OpenAI(api_key=os.environ["HIAPI_KEY"], base_url="https://api.hiapi.ai/v1")
resp = client.chat.completions.create(
    model="deepseek-v4-flash",
    messages=[{"role": "user", "content": "Explain sparse attention in two sentences."}],
    max_tokens=400,
)
print(resp.choices[0].message.content)

公式の openai Pythonパッケージ（3.35.1）でこのコードを実行しました。通常の呼び出しと stream=True 付きの呼び出しの2回で、どちらもエラーなく完了し、SDKも通常どおりレスポンスを解析しました。このテストで1つ落とし穴がありました。max_tokens=20 にすると、deepseek-v4-flash が20トークンすべてを推論に費やしたため（usage.completion_tokens_details.reasoning_tokens は22）、返信が空になりました。推論モデルには数百トークンの余裕を確保してください。

HiAPI上のテキストモデルの一部：

モデルID	ルート	入力 / 出力、100万トークンあたり米ドル
deepseek-v4-flash	Chat Completions	$0.14 / $0.28
deepseek-v4.1-flash	Chat Completions と Responses	$0.44 / $1.32
gpt-5.6-luna	Responses のみ	$0.05 / $0.30
grok-4.6	Chat Completions	$0.84 / $2.52
deepseek-v4-pro	Chat Completions と Responses	$1.50 / $4.50
claude-opus-5-5	Chat Completions	$2.02 / $10.08
kimi-k3	Chat Completions	$2.32 / $11.60
claude-sonnet-4-6	Chat Completions	$3.00 / $15.00

HiAPIのリアルタイム価格、2026年10月20日時点で取得。GET https://api.hiapi.ai/v1/models はあなたのキーで利用可能なすべてのモデルを一覧表示します。

gpt-5.6-luna のようなGPTモデルはResponses API（client.responses.create）でのみ提供されるため、/v1/{path} を転送するプロキシなら両方をカバーできます。

未知のモデルIDは、サポートへの問い合わせ用に request_id を添えたOpenAIのエラー形式でHTTP 400を返します：

{"error": {"code": "model_not_found", "type": "hiapi_error",
           "message": "The requested model does not exist or is not available. Check the exact model ID with GET /v1/models and try again. ...",
           "request_id": "..."}}

ステップ4：画像と動画は同期プロキシに入れるべきではない

同期プロキシは、回答が返るまで接続を開いたままにします。これはテキストには問題ありません。画像は数十秒、動画は数分かかり、これはほとんどのロードバランサーやブラウザが待てる時間より長くなります。

HiAPIは画像と動画のモデルを非同期エンドポイントで動作させます。タスクを送信し、タスクIDを取得し、ポーリングまたはコールバックで受け取ります。

curl -s -X POST https://api.hiapi.ai/v1/tasks \
  -H "Authorization: Bearer $HIAPI_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "z-image",
    "input": {"prompt": "a glass data center at golden hour, isometric", "aspect_ratio": "16:9"},
    "callback": {"url": "https://your-app.example.com/hiapi/callback", "when": "final"}
  }'
# => {"code":200,"message":"success","data":{"taskId":"tk-hiapi-..."}}

あなたのプロキシは送信呼び出しを引き続き転送できます。これは即座に返ります。結果はその後、あなたのコールバックURLまたは GET /v1/tasks/{taskId} で返されます。出力リンクは約7日後に期限切れになるため、ファイルを保存してください。

本番環境のチェックリスト

• ログからキーを取り除く。受信・送信を問わず、Authorization は決して記録しないでください。
• メディアタスクにはコールバックを使う。何千ものタスクをポーリングするとリクエストが無駄になります。コールバックの処理は冪等にし、taskId をキーにしてください。配信がリトライされる可能性があるためです。
• ポーリングの前にタスクIDを保存する。プロセスがクラッシュしたら、再送信して2回課金されるのではなく、ポーリングを再開してください。
• 推論トークンのために余裕を確保する。max_tokens は想定する回答長よりはるかに大きく設定してください。
• キーで何が使えるか確認する。GET /v1/models はそのキーに対応するリストを返します。

よくある質問

HiAPIでOpenAI Python SDKを使えますか？

はい、テキストモデルで使えます。base_url="https://api.hiapi.ai/v1" を設定し、HiAPIキーを使ってください。画像と動画は /v1/tasks を使いますが、OpenAI SDKはこのエンドポイントをカバーしていません。requests または httpx で呼び出してください。

自前のプロキシはまだ必要ですか？

どのベンダーに対しても同じ理由がある場合にのみ必要です。テナントごとのキー、レート制限、監査ログなどです。プロキシ内部でベンダーごとの変換を行う必要はもうありません。

ストリーミングは使えますか？

はい。stream=True は data: [DONE] で終わるサーバー送信イベントを返し、上記のプロキシはそれらをバッファリングせずにストリーミング転送します。

サービスを中断せずにHiAPIキーをローテーションするには？

コンソールで新しいキーを作成し、プロキシまたはシークレットストアにデプロイしてから、古いキーを失効させます。切り替え期間中は両方のキーが使えます。

詳細：Chat Completions ドキュメント、Responses ドキュメント、そしてOpenAI APIプロキシの代替としてのHiAPI。