developersIO
Bedrock上のClaude呼び出しをboto3からAnthropic SDKに移行して得られたメリット

Bedrock上のClaude呼び出しをboto3からAnthropic SDKに移行して得られたメリット

boto3のConverse APIからAnthropic Python SDKのAnthropicBedrockクライアントに移行した実践記録。非同期対応の簡素化、型安全性の向上、$defs/$refワークアラウンドの削除、プロンプトキャッシュの有効化など、具体的なコード差分とともにメリットを紹介します。
Amazon BedrockClaudeAnthropicPython
2026.05.29

はじめに

Amazon Bedrock上でClaude（Sonnet/Haiku）を呼び出すPython製バックエンドを運用しています。もともとboto3のconverse() / converse_stream() APIを使っていましたが、公式のAnthropic Python SDKにAnthropicBedrockクライアントが用意されていることを知り、移行を実施しました。

結論から言うと、コード量が約30%減り、非同期対応が自然になり、Anthropicの最新機能（プロンプトキャッシュなど）をBedrock上で即座に使えるようになったので、Bedrock経由でClaudeを使っているプロジェクトにはおすすめの移行です。

この記事では、実際の移行で何が変わったのか、どんなメリットがあったのかを具体的なコード差分とともに紹介します。

前提・環境

項目 値
Python 3.13
フレームワーク FastAPI（非同期）
移行前 boto3 (converse / converse_stream API)
移行後 anthropic[bedrock] (Messages API)
モデル Claude Sonnet 4.5 / Claude Haiku
認証 EC2インスタンスロール（IAM）
リージョン ap-northeast-1（東京）

なぜ移行したのか

boto3でBedrock Converse APIを使う場合、以下の課題がありました。

1. 非同期対応の複雑さ

boto3は同期クライアントです。FastAPIのような非同期フレームワークで使うには、asyncio.to_thread()でブロッキング呼び出しをラップする必要がありました。

# boto3時代: 同期関数をスレッドプールで実行
def _blocking_call() -> str:
    response = _bedrock_client.converse(
        modelId=settings.bedrock_haiku_model_id,
        messages=[{"role": "user", "content": [{"text": prompt}]}],
        system=[{"text": system}],
        inferenceConfig={"maxTokens": 100, "temperature": 0.0},
    )
    return response["output"]["message"]["content"][0]["text"].strip()

return await asyncio.to_thread(_blocking_call)

特にストリーミングの場合はさらに複雑で、asyncio.Queueとloop.call_soon_threadsafe()を使ったスレッド間ブリッジが必要でした。

# boto3時代: ストリーミングのスレッド間ブリッジ
loop = asyncio.get_running_loop()
queue: asyncio.Queue = asyncio.Queue()

def _blocking_stream() -> None:
    response = _bedrock_client.converse_stream(...)
    for event in response["stream"]:
        chunk = event["contentBlockDelta"].get("delta", {}).get("toolUse", {}).get("input", "")
        loop.call_soon_threadsafe(queue.put_nowait, chunk)
    loop.call_soon_threadsafe(queue.put_nowait, None)  # sentinel

task = asyncio.create_task(asyncio.to_thread(_blocking_stream))
while True:
    item = await queue.get()
    if item is None:
        break
    yield item

2. レスポンスが型なしの辞書

boto3のレスポンスは素のPython辞書です。IDEの補完が効かず、防御的なアクセスが必要でした。

# boto3: 辞書アクセス — タイプミスに気づきにくい
for block in response["output"]["message"]["content"]:
    if block.get("toolUse", {}).get("name") == "classify_problem":
        return ProblemClassification(**block["toolUse"]["input"])

3. JSON Schema $defs/$refの非サポート

Bedrock Converse APIのtoolConfig.inputSchemaは、JSON Schemaの$defs/$refをサポートしていません。Pydanticモデルのmodel_json_schema()はネストされたモデルに対して$refを使うため、送信前に$refを展開するワークアラウンド関数が必要でした。

