AnyIOとは？Pythonでの使い方とasyncio・Trioとの違い【4.15対応】

AnyIO（パッケージ名 anyio）は、同じコードをasyncioとTrioのどちらのイベントループでも動かせるPythonの非同期ライブラリです。FastAPIの土台であるStarletteは、同期関数をスレッドで動かす処理やバックグラウンドタスクの実行をanyioで実装しており、FastAPIを使っていれば知らないうちにanyioが動いています。この記事では最新の4.15.1で実行したコードを使って、インストールからタスクグループ・タイムアウト・スレッド連携・テストまでの使い方と、asyncio・Trioとの違い、どれを選ぶかの判断基準を説明します。

まとめ：AnyIOの位置づけと選び方

• AnyIOはasyncioかTrioの上で動く抽象レイヤーで、Trio由来の「タスクグループ」と「キャンセルスコープ」をasyncio上でも使えるようにします。最新は4.15.1（2026年9月5日）、Python 3.10以上が必要です。
• 4.14でタスクの戻り値を受け取れるTaskHandleとcreate_task()、4.15でgather()・amap()・as_completed()が入り、asyncioからの書き換えがしやすくなりました。
• asyncioとの最大の差はキャンセルの方式です。キャンセル例外を一度握りつぶすと、asyncioのタスクは止まらずに残りますが、AnyIOでは次の待機点で再びキャンセルされます（実測で確認）。
• ライブラリを配布する、FastAPI／Starlette上で同期コードやスレッドを扱う、キャンセル例外の握りつぶしで中断が効かなくなる事故を防ぎたい場合はAnyIOを選びます。依存を増やしたくない単一アプリならasyncio、新規開発で依存ライブラリもTrioに対応しており、タスク管理をTrioのAPIに統一したいならTrioです。
• 速度で選ぶライブラリではありません。今回の計測では、待機時間ゼロのタスク2万件を生成して終了を待つ時間は、AnyIOの方がasyncio.TaskGroupより長くなりました。I/O待ちが中心の処理での影響は、このタスク起動ベンチマークだけでは判断できません。

AnyIOとは：asyncioとTrioの両方で動く非同期ライブラリ

AnyIOはAlex Grönholm氏が開発するMITライセンスのライブラリで、タスクグループ、キャンセルスコープ、TCP/UDP/UNIXソケット、ロックやイベントなどの同期プリミティブ、ワーカースレッドとサブプロセス、非同期ファイルI/O、pytestプラグインを1つのAPIで提供します。実行時にどちらのイベントループの上で呼ばれたかを判定し、asyncio用またはTrio用の実装に振り分けます。かつて対応していたCurioバックエンドは3.0で外れ、現在のバックエンドはasyncioとTrioの2つです。非同期処理そのものの仕組みから押さえたい場合は、非同期処理とは？同期処理との違いから実装方式まで実装者目線で解説を先に読むと流れがつかみやすくなります。

AnyIOを直接使わなくても、依存関係には入っていることが多いライブラリです。Starletteのconcurrency.pyではrun_in_threadpool()がanyio.to_thread.run_sync()を呼んでおり、FastAPIのdefで書いたエンドポイントもこの経路でスレッドに渡されます。Starlette 1.10.0はanyio>=4.0.0,<5を、HTTPクライアントのhttpxもanyioを依存に持っています。

最近の版で増えた機能（4.12〜4.15）

AnyIOはここ1年ほどで、asyncioから移る人向けのAPIを増やしています。公式のバージョン履歴から、使い方に影響する変更を抜き出しました。

4.13以前を前提にした解説やサンプルでは、start_soon()は値を返さず、タスクの戻り値は自前で集めるという説明になっています。4.14以降はこの前提が変わったので、古いコードを写すときは版を確認してください。

AnyIOのインストールとバックエンド指定

anyio本体だけならasyncioバックエンドで動きます。Trioでも動かしたい場合はextrasを付けて入れると、対応する版のTrioが一緒に入ります（4.15.1の要件はtrio>=0.32.0）。

pip install anyio            # asyncioバックエンドのみ
pip install "anyio[trio]"    # Trioバックエンドも使う

