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

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

公式のPython SDKはpip install anthropicで導入します。Python 3.10以上が前提です。0.x系から上げる場合は、公式のv1マイグレーションガイドに破壊的変更の一覧があります。

プラットフォーム統合用のクライアントクラス自体は基本パッケージ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]"

このあとのコードは、空の仮想環境にSDK 1.11.0を入れ、実際に動かした結果に基づいています。

python3 -m venv .venv && .venv/bin/pip install anthropic
.venv/bin/python -c "import anthropic; print(anthropic.__version__)"
# 1.11.0

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-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-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-5",
        )
        print(message.content)
 
 
asyncio.run(main())

DefaultAioHttpClientは拡張を入れていない環境でもインポートできますが、インスタンスを作る時点でRuntimeErrorになります。メッセージは「aiohttpのextraを付けてインストールしてください」という趣旨です。

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

同期・非同期どちらのクライアントも、Server-Sent Events(SSE)によるストリーミングを同じインターフェースで扱えます。入口は2つあり、メモリ消費と扱いやすさが逆向きです。

async def main() -> None:
    async with client.messages.stream(
        max_tokens=1024,
        messages=[{"role": "user", "content": "Say hello there!"}],
        model="claude-opus-5-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())

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

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

count = client.messages.count_tokens(
    model="claude-opus-5-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-5",
    tools=[get_weather],
    messages=[{"role": "user", "content": "What is the weather in SF?"}],
)
for message in runner:
    print(message)

ツール実行ループの一般的な設計方針や他言語との比較はTool RunnerでAnthropic APIのツール呼び出しループを自動化するで扱っています。

@beta_toolが実際に作るスキーマはto_dict()で見られます。unit: str = "celsius"のようにデフォルト値付きの引数を足してSDK 1.11.0で出力すると、次のようになります。

print(json.dumps(get_weather.to_dict(), indent=2))

{
  "name": "get_weather",
  "description": "Get the weather for a given location.",
  "input_schema": {
    "additionalProperties": false,
    "properties": {
      "location": {
        "description": "The city and state, for example, San Francisco, CA",
        "title": "Location",
        "type": "string"
      },
      "unit": {
        "default": "celsius",
        "description": "celsius or fahrenheit",
        "title": "Unit",
        "type": "string"
      }
    },
    "required": ["location"],
    "type": "object"
  }
}

docstringのArgs:に書いた説明が各プロパティのdescriptionに入ります。デフォルト値を持つunitはrequiredに入りません。モデルに渡る説明文はdocstringで決まるため、引数の意味はdocstringに書いておきます。

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-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系という切り分けです。ステータスごとのサブクラスもあり、401はAuthenticationError、403はPermissionDeniedError、404はNotFoundErrorです。

例外クラスの継承関係はSDK 1.11.0で見ると次のとおりです。exceptの順番に直結します。

APITimeoutError    -> APIConnectionError -> APIError -> AnthropicError
RateLimitError     -> APIStatusError     -> APIError -> AnthropicError
InternalServerError-> APIStatusError     -> APIError -> AnthropicError
BadRequestError    -> APIStatusError     -> APIError -> AnthropicError

タイムアウトはAPIConnectionErrorの仲間で、APIStatusErrorには入りません。429を別処理にしたいときは、APIStatusErrorより先にRateLimitErrorを書きます。

接続エラー・408・409・429・500番台はデフォルトで2回まで指数バックオフ付きの自動リトライが入り、max_retriesで全体設定またはリクエスト単位の上書きができます。タイムアウトはデフォルト10分で、httpx2.Timeoutオブジェクトを渡せば接続・読み取り・書き込みごとに個別の秒数を設定できます。タイムアウトしたリクエストも既定で2回再試行されるため、既定値のままだと、最大10分×3回待つことがあります。

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

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-5",
)

大きなmax_tokensを非ストリーミングで指定するのは避けます。ネットワークによってはidle接続を一定時間で切断するものがあり、応答を受け取れないままリクエストが失敗する原因になります。SDKは非ストリーミングリクエストの所要時間が概ね10分を超えると判断した場合にValueErrorを送出します。

この判定は送信前にSDK側で行われます。到達できないURLをbase_urlに向け、claude-opus-5-5でmax_tokensだけを変えると、SDK 1.11.0では21,333までが通常どおり送信に進み、21,334でValueErrorになりました。

max_tokens=21333  -> APIConnectionError(送信に進んだ)
max_tokens=21334  -> ValueError
ValueError: Streaming is required for operations that may take longer than 10 minutes.

stream=Trueを渡すか、with_options(timeout=5.0)のようにtimeoutを上書きすると、max_tokens=128000でもこのエラーは出ません。境目の数値はモデルやSDKのバージョンで変わりうるため、大きな出力を扱う処理は最初からストリーミングにしておくと境目を意識せずに済みます。

デフォルトではSDKがidleタイムアウトの影響を減らすためTCP socket keep-aliveを設定しますが、独自のhttp_clientを渡すとこの挙動は上書きされます。プロキシやトランスポートを変えたいときの入口もhttp_clientで、公式の例ではDefaultHttpxClient(proxy=...)を渡します。素のhttpxのクライアントを渡すとTypeErrorになります。企業プロキシ配下でAPIConnectionErrorが出る場合の切り分けはAnthropicのPython SDKでプロキシ環境変数が反映されない問題を解決するにまとめました。

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-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello, Claude"}],
    betas=["context-management-2025-06-27"],
)

ベータ機能は一般提供前の早期フィードバック段階なので、利用できる機能一覧は随時変わります。対応状況はbuild with Claude overviewで確認できます。

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

プラットフォーム統合用のクライアントクラスは基本パッケージanthropicにすべて含まれています。違いは追加で入れる依存だけです。

AnthropicAWSはworkspace_idをコンストラクタか環境変数ANTHROPIC_AWS_WORKSPACE_IDで渡します。

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

よくあるつまずき

Anthropic(同期)のメソッドをasync def関数の中でそのまま呼ぶと、イベントループをブロックします。非同期文脈ではAsyncAnthropicに統一します。

まとめ

同期クライアントのまま始めて、並列度が足りなくなった処理だけをAsyncAnthropicとaiohttpに移す構成が最小です。クライアントクラスは動かす先のプラットフォームで決まるため、Bedrockなどへ移す可能性があるなら、最初にどの統合クラスを使うかだけ決めておくと後の差し替えが小さく済みます。

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