# boto3時代: $defs/$refを手動インライン展開するヘルパー
def _resolve_schema_refs(schema: dict) -> dict:
    defs = schema.pop("$defs", {})

    def _resolve(obj: Any) -> Any:
        if isinstance(obj, dict):
            if "$ref" in obj:
                ref_name = obj["$ref"].split("/")[-1]
                return _resolve(dict(defs[ref_name]))
            return {k: _resolve(v) for k, v in obj.items()}
        if isinstance(obj, list):
            return [_resolve(item) for item in obj]
        return obj

    return _resolve(schema)

# 呼び出し箇所（3箇所）
tool_schema = _resolve_schema_refs(ChatToolInput.model_json_schema())

4. Anthropic最新機能の利用不可

boto3のConverse APIはAWS独自のインターフェースであり、Anthropicが新機能をリリースしても、AWSがConverse APIに対応するまで使えません。プロンプトキャッシュや拡張thinking（extended thinking）などの機能がこれに該当します。

移行の実際

インストール

uv add "anthropic[bedrock]>=0.104.0"

[bedrock]エクストラがboto3署名処理を内包するため、LLM呼び出しに関してはboto3を直接importする必要がなくなります。ただし、S3やKnowledge Baseなど他のAWSサービスを使っている場合はboto3も引き続き必要です。

クライアント初期化

# Before
import boto3
_bedrock_client = boto3.client("bedrock-runtime", region_name=settings.aws_region)

# After
from anthropic.lib.bedrock import AsyncAnthropicBedrock
_client = AsyncAnthropicBedrock(aws_region=settings.aws_region)

EC2インスタンスロールによるデフォルト認証チェーンはそのまま動作します。aws_access_key/aws_secret_keyを明示的に渡すこともできますが、IAMロールを使っていれば不要です。

非ストリーミング呼び出し

変更は驚くほどシンプルでした。

# Before (boto3)
def _blocking_call() -> str:
    response = _bedrock_client.converse(
        modelId=settings.bedrock_haiku_model_id,
        messages=[{"role": "user", "content": [{"text": prompt}]}],
        system=[{"text": system}],
        inferenceConfig={"maxTokens": 100, "temperature": 0.0},
    )
    return response["output"]["message"]["content"][0]["text"].strip()

return await asyncio.to_thread(_blocking_call)

# After (Anthropic SDK)
response = await _client.messages.create(
    model=settings.bedrock_haiku_model_id,
    messages=[{"role": "user", "content": prompt}],
    system=system,
    max_tokens=100,
    temperature=0.0,
)
block = response.content[0]
return block.text.strip() if block.type == "text" else ""

変わった点:

  • asyncio.to_thread() + 内部関数が不要になった（AsyncAnthropicBedrockがネイティブ非同期）
  • messagesの形式がシンプルに（[{"text": "..."}]のラッピングが不要）
  • systemが文字列をそのまま受け取る（[{"text": "..."}]不要）
  • レスポンスが型付きオブジェクト（block.text, block.typeで補完が効く）

Tool Use

Tool Useの定義もフラットになりました。

# Before (boto3 Converse API)
toolConfig={
    "tools": [
        {
            "toolSpec": {
                "name": "classify_problem",
                "description": "...",
                "inputSchema": {"json": _resolve_schema_refs(tool_schema)},
            }
        }
    ],
    "toolChoice": {"tool": {"name": "classify_problem"}},
}

# After (Anthropic SDK)
tools=[
    {
        "name": "classify_problem",
        "description": "...",
        "input_schema": tool_schema,  # $defs/$refがそのまま使える
    }
],
tool_choice={"type": "tool", "name": "classify_problem"},

Messages APIは$defs/$refをネイティブにサポートするため、_resolve_schema_refs()ヘルパー（20行）と3箇所の呼び出しを丸ごと削除できました。

レスポンスの取得も型付きになります。

# Before
for block in response["output"]["message"]["content"]:
    if block.get("toolUse", {}).get("name") == "classify_problem":
        return ProblemClassification(**block["toolUse"]["input"])

# After
for block in response.content:
    if block.type == "tool_use" and block.name == "classify_problem":
        return ProblemClassification(**block.input)

ストリーミング

