FastAPIで実践する外部APIクライアント設計入門：HTTPX・タイムアウト・再試行・接続プール・障害分離で壊れにくい連携を作る

要約

FastAPIから外部APIを呼ぶ処理は、単なる httpx.get() の集まりではなく、アプリの信頼性を左右する基盤として設計する必要があります。FastAPIは非同期処理に強く、外部I/Oの多いAPI連携と相性が良い一方で、イベントループを塞ぐ処理や無制限な接続は避けたい設計です。
HTTPXでは、AsyncClient を再利用し、Timeout と Limits を明示することで、接続プールと待ち時間を制御できます。HTTPXのタイムアウトは connect、read、write、pool の4種類に分かれており、接続数も max_connections や max_keepalive_connections で調整できます。
再試行は「何でも何度でも」では危険です。HTTPXの transport レイヤーでは ConnectError や ConnectTimeout への接続リトライが扱え、より広い条件や指数バックオフには Tenacity のような汎用リトライライブラリが向いています。
実務では、タイムアウト、例外変換、再試行、障害分離、監査ログ、メトリクスをまとめて「外部APIクライアント層」として持つと、ルーターやサービス層がすっきりし、障害時の切り分けもしやすくなります。HTTPXには RequestError、HTTPStatusError など整理された例外階層があり、扱い分けしやすい設計です。
この記事では、FastAPIでの外部APIクライアント設計を、基本方針 → AsyncClient共有 → タイムアウト → 再試行 → サーキットブレーカ的発想 → 監視 → テストの順で、実務向けに丁寧に整理します。

誰が読んで得をするか

個人開発・学習者さん

天気API、決済API、生成AI API、配送APIなど、外部サービスと連携し始めた方。
requests や httpx をその場で呼んではいるものの、タイムアウトやエラー処理を深く考えたことがない方。
「動くコード」から「落ちにくいコード」へ進みたい方。

この方には、まずFastAPIの中で外部API呼び出しをどう置くと後から困りにくいかを掴んでいただけます。FastAPIは非同期I/Oに適しており、HTTPXの AsyncClient はその文脈で使いやすいクライアントです。

小規模チームのバックエンドエンジニアさん

複数の外部API呼び出しがルーターやサービスに散らばってきた方。
タイムアウトや接続数、リトライ条件が人によってバラバラになっている方。
失敗時の戻り値や例外設計を、チームで揃えたい方。

この方には、外部APIクライアントを依存関数や専用クラスへ寄せる設計が特に役立ちます。FastAPIの依存関係システムと lifespan は、共有クライアントの管理に向いています。

SaaS開発チーム・スタートアップの皆さま

外部APIの不調が自社サービスのレイテンシや障害に直結している方。
請求、配送、通知、認証基盤、LLM API など、複数外部サービスに依存している方。
将来的にジョブキューやサーキットブレーカ、障害隔離を強めたい方。

この方には、クライアント層を一段抽象化し、接続プール、タイムアウト、再試行、メトリクスを一箇所で管理する設計が効いてきます。HTTPXの Limits や Timeout は、その土台としてとても使いやすいです。

アクセシビリティ評価

最初に要点を示し、そのあとに「なぜ危ないか」「どう設計するか」「どこに実装するか」の順で段階的に進めています。
専門用語は初出で短く補足し、その後は同じ言葉を使い続けています。
コードは短い責務ごとに分け、1ブロックで1つの役割が分かるようにしています。
目標レベルはAA相当です。

1. なぜ外部APIクライアント設計が大事なのか

FastAPIのルーターの中で、つい次のように書きたくなりますよね。

import httpx
from fastapi import APIRouter

router = APIRouter()

@router.get("/weather")
async def get_weather():
    async with httpx.AsyncClient() as client:
        resp = await client.get("https://api.example.com/weather")
        return resp.json()

この形でも動くのですが、実務では少しずつ問題が増えていきます。

リクエストごとに AsyncClient を作ってしまい、接続プールを再利用できない
タイムアウトが曖昧で、遅い外部APIに引きずられる
エラーが httpx の例外のまま上へ漏れて、サービス全体で扱いが揃わない
再試行条件が場所ごとに違う
メトリクスやログがなく、どの外部APIが遅いのか分からない

