Claude Media
ClaudeのPython SDKで非同期実行とtool_runnerを実装する
Anthropicチュートリアル

ClaudeのPython SDKで非同期実行とtool_runnerを実装する

公式Python SDKの同期/非同期クライアント、aiohttpによる並列化、tool_runnerの自動実行ループをコード例で解説します。

読了目安 約34分

Python SDKの導入 — インストールと動作要件

公式のPython SDKはpip install anthropicで導入します。Python 3.10以上が前提です。0.x系からの移行では破壊的変更が多いため、公式のマイグレーションガイドを先に確認する必要があります。

プラットフォーム統合用のクライアントクラス自体は基本パッケージanthropicに同梱されていますが、Amazon Bedrock・Google CloudのAgent Platform・Claude Platform on AWSを実際に動かすにはそれぞれ実行時依存を追加する拡張パッケージのインストールが要ります。Microsoft Foundryはクラス・依存とも基本パッケージだけで動くため追加インストールは不要です。非同期処理を高速化するaiohttpバックエンドも拡張の一つです。

pip install anthropic
# Bedrockを使うなら
pip install "anthropic[bedrock]"
# 非同期の並列度を上げるなら
pip install "anthropic[aiohttp]"
import os
from anthropic import Anthropic
 
client = Anthropic(
    api_key=os.environ.get("ANTHROPIC_API_KEY"),  # デフォルト値なので省略可能
)
 
message = client.messages.create(
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello, Claude"}],
    model="claude-opus-5",
)
 
for block in message.content:
    if block.type == "text":
        print(block.text)

APIキーをソース管理に含めたくない場合はpython-dotenvで.envファイルから読み込む方法が公式に推奨されています。個人・サービスアカウントキーが複数ワークスペースにアクセスできる設定なら、リクエストヘッダーanthropic-workspace-idでワークスペースを明示します。

同期クライアントと非同期クライアントを使い分ける

Python SDKはAnthropic(同期)とAsyncAnthropic(非同期)の2クラスを提供します。インターフェースはほぼ共通で、非同期版はawaitを付けて呼び出す点だけが違います。Webアプリのリクエストハンドラ内から呼ぶ、複数のClaude呼び出しを並列で走らせたいといった場面では非同期クライアントが自然な選択です。

import os
import asyncio
from anthropic import AsyncAnthropic
 
client = AsyncAnthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))
 
 
async def main() -> None:
    message = await client.messages.create(
        max_tokens=1024,
        messages=[{"role": "user", "content": "Hello, Claude"}],
        model="claude-opus-5",
    )
    print(message.content)
 
 
asyncio.run(main())

aiohttpでI/Oバウンドな並列度を上げる

AsyncAnthropicのデフォルトHTTPバックエンドはhttpx2です。数百件のリクエストを同時に投げるようなバッチ的な非同期処理では、aiohttpバックエンドに切り替えることで並列時のスループットが改善します。DefaultAioHttpClientをコンテキストマネージャとして使い、処理が終わったらコネクションを明示的に閉じます。

import os
import asyncio
from anthropic import AsyncAnthropic, DefaultAioHttpClient
 
 
async def main() -> None:
    async with AsyncAnthropic(
        api_key=os.environ.get("ANTHROPIC_API_KEY"),
        http_client=DefaultAioHttpClient(),
    ) as client:
        message = await client.messages.create(
            max_tokens=1024,
            messages=[{"role": "user", "content": "Hello, Claude"}],
            model="claude-opus-5",
        )
        print(message.content)
 
 
asyncio.run(main())

pip install "anthropic[aiohttp]"で拡張を追加しないとDefaultAioHttpClientは使えません。多数の並列リクエストを持つワークロードほど、この切り替えの効果が出やすくなります。単発のスクリプトやCLIツールのように並列度を必要としない場面では、素のhttpx2バックエンドのままで十分です。拡張ライブラリを増やすほど依存関係の管理コストも増えるため、実際に並列リクエストがボトルネックになっているかを計測してから切り替えるのが手堅い進め方です。

Streaming helpersで応答を逐次取得する

同期・非同期どちらのクライアントも、Server-Sent EventsによるストリーミングをサポートしSDK共通のインターフェースで扱えます。イベントをそのまま受け取るだけならstream=Trueで十分ですが、テキスト断片の蓄積と最終メッセージの取得をまとめて扱いたい場合はclient.messages.stream()のコンテキストマネージャが便利です。

async def main() -> None:
    async with client.messages.stream(
        max_tokens=1024,
        messages=[{"role": "user", "content": "Say hello there!"}],
        model="claude-opus-5",
    ) as stream:
        async for text in stream.text_stream:
            print(text, end="", flush=True)
        print()
 
        message = await stream.get_final_message()
        print(message.to_json())
 
 
asyncio.run(main())