最も劇的に改善されたのがストリーミングです。asyncio.Queue、call_soon_threadsafe、sentinel値、asyncio.create_task(asyncio.to_thread(...))のスレッド間ブリッジが、async with ... stream()に置き換わりました。

# Before (boto3): 60行以上のスレッド間ブリッジ
loop = asyncio.get_running_loop()
queue: asyncio.Queue = asyncio.Queue()

def _blocking_stream() -> None:
    try:
        response = _bedrock_client.converse_stream(...)
        for event in response["stream"]:
            if "contentBlockDelta" not in event:
                continue
            chunk = event["contentBlockDelta"].get("delta", {}).get("toolUse", {}).get("input", "")
            # ...処理...
            loop.call_soon_threadsafe(queue.put_nowait, token)
    except Exception as exc:
        loop.call_soon_threadsafe(queue.put_nowait, exc)
    finally:
        loop.call_soon_threadsafe(queue.put_nowait, None)

task = asyncio.create_task(asyncio.to_thread(_blocking_stream))
try:
    while True:
        item = await queue.get()
        if item is None:
            break
        if isinstance(item, BaseException):
            raise item
        yield item
finally:
    await task

# After (Anthropic SDK): ネイティブ非同期ストリーム
async with _client.messages.stream(
    model=settings.bedrock_model_id,
    messages=sdk_messages,
    system=CHAT_SYSTEM_PROMPT,
    max_tokens=8192,
    temperature=0.3,
    tools=[...],
    tool_choice={"type": "tool", "name": "submit_response"},
) as stream:
    async for event in stream:
        if event.type != "input_json":
            continue
        chunk = event.partial_json
        # ...同じ状態マシンでトークン処理...
        yield token

状態マシン（general_adviceのトークン抽出やTicketResultのJSON解析）はboto3時代と完全に同じロジックですが、それを囲むインフラ部分のコードが劇的に減りました。

移行後に得られたメリット: プロンプトキャッシュ

移行によるメリットの具体例として、プロンプトキャッシュを紹介します。

プロンプトキャッシュとは

Anthropicのプロンプトキャッシュは、リクエスト間で共通するプレフィックス（システムプロンプト、ツール定義など）をキャッシュし、2回目以降の処理を高速化する機能です。キャッシュされたトークンは入力コストが90%削減され、TTFT（最初のトークンまでの時間）も短縮されます。

この機能はAnthropic Messages APIでのみサポートされており、boto3のConverse APIでは利用できません。これが移行の決め手の一つでもありました。

実装

移行後のコードでプロンプトキャッシュを有効にするのに必要な変更は、たった1行です。

tools=[
    {
        "name": "submit_response",
        "description": "Submit the structured helpdesk response",
        "input_schema": tool_schema,
        "cache_control": {"type": "ephemeral"},  # この1行を追加
    }
],

cache_controlをツール定義の最後の要素に配置することで、システムプロンプトからツール定義までの静的プレフィックス全体がキャッシュされます。プロンプトキャッシュはプレフィックスベースで動作するため、cache_controlの位置より前のすべてのトークンがキャッシュ対象になります。

Sonnetの場合、最低1,024トークンのプレフィックスが必要です。本プロジェクトのシステムプロンプト（約1,000+トークン）とツール定義（約300〜500トークン）を合わせると約1,500+トークンになるため、条件を満たしています。

キャッシュの確認

ストリーム完了後にusageオブジェクトからキャッシュメトリクスを取得できます。

final_msg = await stream.get_final_message()
usage = final_msg.usage
if usage.cache_read_input_tokens:
    logger.info("prompt cache HIT: %d tokens read from cache", usage.cache_read_input_tokens)
elif usage.cache_creation_input_tokens:
    logger.info("prompt cache MISS (created): %d tokens written", usage.cache_creation_input_tokens)

実際のログ出力:

INFO: prompt cache MISS (created): 1587 tokens written   # 1回目
INFO: prompt cache HIT: 1587 tokens read from cache       # 2回目以降（5分以内）

キャッシュはアカウント/リージョン/モデル単位で共有されるため、異なるユーザーのリクエストでも5分以内であればキャッシュヒットします。セッション管理やキャッシュキーの設計は不要です。

その他のメリット