HTTPXの公式ドキュメントでも、非同期環境では AsyncClient を使い、クライアントインスタンスをうまく再利用することが重要だと説明されています。むやみにホットループ内で新規クライアントを作らない方がよいとも案内されています。

つまり、外部API呼び出しは「その場しのぎ」で置くのではなく、クライアント層としてまとめる方が、長い目で見るとかなり楽です。

2. 基本方針：外部API呼び出しを専用クラスに寄せる

おすすめの基本方針は、次のような分離です。

ルーター
- HTTPの入出力だけ担当する
サービス層
- 業務ロジックを担当する
外部APIクライアント層
- HTTPXを使った通信、タイムアウト、例外変換、再試行を担当する

この形にしておくと、「外部APIが失敗したときに、何を返すか」「どこまで再試行するか」を一箇所で揃えられます。

たとえば、配送API用のクライアントを切り出すイメージです。

# app/clients/shipping_client.py
class ShippingClient:
    async def create_shipment(self, payload: dict) -> dict:
        ...

こうしておけば、ルーターやサービスは「配送作成を依頼する」という意図だけを持ち、HTTPの細かい制御はクライアント層に閉じ込められます。

3. FastAPIで共有 `AsyncClient` を持つ：lifespanを使う

FastAPIでは、アプリ起動時と終了時の処理を lifespan で定義できます。公式ドキュメントでも、起動時にリソースを作り、終了時に解放するための仕組みとして案内されています。

外部APIクライアントで共有したいものの代表が、httpx.AsyncClient です。

3.1 最小のlifespan例

# app/main.py
from contextlib import asynccontextmanager

import httpx
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.http_client = httpx.AsyncClient()
    try:
        yield
    finally:
        await app.state.http_client.aclose()

app = FastAPI(lifespan=lifespan)

この形にすると、アプリ全体で AsyncClient を再利用できます。HTTPXは AsyncClient が接続プーリングなどを提供しており、同じホストに対する複数リクエストでは再利用の恩恵が大きいです。

3.2 依存関数で取り出す

# app/deps/http_client.py
import httpx
from fastapi import Request

def get_http_client(request: Request) -> httpx.AsyncClient:
    return request.app.state.http_client

こうしておくと、クライアント層やルーターから依存関数として注入できます。FastAPIの依存性注入は、こうしたコンポーネント共有と相性が良いです。

4. タイムアウトは必ず明示する：HTTPXの4種類を理解する

HTTPXでは、タイムアウトはかなり細かく制御できます。公式ドキュメントでは、connect、read、write、pool の4種類が説明されています。

connect
- 接続確立までの上限時間
read
- サーバからデータを読み込む待ち時間
write
- リクエストデータ送信時の待ち時間
pool
- 接続プールから空き接続を取る待ち時間

これを雑に全部同じ値にするのではなく、外部APIの性質に合わせて考えると安定します。

4.1 例：少し細かく分けたタイムアウト

# app/core/http.py
import httpx

DEFAULT_TIMEOUT = httpx.Timeout(
    connect=2.0,
    read=5.0,
    write=5.0,
    pool=1.0,
)

4.2 `AsyncClient` に適用する

# app/main.py
from contextlib import asynccontextmanager
import httpx
from fastapi import FastAPI

from app.core.http import DEFAULT_TIMEOUT

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.http_client = httpx.AsyncClient(timeout=DEFAULT_TIMEOUT)
    try:
        yield
    finally:
        await app.state.http_client.aclose()

app = FastAPI(lifespan=lifespan)

タイムアウトが未設定だったり緩すぎたりすると、外部APIの不調がそのまま自分のAPIの遅さになります。
「最悪どれくらい待てるか」を決めておくことは、外部連携では本当に大切です。HTTPXはタイムアウトの細分化を公式に提供しており、ここを活かすとかなり扱いやすくなります。

5. 接続プールを制御する：`Limits` を入れる

HTTPXでは、接続プールの上限も Limits で設定できます。公式ドキュメントでは、max_keepalive_connections、max_connections、keepalive_expiry が説明されています。デフォルト値も案内されています。

5.1 `Limits` の例

# app/core/http.py
import httpx

DEFAULT_LIMITS = httpx.Limits(
    max_keepalive_connections=20,
    max_connections=100,
    keepalive_expiry=5.0,
)