client.messages.create(..., stream=True)は最終メッセージを組み立てないぶんメモリ消費が少なく、イベント単位の処理だけで完結する用途に向きます。

送信前にトークン数を数える

レスポンスのusageプロパティで実際の消費トークン数は分かりますが、送信前に見積もりたい場合はcount_tokensエンドポイントを使います。

count = client.messages.count_tokens(
    model="claude-opus-5", messages=[{"role": "user", "content": "Hello, world"}]
)
print(count.input_tokens)  # 10

レート制限との関係や、Start/Build/Scale各段階での上限まで含めた詳しい使い方はcount_tokensで送信前にトークン数を数える方法にまとめています。

tool_runnerでツール呼び出しの自動実行ループを組む

ツール呼び出しの往復(モデルの応答を受け取り、該当ツールを実行し、結果を返す)を自分で実装すると、ループの終了条件やエラー処理を毎回書くことになります。Python SDKの@beta_toolデコレータは、関数シグネチャとdocstringからツールスキーマを自動生成し、tool_runnerにそのまま渡せる形にします。

import json
from anthropic import Anthropic, beta_tool
 
client = Anthropic()
 
 
@beta_tool
def get_weather(location: str) -> str:
    """Get the weather for a given location.
 
    Args:
        location: The city and state, for example, San Francisco, CA
    Returns:
        A JSON-encoded string with the location, temperature, and weather condition.
    """
    return json.dumps({"location": location, "temperature": "68°F", "condition": "Sunny"})
 
 
runner = client.beta.messages.tool_runner(
    max_tokens=1024,
    model="claude-opus-5",
    tools=[get_weather],
    messages=[{"role": "user", "content": "What is the weather in SF?"}],
)
for message in runner:
    print(message)

runnerをイテレートするたびに1回のAPIリクエストが発生します。応答がツール呼び出しを含んでいれば自動でツールが実行され、その結果が次のイテレーションでモデルへ返される仕組みです。ツール実行ループの一般的な設計方針や他言語との比較はTool RunnerでAnthropic APIのツール呼び出しループを自動化するで扱っています。

Message Batchesとエラー処理・リトライ

client.messages.batches名前空間でMessage Batchesにも対応しています。基本の流れはbatches.create()でリクエストの配列を送り、processing_statusが"ended"になったらbatches.results()で1件ずつ結果を取り出すというものです。

message_batch = client.messages.batches.create(
    requests=[
        {
            "custom_id": "my-first-request",
            "params": {"model": "claude-opus-5", "max_tokens": 1024, "messages": [{"role": "user", "content": "Hello, world"}]},
        },
    ]
)

custom_idは1〜64文字の英数字・ハイフン・アンダースコア(^[a-zA-Z0-9_-]{1,64}$)に限られ、バッチ内でリクエストを一意に識別するために使います。processing_statusは作成直後in_progressで始まり、全リクエストの処理が終わるとendedになります。キャンセルすると直後はcancelingを経てendedに落ち着き、キャンセルまでに処理済みだった分の結果はそのまま残ります。batches.results()で返る各エントリのresult.typeはsucceeded(成功)・errored(リクエスト不正やサーバーエラー)・canceled(キャンセルによる未送信)・expired(24時間以内に送信できず失効)の4種類で、errored・canceled・expiredの3つは課金対象外です。結果の返却順序はリクエスト順と一致しない場合があるため、突き合わせには必ずcustom_idを使います。

for entry in client.messages.batches.results(message_batch.id):
    if entry.result.type == "succeeded":
        print(entry.custom_id, entry.result.message.content)
    elif entry.result.type == "errored":
        print(entry.custom_id, "error:", entry.result.error)

バッチは1回あたり最大100,000リクエストまたは256MBのいずれか早く達した方が上限です。多くは1時間以内に完了しますが、混雑時は処理が遅れ24時間の期限切れに達するリクエストが増えることもあります。結果を取得できるのはバッチ作成から29日以内で、これを過ぎるとバッチ自体は参照できても結果(results)はダウンロードできなくなります。

接続に失敗した場合やAPIが4xx/5xx応答を返した場合はAPIErrorのサブクラスが送出されます。APIConnectionErrorはネットワーク到達不能、RateLimitErrorは429、APIStatusErrorはそれ以外の非200系という切り分けです。接続エラー・408・409・429・500番台はデフォルトで2回まで指数バックオフ付きの自動リトライが入り、max_retriesで全体設定またはリクエスト単位の上書きができます。タイムアウトはデフォルト10分で、httpx2.Timeoutオブジェクトを渡せば接続・読み取り・書き込みごとに個別の秒数を設定できます。

接続・読み取り・書き込みを別々に設定したい場合の書き方は次の通りです。

import httpx2
from anthropic import Anthropic
 
# 全リクエスト共通のデフォルト
client = Anthropic(timeout=20.0)  # 20秒(デフォルトは10分)
 