エントリポイントはanyio.run()です。asyncio.run(main())と違い、コルーチンではなく関数オブジェクトmainを渡します。backendを省略するとasyncioで、backend="trio"でTrioに切り替わります。asyncioでuvloopを使う場合はbackend_options={"use_uvloop": True}を渡します。

タスクグループによる並行実行と戻り値の取得

AnyIOでタスクを起動する方法は、タスクグループ（Trioでいうナーサリー）だけです。async withのブロックを抜けるときに中のタスクの終了をすべて待ち、どれかが例外を投げれば残りをキャンセルして、例外をExceptionGroupにまとめて送出します。起動したまま放置されるタスクが構造上できません。

import anyio

async def fetch(name: str, delay: float) -> str:
    await anyio.sleep(delay)
    return f"{name}:{delay}"

async def main() -> None:
    async with anyio.create_task_group() as tg:
        h1 = tg.start_soon(fetch, "A", 0.2)     # 4.14以降はTaskHandleが返る
        h2 = tg.create_task(fetch("B", 0.1))    # asyncioと同じ書き方（4.14で追加）
    print(h1.return_value, h2.return_value)

anyio.run(main)                    # asyncioバックエンド（既定）
anyio.run(main, backend="trio")    # Trioバックエンド

4.15.1ではどちらのバックエンドでもA:0.2 B:0.1と表示されました。start_soon()は関数と引数を別々に、create_task()はコルーチンを渡します。公式のタスク解説によると、TaskHandleはタスクの完了待ち（await handle）、戻り値や例外の取得、個別のキャンセル（handle.cancel()）、状態の確認に使えます。

gather・amap・as_completedによる結果の収集

結果を順番どおりに集めたいだけなら、4.15で入ったユーティリティ関数が短く書けます。いずれも内部でタスクグループを使うため、1つが失敗すれば残りはキャンセルされます。

import anyio

async def square(n: int) -> int:
    await anyio.sleep(0.1 * (3 - n))
    return n * n

async def main() -> None:
    print(await anyio.gather(square(1), square(2)))   # (1, 4)
    print(await anyio.amap(square, [0, 1, 2]))        # [0, 1, 4]
    async with anyio.as_completed(square(0), square(2)) as results:
        async for handle in results:                  # 終わった順に届く
            print("done", handle.return_value)        # done 4 → done 0

anyio.run(main)

gather()はタプル、amap()はリストを返し、どちらも引数の順序を保ちます。as_completed()は待ち時間の短いsquare(2)から届きました。1秒かかるタスクと0.1秒でValueErrorを出すタスクをanyio.gather()に渡すと、0.1秒でExceptionGroupが送出されました。asyncio.gather()は1つが失敗しても他のタスクを止めず、最初の例外をそのまま送出するので、置き換えるときは個別の例外型を捕捉する処理も見直します。Python 3.11以降ではexcept*を使えますが、Python 3.10ではexceptiongroup.catch()を使います。

TaskGroup.start()による初期化完了の待機

サーバーを起動してから接続するクライアントを動かす、といった順序の制御にはawait tg.start(func)を使います。起動される側がtask_status.started()を呼ぶまで呼び出し元が待ち、初期化中の例外（ポートのbind失敗など）は呼び出し元へ送られます。asyncio.TaskGroupにはこの仕組みがなく、asyncio.Eventなどを自分で組む必要があります。

タイムアウトとキャンセルスコープ

タイムアウトはwithブロックの範囲に掛けます。move_on_after()は時間切れで黙ってブロックを抜け、fail_after()はTimeoutErrorを送出します。4.15からはreasonで例外にメッセージを付けられます。

import anyio

async def main() -> None:
    with anyio.move_on_after(0.5) as scope:
        await anyio.sleep(10)
    print("move_on_after:", scope.cancelled_caught)    # True

    try:
        with anyio.fail_after(0.5, reason="外部APIが応答しない"):
            await anyio.sleep(10)
    except TimeoutError as exc:
        print("fail_after:", exc)                       # 外部APIが応答しない

anyio.run(main)