5.2 `AsyncClient` に適用する

# app/main.py
from contextlib import asynccontextmanager
import httpx
from fastapi import FastAPI

from app.core.http import DEFAULT_TIMEOUT, DEFAULT_LIMITS

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.http_client = httpx.AsyncClient(
        timeout=DEFAULT_TIMEOUT,
        limits=DEFAULT_LIMITS,
    )
    try:
        yield
    finally:
        await app.state.http_client.aclose()

app = FastAPI(lifespan=lifespan)

接続数を無制限にしてしまうと、外部APIが遅いときに自分のアプリ側のリソースまで圧迫しやすくなります。
逆に厳しすぎると pool timeout が増えるので、メトリクスを見ながら少しずつ調整するのが実務的です。

6. 例外をそのまま漏らさない：HTTPX例外を自分の例外へ変換する

HTTPXには整理された例外階層があります。公式ドキュメントや例外一覧では、RequestError や HTTPStatusError、さらに ConnectTimeout、ReadTimeout、PoolTimeout などが確認できます。

このまま上位へ漏らすより、アプリ独自の例外へ変換して扱いを揃える方がおすすめです。

6.1 アプリ独自例外を定義する

# app/clients/exceptions.py
class ExternalAPIError(Exception):
    pass

class ExternalAPITimeoutError(ExternalAPIError):
    pass

class ExternalAPIUnavailableError(ExternalAPIError):
    pass

class ExternalAPIBadResponseError(ExternalAPIError):
    pass

6.2 HTTPX例外を変換する

# app/clients/base.py
import httpx

from app.clients.exceptions import (
    ExternalAPIBadResponseError,
    ExternalAPITimeoutError,
    ExternalAPIUnavailableError,
)

async def safe_request(client: httpx.AsyncClient, method: str, url: str, **kwargs) -> httpx.Response:
    try:
        response = await client.request(method, url, **kwargs)
        response.raise_for_status()
        return response
    except httpx.TimeoutException as exc:
        raise ExternalAPITimeoutError(str(exc)) from exc
    except httpx.HTTPStatusError as exc:
        raise ExternalAPIBadResponseError(str(exc)) from exc
    except httpx.RequestError as exc:
        raise ExternalAPIUnavailableError(str(exc)) from exc

ここで、タイムアウト系は TimeoutException、接続障害などは RequestError、HTTPステータス異常は HTTPStatusError に分けています。これはHTTPXの公式例外分類に沿った形です。

7. 外部APIクライアントクラスを作る：配送APIの例

ここまでの部品を組み合わせて、具体的なクライアントクラスにします。

# app/clients/shipping_client.py
import httpx

from app.clients.base import safe_request

class ShippingClient:
    def __init__(self, client: httpx.AsyncClient, base_url: str, api_key: str):
        self.client = client
        self.base_url = base_url.rstrip("/")
        self.api_key = api_key

    async def create_shipment(self, payload: dict) -> dict:
        response = await safe_request(
            self.client,
            "POST",
            f"{self.base_url}/shipments",
            json=payload,
            headers={"Authorization": f"Bearer {self.api_key}"},
        )
        return response.json()

    async def get_shipment(self, shipment_id: str) -> dict:
        response = await safe_request(
            self.client,
            "GET",
            f"{self.base_url}/shipments/{shipment_id}",
            headers={"Authorization": f"Bearer {self.api_key}"},
        )
        return response.json()

このクラスの利点は、ルーターやサービスが httpx を直接知らなくてよくなることです。
また、APIキーの付け方、URLの組み立て、例外変換の流儀も揃います。

8. FastAPIの依存関数でクライアントを組み立てる

FastAPIの依存関数を使って、共有 AsyncClient から ShippingClient を組み立てます。

# app/deps/clients.py
from fastapi import Depends
import httpx

from app.clients.shipping_client import ShippingClient
from app.deps.http_client import get_http_client

def get_shipping_client(
    client: httpx.AsyncClient = Depends(get_http_client),
) -> ShippingClient:
    return ShippingClient(
        client=client,
        base_url="https://shipping.example.com/api",
        api_key="replace-me",
    )

8.1 ルーター側は薄く保つ