# 接続・読み取り・書き込みを個別に設定
client = Anthropic(timeout=httpx2.Timeout(60.0, read=5.0, write=10.0, connect=2.0))
 
# リクエスト単位の上書き
client.with_options(timeout=5.0).messages.create(
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello, Claude"}],
    model="claude-opus-5",
)

大きなmax_tokensを非ストリーミングで指定するのは避けます。ネットワークによってはidle接続を一定時間で切断するものがあり、応答を受け取れないままリクエストが失敗する原因になります。SDKは非ストリーミングリクエストの所要時間が概ね10分を超えると判断した場合にValueErrorを送出します。stream=Trueを渡すかtimeoutオプションを上書きするとこの防御は無効化されます。デフォルトではSDKがidleタイムアウトの影響を減らすためTCP socket keep-aliveを設定しますが、独自のhttp_clientを渡すとこの挙動は上書きされます。

Auto-paginationとBeta機能へのアクセス

一覧系エンドポイントはページネーションされています。同期クライアントならfor、非同期クライアントならasync forで回すだけで、SDKが必要に応じて次ページを自動取得します。

client = Anthropic()
 
all_batches = []
for batch in client.messages.batches.list(limit=20):
    all_batches.append(batch)

細かい制御が必要なら.has_next_page()と.get_next_page()を使い、ページ単位で処理を分けることもできます。

ベータ機能はほとんどがclient.beta配下から利用でき、有効化にはbetasフィールドへベータヘッダーの識別子を渡します。たとえばcontext editingを使う場合は次のようになります。

response = client.beta.messages.create(
    model="claude-opus-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello, Claude"}],
    betas=["context-management-2025-06-27"],
)

ベータ機能は一般提供前の早期フィードバック段階なので、利用できる機能一覧は随時変わります。実装前にbuild with Claude overviewで最新の対応状況を確認する習慣をつけておくと、想定外の404に遭遇しにくくなります。

Bedrock・Agent Platform・AWS・Foundryのクライアントを使い分ける

プラットフォーム統合用のクライアントクラスは基本パッケージanthropicにすべて含まれています。Agent PlatformはAnthropicVertex(pip install "anthropic[vertex]")、BedrockはInvokeModel APIを使う既存アプリ向けのAnthropicBedrockと新規プロジェクト向けのAnthropicBedrockMantle(どちらもpip install "anthropic[bedrock]")、Claude Platform on AWSはAnthropicAWS(pip install "anthropic[aws]"、ベータ)、Foundryは追加インストール不要のAnthropicFoundryです。AnthropicAWSはworkspace_idをコンストラクタか環境変数ANTHROPIC_AWS_WORKSPACE_IDで渡します。

パッケージのバージョンはanthropic.__version__で実行時に確認できます。アップグレードしたはずの新機能が見当たらないときは、まずこの値で意図した通りの環境が使われているかを疑います。

よくあるつまずき

  • 同期クライアントを非同期コードから呼ぶ: Anthropic(同期)のメソッドをasync def関数の中でそのまま呼ぶとイベントループをブロックします。非同期文脈ではAsyncAnthropicに統一します
  • aiohttp拡張の入れ忘れ: http_client=DefaultAioHttpClient()を指定してもpip install "anthropic[aiohttp]"をしていないとインポートエラーになります
  • AnthropicBedrockとAnthropicBedrockMantleの取り違え: 新規プロジェクトはAnthropicBedrockMantleが前提で、AnthropicBedrockは既存のInvokeModel API利用アプリを移行させないための後方互換クライアントです
  • tool_runnerをベータ名前空間の外で探す: tool_runnerはclient.beta.messages配下にあり、client.messages直下には存在しません
  • max_tokensを大きくしたまま非ストリーミングで送る: SDKが所要時間を10分超と見積もるとValueErrorが発生します。長い出力が想定される処理はストリーミングに切り替えるか、タイムアウトを明示的に延ばします

まとめ

Python SDKはpip install anthropicで導入でき、Python 3.10以上が前提です。並列度が必要な非同期処理ではAsyncAnthropicにaiohttp拡張を組み合わせ、ツール呼び出しの往復を自作したくなければ@beta_toolとtool_runnerに任せるのが基本形になります。送信前のトークン見積もりにはcount_tokens、大量リクエストの非同期バルク処理にはMessage Batchesという役割分担です。同期・非同期のどちらを選ぶかで実装の見通しが大きく変わるため、着手前にワークロードの並列要件を確認しておくと手戻りが少なくなります。

TypeScript SDKとの実装上の違いを比較したい場合はClaudeのTypeScript SDKで実装するStreaming・Tool・MCPヘルパー、料金・モデル選択を含むAPI全体の見取り図はAnthropic API完全ガイドもあわせて参照してください。

この記事を共有:XはてブLinkedIn

関連する記事

Anthropic をもっと見る →