このブロックの実体がCancelScopeで、scope.cancel()で手動キャンセル、CancelScope(shield=True)で後片付けの処理を外側のキャンセルから守れます。期限を絶対時刻で指定したい場合は4.15で入ったmove_on_at()とfail_at()を使い、基準の時刻はanyio.current_time()で取得します。

同期コードとスレッドを扱う：to_threadとfrom_thread

requestsや同期版のDBドライバのように、ブロックする関数をイベントループの中で直接呼ぶと、その間ほかのタスクがすべて止まります。anyio.to_thread.run_sync()でワーカースレッドへ逃がし、スレッド側からイベントループの処理を呼ぶときはanyio.from_threadを使います。

import time
import anyio
from anyio import from_thread, to_thread

def blocking_io(n: int) -> int:
    time.sleep(0.5)                                   # 同期ライブラリの代役
    from_thread.run_sync(print, f"thread {n} done")   # イベントループ側で実行
    return n

async def main() -> None:
    print(to_thread.current_default_thread_limiter().total_tokens)   # 40
    async with anyio.create_task_group() as tg:
        for i in range(3):
            tg.start_soon(to_thread.run_sync, blocking_io, i)

anyio.run(main)

0.5秒の処理3本は約0.65秒で終わり、スレッドで並行に動いたことが分かります。公式のスレッド解説のとおり、to_thread.run_sync()の既定の同時実行数は40です。FastAPIのdefエンドポイントもこの上限を共有するため、同期処理の多いAPIでスレッド待ちが発生するときは、DB接続数やメモリ使用量を確認したうえで、必要に応じて起動時にcurrent_default_thread_limiter().total_tokensを引き上げます。この上限はasyncio自体のrun_in_executor()のスレッドプールには効きません。

BlockingPortalによる同期コードからの非同期関数呼び出し

逆向き、つまりイベントループの外にある同期コード（CLIや同期フレームワークのビュー）から非同期関数を呼ぶ場合は、start_blocking_portal()で別スレッドにイベントループを立てます。

import anyio
from anyio.from_thread import start_blocking_portal

async def add(a: int, b: int) -> int:
    await anyio.sleep(0.1)
    return a + b

with start_blocking_portal(backend="asyncio") as portal:
    print(portal.call(add, 1, 2))    # 3

StarletteのTestClientも、同期のテストコードから非同期のアプリを呼ぶためにこの仕組みを使っています。

メモリオブジェクトストリームによるタスク間のデータ受け渡し

タスク間のデータの受け渡しには、asyncio.Queueの代わりにメモリオブジェクトストリームを使います。送信側と受信側が別オブジェクトに分かれ、送信側を閉じると受信側のasync forが終わります。asyncio.Queueは既定で上限なしですが、こちらはmax_buffer_sizeの既定が0で、受け手がいなければ送信側が待たされるため、メモリを食いつぶしません。

import anyio

async def producer(send) -> None:
    async with send:                  # 閉じると受信側のループが終わる
        for i in range(3):
            await send.send(i)

async def consumer(receive) -> None:
    async with receive:
        async for item in receive:
            print("received", item)

async def main() -> None:
    send, receive = anyio.create_memory_object_stream[int](max_buffer_size=1)
    async with anyio.create_task_group() as tg:
        tg.start_soon(producer, send)
        tg.start_soon(consumer, receive)

anyio.run(main)

複数の送り手がいる場合はsend.clone()で送信側を複製します。受信側の反復を終了させるには、複製した送信側だけでなく、元の送信側も閉じる必要があります。単発の通知にはanyio.Event（set()とawait event.wait()）を、1回だけ値を受け渡すには4.15で入ったanyio.Futureを使います。

pytestによる非同期関数のバックエンド別テスト

AnyIOはpytestプラグインを同梱しており、pytest-asyncioを入れなくてもasync defのテストを実行できます。anyio_backendフィクスチャをパラメータ化すると、同じテストをasyncioとTrioの両方で回せます。

# conftest.py
import pytest

@pytest.fixture(params=["asyncio", "trio"])
def anyio_backend(request):
    return request.param