移行してわかったその他のメリットもまとめます。

APIの一貫性

Anthropicの公式ドキュメントやSDKリファレンスがそのまま使えます。boto3のConverse APIはAWS独自のインターフェースであり、パラメータ名やレスポンス形式がAnthropic APIと微妙に異なります（maxTokens vs max_tokens、toolConfig vs toolsなど）。これにより、ドキュメント参照時にConverse APIへの読み替えが不要になります。

型安全性

response.content[0].textのようにIDEの補完とPyrightの型チェックが効きます。boto3の場合、レスポンスはdict[str, Any]なので、キー名のタイプミスが実行時まで発見できません。

将来の機能追加への対応

Anthropicが新機能をリリースした場合、anthropicパッケージをアップデートするだけで即座に利用可能です。boto3のConverse APIではAWS側の対応を待つ必要があり、機能によっては数週間〜数ヶ月のタイムラグが生じます。extended thinking（拡張思考）や、今後追加される新機能にもこの方法で対応できます。

注意点

boto3が不要になるわけではない

anthropic[bedrock]はLLM呼び出し（Messages API）のみをカバーします。S3、Knowledge Base、Cost Explorerなど他のAWSサービスにはboto3が引き続き必要です。本プロジェクトでもboto3は依存関係に残っています。

メッセージ形式の差異

既存のコードがBedrock Converse形式（[{"role": "user", "content": [{"text": "..."}]}]）でメッセージを構築している場合、Anthropic SDK形式（[{"role": "user", "content": "..."}]）への変換が必要です。本プロジェクトでは呼び出し元のルートハンドラを変更せず、サービス層に薄い変換ヘルパーを追加しました。

def _convert_messages(bedrock_messages: list[dict]) -> list[dict]:
    result = []
    for msg in bedrock_messages:
        content = msg["content"]
        if len(content) == 1 and "text" in content[0]:
            result.append({"role": msg["role"], "content": content[0]["text"]})
        else:
            result.append({
                "role": msg["role"],
                "content": [{"type": "text", "text": b["text"]} for b in content],
            })
    return result

モデルIDのプレフィックス

Bedrock上のモデルIDはリージョンプレフィックスを含む場合があります（例: ap-northeast-1.anthropic.claude-sonnet-4-5-20250514-v1:0）。AnthropicBedrockクライアントはこの形式をそのまま受け入れるため、設定変更は不要でした。

まとめ

boto3からAnthropic SDKへの移行は、LLM呼び出しが1ファイルに集約されていたこともあり、半日程度で完了しました。結果として:

  • コード量: 474行 → 265行削除、430行追加（ネットで約30%減）
  • 非同期対応: asyncio.to_thread + Queue → ネイティブasync/await
  • 型安全性: dict[str, Any] → 型付きオブジェクト
  • スキーマ互換: _resolve_schema_refs()ワークアラウンド → 不要（$defs/$refネイティブサポート）
  • 最新機能: プロンプトキャッシュが1行追加で有効化可能

移行の敷居は低く、メリットは大きいです。Bedrock上でClaudeを使っているプロジェクトでは、検討する価値があると思います。

生成AI活用はクラスメソッドにお任せ

過去に支援してきた生成AIの支援実績100+を元にホワイトペーパーを作成しました。御社が抱えている課題のうち、どれが解決できて、どのようなサービスが受けられるのか？4つのフェーズに分けてまとめています。どうぞお気軽にご覧ください。

生成AI資料イメージ

無料でダウンロードする

この記事をシェアする

AWSのお困り事はクラスメソッドへ

関連記事

話題の記事
Claude Platform on AWS が発表されたので、Bedrock との違いを整理してみた
suzuki.ryo
2026.04.27
Bedrock の Anthropic 互換エンドポイント(Mantle)で、LangChain から Opus 4.7 を呼んでみた
suzuki.ryo
2026.04.17
Streamlit公式のAgent Skillsを使ってAmazon BedrockのチャットUIをClaude Codeに作らせてみた
asano haruki
2026.05.19
[アップデート]Amazon BedrockでAnthropicのClaude Opus 4.1が利用可能になりました
Takaaki Kakei
2025.08.06