# app/api/v1/routers/shipments.py
from fastapi import APIRouter, Depends

from app.clients.shipping_client import ShippingClient
from app.deps.clients import get_shipping_client

router = APIRouter(prefix="/shipments", tags=["shipments"])

@router.post("")
async def create_shipment(
    payload: dict,
    shipping_client: ShippingClient = Depends(get_shipping_client),
):
    result = await shipping_client.create_shipment(payload)
    return result

この形なら、外部API連携の複雑さはクライアント層へ閉じ込められ、ルーターはかなり読みやすくなります。

9. 再試行は慎重に：HTTPX transport と Tenacity の使い分け

HTTPXの transport レイヤーでは、接続リトライが扱えます。公式ドキュメントによると、接続リトライは ConnectError と ConnectTimeout に対して有効で、もっと広い再試行条件には Tenacity のような汎用ツールを検討するよう案内されています。

9.1 接続リトライを transport で入れる例

# app/core/http.py
import httpx

TRANSPORT = httpx.AsyncHTTPTransport(retries=1)

# app/main.py
from contextlib import asynccontextmanager
import httpx
from fastapi import FastAPI

from app.core.http import DEFAULT_TIMEOUT, DEFAULT_LIMITS, TRANSPORT

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.http_client = httpx.AsyncClient(
        timeout=DEFAULT_TIMEOUT,
        limits=DEFAULT_LIMITS,
        transport=TRANSPORT,
    )
    try:
        yield
    finally:
        await app.state.http_client.aclose()

app = FastAPI(lifespan=lifespan)

これは「接続が一瞬失敗した」ようなケースには効きますが、HTTP 503 や 429 を含む柔軟な再試行には向きません。そこで Tenacity のようなライブラリが役立ちます。Tenacityは汎用の再試行ライブラリとして公式ドキュメントがあり、指数バックオフなども扱えます。

9.2 Tenacityで指数バックオフを入れる例

# app/clients/retry.py
from tenacity import (
    retry,
    retry_if_exception_type,
    stop_after_attempt,
    wait_exponential,
)

from app.clients.exceptions import (
    ExternalAPITimeoutError,
    ExternalAPIUnavailableError,
)

@retry(
    retry=retry_if_exception_type((ExternalAPITimeoutError, ExternalAPIUnavailableError)),
    wait=wait_exponential(multiplier=0.5, min=0.5, max=5),
    stop=stop_after_attempt(3),
    reraise=True,
)
async def retryable_call(func, *args, **kwargs):
    return await func(*args, **kwargs)

9.3 クライアントに組み込む

# app/clients/shipping_client.py
from app.clients.retry import retryable_call

class ShippingClient:
    # 省略

    async def create_shipment(self, payload: dict) -> dict:
        response = await retryable_call(
            safe_request,
            self.client,
            "POST",
            f"{self.base_url}/shipments",
            json=payload,
            headers={"Authorization": f"Bearer {self.api_key}"},
        )
        return response.json()

ここで大切なのは、書き込み系APIに無条件で再試行しないことです。POSTの再試行は二重作成の危険があります。
外部API側が冪等キーを受け付けるなら、それを必ず使ったうえで再試行する方が安全です。

10. 何を再試行してよいか：読み取り系と書き込み系を分ける

再試行は便利ですが、すべての外部APIに対して無条件に入れると事故になります。

再試行しやすいもの

GETなどの読み取り
接続確立失敗
一時的なネットワーク障害
一部の 429 / 503（相手の仕様を確認）

慎重にすべきもの

POST / PATCH / DELETE など副作用がある呼び出し
決済実行、注文確定、外部通知送信
冪等キーなしの書き込みAPI

つまり、再試行は「とりあえず3回」ではなく、この呼び出しは再試行して安全かを先に考えて決めるべきです。

11. サーキットブレーカ的な発想：失敗し続ける相手を引きずらない

サーキットブレーカは、「失敗が続く相手への呼び出しを一時的に止め、自分のアプリ全体が巻き込まれないようにする」考え方です。
この記事では専用ライブラリには踏み込みすぎませんが、設計上の発想だけでも持っておくと役立ちます。