# test_sample.py
import anyio
import pytest

pytestmark = pytest.mark.anyio

async def test_sleep_returns_quickly():
    with anyio.fail_after(1):
        await anyio.sleep(0.01)

pytest 9.2.0では1つのテストが2件（asyncio・Trio）として実行され、両方とも成功しました。anyio_backendを上書きしなければ、既定のフィクスチャが対象とするバックエンドで実行され、Trioを入れた今回の環境ではconftest.pyを外しても同じく2件になりました。asyncioだけに限定する場合は、このフィクスチャが"asyncio"を返すように定義します。fixtureやparametrizeなどpytest側の書き方はpytestの使い方と書き方｜fixture・parametrize・docstringの扱いまで【9.1対応】で解説しています。

AnyIO・Trio・asyncioの違い

Python 3.15のasyncio.TaskGroup.cancel()追加と残る差異

Trioのナーサリーが示したモデルは、Python 3.11でasyncio.TaskGroupとasyncio.timeout()として標準に入りました。さらに2026年10月9日に公開されたPython 3.15では、asyncio.TaskGroupにcancel()が追加され、グループ全体を例外なしで打ち切れるようになりました。3.15.0で実行するとブロックを正常に抜け、3.12ではAttributeErrorになります。グループ内タスクの一括キャンセルはもうAnyIOだけの機能ではありません。

それでも残る差は、AnyIO公式の比較によると、start()による起動完了待ちが無いこと、開始前にキャンセルされたタスクはキャンセル例外を受け取る機会がないこと、1つのタスクの失敗で終了処理に入ったグループには新しいタスクを足せないこと、そして次に述べるキャンセル方式の違いです。

キャンセル方式の違い（edge cancellationとlevel cancellation）

asyncioはキャンセル時にCancelledErrorを1回だけ送るエッジ型です。except BaseException:のような広い例外処理で誤って握りつぶすと、そのタスクは止まらずに走り続けます。TrioとAnyIOはレベル型で、キャンセルされたスコープの中にいる限り、次の待機点で何度でも例外が送られます。AnyIO公式の解説にある例をもとに、両方を5秒の制限付きで実行しました。

import asyncio

async def sleeper():
    try:
        await asyncio.sleep(1)
    except BaseException:
        print("1回目のキャンセルを握りつぶした")
    await asyncio.sleep(float("inf"))    # 2回目のキャンセルは来ない

async def main():
    async with asyncio.TaskGroup() as tg:
        task = tg.create_task(sleeper())
        await asyncio.sleep(0.1)
        task.cancel()
    print("done")                        # 表示されない

asyncio.run(main())

asyncio版はPython 3.12・3.15.0ともに「1回目のキャンセルを握りつぶした」を表示したまま止まり、5秒の制限で強制終了しました。同じ処理をAnyIOで書き、task.cancel()をtg.cancel()に替えると、asyncio・Trioどちらのバックエンドでもdoneまで進みます。キャンセルに対応した非同期処理では、この違いが中断制御に役立ちます。ただし、ワーカースレッド内の同期処理は強制停止できません。外部APIやDBドライバにも個別のタイムアウトを設定してください。

AnyIO・asyncio・Trioの選定基準

判断の軸は、配布するかどうか、既存のasyncio資産の量、キャンセル制御の複雑さの3つです。

• AnyIOを選ぶ：ライブラリやSDKを配布する（利用者のイベントループを選ばない）、FastAPI／Starlette上でスレッドやバックグラウンド処理を扱う、タイムアウトや中断が入り組んだ処理を書く。FastAPIとは？最短の起動手順とasync/defの使い分け、0.162の変更点のとおりFastAPIはanyio上で動くので、追加の依存も増えません。
• asyncioのままでよい：単一のアプリで、タスクの起動・待ち合わせ・タイムアウトがTaskGroupとasyncio.timeout()で足りる。Python 3.15に上げられるなら一括キャンセルも標準で書けます。
• Trioを選ぶ：新規のプロジェクトで、使いたいライブラリがTrioに対応している。タスクの寿命をナーサリーで管理する設計に統一できます。採用前に、必要なDBドライバやSDKがTrioで動作するかを確認してください。

避けるべきなのは、asyncio前提のライブラリ（多くのDBドライバやSDK）を抱えたチームがTrio単独に移ることです。trio-asyncioでの橋渡しが必要になり、移行の手間に見合いません。その場合はAnyIOをasyncioバックエンドで使い、既存のasyncioライブラリと併用できます。ただし、そのライブラリが独自に生成するasyncioタスクにはAnyIOのキャンセル規則が適用されないため、タスク生成方法も確認してください。もう一つ、何もしない短いタスクを数万単位で起動する処理（細かいメッセージの扇形展開など）は、次のFAQの実測どおりAnyIOの起動コストが目立つため、asyncioのTaskGroupで書いた方が速く済みます。

非同期APIの設計や、同期ライブラリが混在する既存システムの移行を外部に相談したい場合は、一創のフルスクラッチ開発で要件整理から対応しています。

よくある質問

anyioとasyncioはどちらが速いですか？

タスクの起動はasyncioの方が速いです。sleep(0)するだけのタスクを2万件起動して終了を待つ時間を、Core i9-9880H・Python 3.12.13で計測したところ、asyncio.TaskGroupが0.55〜0.95秒、AnyIO（asyncioバックエンド）が2.35〜2.47秒、AnyIO（Trioバックエンド）が2.70〜2.79秒でした。各方式を5回実行して最短値を採る測定を2セット行い、各セットの最短値の範囲を示しています。計測差をタスク数で割ると1件あたり約0.1ミリ秒ですが、これは今回の条件での平均です。実際のネットワークやDB処理への影響は、タスク数・生成頻度・待ち時間を含めて測定する必要があります。

「NoEventLoopError: Not currently running on any asynchronous event loop」はなぜ出ますか？

anyio.current_time()やanyio.create_task_group()など、実行中のイベントループを必要とする同期APIを、イベントループの外（普通の同期関数やモジュールの読み込み時）で呼んだときに出ます。4.12.1から例外名がanyio.NoEventLoopErrorに統一されました。呼び出しをanyio.run()の中に移すか、同期コードから呼ぶ必要があるならstart_blocking_portal()を経由してください。

「The anyio.abc.BlockingPortal alias is deprecated」の警告はどう直しますか？

この警告は4.15.0から出るようになりました。from anyio.abc import BlockingPortalをfrom anyio.from_thread import BlockingPortalに書き換えます。4.15.1では古いimportでもクラス自体は同じものが返り、DeprecationWarningが出るだけです。自分のコードに見当たらない場合は、依存ライブラリ側のimportが原因なので、まずそのライブラリに修正版が公開されていないか確認し、互換性を確認して更新します。修正版がない場合に限り、対象の警告を限定して一時的に抑制します。

FastAPIでanyioを直接使う場面はありますか？

あります。async defのエンドポイントの中で同期ライブラリを呼ぶときにawait anyio.to_thread.run_sync(func)でスレッドへ逃がす、スレッド上限を引き上げる、並行して複数の外部APIを呼ぶためにタスクグループを使う、の3つが代表例です。サーバー側の起動や本番設定はUvicornとは？読み方・Gunicornとの違いと起動・本番設定【0.54対応】を参照してください。

Python 3.9でanyioは使えますか？

最新版は使えません。4.13.0でPython 3.9のサポートが終わり、3.9に対応する最後の版は4.12.1です。3.9環境ではpip install "anyio<4.13"で版を固定できますが、4.14以降のTaskHandleやgatherは使えないため、Python本体の更新を先に検討してください。

関連記事

• 非同期処理とは？同期処理との違いから実装方式まで実装者目線で解説
• FastAPIとは？最短の起動手順とasync/defの使い分け、0.162の変更点
• Uvicornとは？読み方・Gunicornとの違いと起動・本番設定【0.54対応】
• FastAPIのベストプラクティス｜ディレクトリ構成・SQLAlchemy 2.1の設計・テストを実測で解説
• pytestの使い方と書き方｜fixture・parametrize・docstringの扱いまで【9.1対応】