たとえば、外部APIが数分単位で不調なときに、毎回フルタイムアウトまで待っていると、自分のAPI全体が遅くなります。
そこで、一定回数失敗したら「今はこのAPI不調」と判断して、しばらく素早く失敗させる方が全体の被害を抑えられます。

最初から本格的なサーキットブレーカを入れなくても、次のような段階で考えると無理がありません。

まずは短いタイムアウトを入れる
例外を自前例外へ変換する
リトライを絞って入れる
失敗率メトリクスを見ながら、必要ならサーキットブレーカを検討する

12. ログとメトリクス：どの外部APIが遅いか見えるようにする

外部API連携で後から困りやすいのは、「何が遅いのか分からない」ことです。
そのため、少なくとも次の情報はログやメトリクスに残したいです。

どの外部サービスか
どのエンドポイントを呼んだか
成功したか失敗したか
何秒かかったか
何回再試行したか
どの例外だったか

12.1 最小のロギング例

# app/clients/base.py
import logging
import time
import httpx

logger = logging.getLogger("external_api")

async def safe_request(client: httpx.AsyncClient, method: str, url: str, **kwargs) -> httpx.Response:
    started = time.perf_counter()
    try:
        response = await client.request(method, url, **kwargs)
        response.raise_for_status()
        logger.info(
            "external api success",
            extra={
                "method": method,
                "url": url,
                "status_code": response.status_code,
                "elapsed_ms": int((time.perf_counter() - started) * 1000),
            },
        )
        return response
    except httpx.HTTPError as exc:
        logger.warning(
            "external api failure",
            extra={
                "method": method,
                "url": url,
                "elapsed_ms": int((time.perf_counter() - started) * 1000),
                "error_type": exc.__class__.__name__,
            },
        )
        raise

このレベルでも、あとで「あの配送APIだけ遅いですね」と分かるようになります。

13. テスト方針：外部APIを本物で叩かない

外部APIクライアントのテストで重要なのは、本番相手を直接叩かないことです。
理由は単純で、不安定・高コスト・再現性が低いからです。

おすすめの分け方は次のとおりです。

ユニットテスト
- クライアントクラスが正しいURL、ヘッダ、例外変換をしているか
統合テスト
- モックサーバやテスト用 transport を使い、HTTPXの通信を模擬する
E2Eテスト
- 本当に必要な連携だけ、外部のテスト環境で確認する

HTTPXには transport の仕組みがあり、公式ドキュメントでもカスタム transport やテストに使える形が説明されています。

13.1 クライアント層のユニットテストのイメージ

# 疑似例
# create_shipment が Authorization ヘッダ付きで POST するか
# 404 や timeout を自前例外へ変換するか

テストの中心は、自分の変換ロジックや再試行条件に置くと効率が良いです。

14. よくある失敗パターン

14.1 ルーターの中で毎回 `AsyncClient()` を作る

HTTPXはクライアント再利用を前提にした方が効率的です。共有クライアントを使う方が、接続プーリングの恩恵を受けやすくなります。

14.2 タイムアウト未設定

外部APIが遅いと、自分のAPIもそのまま遅くなります。HTTPXには細かいタイムアウト制御があるので、明示した方が安全です。

14.3 書き込み系を無条件で再試行する

POSTの二重実行事故につながりやすいです。冪等キーや相手APIの仕様を必ず確認したいところです。

14.4 例外を `httpx` のまま上へ漏らす

サービス全体でエラー設計が揃わなくなります。自前例外へ寄せる方が扱いやすいです。HTTPXの例外階層は整理されているので、分類もしやすいです。

14.5 どの外部APIが遅いか観測していない

接続数、レイテンシ、失敗率を見えるようにしておくと、後で本当に助かります。

15. 読者別ロードマップ

個人開発・学習者さん

まずは requests ではなく httpx.AsyncClient を使う
lifespan で共有クライアントを持つ
タイムアウトと Limits を明示する
例外を自前例外へ変換する
読み取り系だけ再試行を検討する

小規模チームのエンジニアさん

外部API呼び出しを棚卸しする
ルーター直書きをやめて、クライアント層へ寄せる
接続プール、タイムアウト、再試行方針を共通化する
ログとメトリクスを追加する
外部サービスごとに責務を分けたクライアントモジュールを作る

SaaS開発チーム・スタートアップの皆さま

外部APIを「重要度」「副作用」「冪等性」で分類する
HTTPX共有クライアントと接続制御を整える
例外変換、再試行、ジョブキュー連携を設計する
障害率・タイムアウト率・再試行回数を監視する
必要ならサーキットブレーカやフォールバック戦略を追加する

参考リンク

まとめ

FastAPIでの外部API連携は、単発のHTTP呼び出しではなく、クライアント層として設計すると壊れにくくなります。
AsyncClient の共有、Timeout と Limits の明示、例外変換、再試行条件の整理は、どの規模でも効果が大きい基本セットです。HTTPXはこれらを公式にサポートしており、FastAPIの lifespan と依存関係システムとも相性が良いです。
再試行は便利ですが、何でも繰り返すのではなく、読み取り系か、副作用がある書き込み系か、相手APIが冪等かどうかを見て慎重に決めるのが安全です。HTTPXの transport リトライは接続系に向き、広い条件には Tenacity のような汎用ライブラリが適しています。
最初から完璧な障害隔離を作る必要はありませんが、クライアント層を切り出しておくだけで、将来のサーキットブレーカ導入、ジョブキュー連携、メトリクス強化がかなり楽になります。

次の記事としては、この流れと相性が良い「FastAPIで作る社内管理画面APIの設計パターン」や、「FastAPIで実践するサーキットブレーカ／フォールバック設計」が自然につながります。

FastAPIで実践する外部APIクライアント設計入門：HTTPX・タイムアウト・再試行・接続プール・障害分離で壊れにくい連携を作る

要約

誰が読んで得をするか

個人開発・学習者さん

小規模チームのバックエンドエンジニアさん

SaaS開発チーム・スタートアップの皆さま

アクセシビリティ評価

1. なぜ外部APIクライアント設計が大事なのか

2. 基本方針：外部API呼び出しを専用クラスに寄せる

3. FastAPIで共有 AsyncClient を持つ：lifespanを使う

3.1 最小のlifespan例

3.2 依存関数で取り出す

4. タイムアウトは必ず明示する：HTTPXの4種類を理解する

4.1 例：少し細かく分けたタイムアウト

4.2 AsyncClient に適用する

5. 接続プールを制御する：Limits を入れる

5.1 Limits の例

5.2 AsyncClient に適用する

6. 例外をそのまま漏らさない：HTTPX例外を自分の例外へ変換する

6.1 アプリ独自例外を定義する

6.2 HTTPX例外を変換する

7. 外部APIクライアントクラスを作る：配送APIの例

8. FastAPIの依存関数でクライアントを組み立てる

8.1 ルーター側は薄く保つ

9. 再試行は慎重に：HTTPX transport と Tenacity の使い分け

9.1 接続リトライを transport で入れる例

9.2 Tenacityで指数バックオフを入れる例

9.3 クライアントに組み込む

10. 何を再試行してよいか：読み取り系と書き込み系を分ける

再試行しやすいもの

慎重にすべきもの

11. サーキットブレーカ的な発想：失敗し続ける相手を引きずらない

12. ログとメトリクス：どの外部APIが遅いか見えるようにする

12.1 最小のロギング例

13. テスト方針：外部APIを本物で叩かない

13.1 クライアント層のユニットテストのイメージ

14. よくある失敗パターン

14.1 ルーターの中で毎回 AsyncClient() を作る

14.2 タイムアウト未設定

14.3 書き込み系を無条件で再試行する

14.4 例外を httpx のまま上へ漏らす

14.5 どの外部APIが遅いか観測していない

15. 読者別ロードマップ

個人開発・学習者さん

小規模チームのエンジニアさん

SaaS開発チーム・スタートアップの皆さま

参考リンク

まとめ

共有:

投稿者 greeden

関連投稿

コメントを残す コメントをキャンセル

見逃しています

3. FastAPIで共有 `AsyncClient` を持つ：lifespanを使う

4.2 `AsyncClient` に適用する

5. 接続プールを制御する：`Limits` を入れる

5.1 `Limits` の例

5.2 `AsyncClient` に適用する

14.1 ルーターの中で毎回 `AsyncClient()` を作る

14.4 例外を `httpx` のまま上へ漏らす

コメントを残すコメントをキャンセル