⛳
MCPホスト構築：非同期処理とリソース管理の基礎と実装

李碩根
2025/06/10に公開
Python
Model Context Protocol
AIエージェント
tech
この記事では、MCPホストを実際に作成するプロセスを通じて、その際に不可欠となる「非同期処理」の考え方と、通信セッションなどを安全に扱うための「リソース管理」のテクニックについて、基礎から具体的な実装までを解説します。記事では「MCPホスト」と「AIエージェント」の両方の表現が使われていますが、AIエージェントもMCPホストで実装されていることを前提に書いています。

MCPの概念については以下の記事を参考にしてください。
https://zenn.dev/mkj/articles/0ed4d02ef3439c
AIエージェントがMCPのような外部サーバと連携して動作する際には、必然的に以下のような課題に直面します。

例えば、AIエージェントが「今日の天気」を検索する場合、MCPサーバとの通信が発生し、その応答を待つ必要があります。
このとき、
通信処理における「待ち時間」をどう効率的に扱うか？

使用した通信路などのシステム「リソース」をどう安全に管理するか？（特に非同期処理の文脈で）
といった点が、AIエージェントの応答性や安定性を大きく左右します。
本記事では、これらの課題を解決し、AIエージェントを構築するために中心的な役割を果たす「非同期プログラミング（async/await）」と「（非同期）コンテキストマネージャによるリソース管理」の仕組みを深く掘り下げ、最終的にサンプルコードを基に、AIエージェント全体の処理フローの中でこれらの技術がどのように活用されるのかを具体的に見ていきます。
なお、この記事はPythonの基本的な文法は理解しているものの、非同期プログラミングやリソース管理についてはこれから学ぶ方、あるいは知識を整理したい方を主な対象としています。

そのため、理解を深めていただくことを目的としており、説明がやや長くなる箇所があります。リファレンスとして手早く情報を得たい方には不向きかもしれません。

 1. MCPサーバを使う時に発生する二つの課題1.1. 課題意識1：通信の「待ち時間」をどうするか？
AIエージェントがMCPサーバと通信する際、その基本的な方法はストリーム通信になっています。これは、データを連続的な流れとして送受信する方式で、例えばウェブブラウザがウェブサイトから情報を取得する際にも使われています。
ストリーム通信の性質上、データを送信してから相手の応答を受け取るまでには、必ず「待ち時間」が発生します。ツール呼び出しの結果を待つ間などがこれにあたります。
もし、この応答待ちの間にプログラム全体の実行が停止（ブロッキング）してしまうと、AIエージェントは新たなリクエストの受付や他のタスクの実行ができなくなり、全体の応答性が著しく低下します。
1.2. 課題意識2：使った「リソース」の後始末をどうするか？
MCPホストがMCPサーバとストリーム通信を行うためには、その通信路（ストリーム）を確立し維持するためのシステムリソース（メモリ、ファイルディスクリプタ、ネットワークソケットなど）を確保する必要があります。
これらのリソースは、使用後に適切に解放（クローズ）されなければ、メモリリークやリソース枯渇といった問題を引き起こし、システムの安定性を損なう可能性があります。特に、長時間稼働するAIエージェントにとっては深刻な問題であり、確実なリソース管理が求められます。
このような二つの課題は外部サーバとの通信を行うようなプログラムでは基本的な課題であり、その課題を解決するために一般的に使われる方法が「非同期処理」と「コンテキストマネージャーによるリソース管理」になります。
MCP SDKも同じくこの課題を「非同期処理」と「（非同期）コンテキストマネージャによるリソース管理」という2つの主要なプログラミング概念を活用しています。MCPを使ってAIエージェントを開発するには、この二つの概念を理解する必要があります。

 2. 解決策1：通信の「待ち」を効率化する ～非同期処理による並行動作～AIエージェントが外部のMCPサーバと通信する際には、応答を待つ「待ち時間」が必ず発生します。この「待ち時間」をいかに効率的に扱うかが、AIエージェントの応答性を高める上で重要です。本章では、この課題に対する一つ目の解決策として、非同期処理を用いたアプローチを詳しく見ていきます。
2.1. I/O処理における「待ち時間」の技術的背景
MCPサーバとの通信では応答待ちが発生し、AIエージェントの応答性低下に繋がる可能性がありました。この「待ち時間」は、技術的にはI/O（入出力）処理に起因します。I/O処理とは、プログラムがCPUで行う計算処理とは異なり、外部システムとのデータのやり取りを伴うものです。例えば、ネットワーク越しにサーバへリクエストを送信し、その応答を待つ場合などがこれにあたります。
このI/O処理は、CPUの計算速度と比較して非常に時間がかかることがあります。もし、プログラムがこのI/O処理の完了を同期的に（つまり、処理が終わるまで他のことを一切せずに）待つ場合、その間、プログラムの実行は停止（ブロッキング）してしまいます。結果として、AIエージェントは新たなリクエストの受付や他のタスクの実行ができなくなり、応答性が低下するのです。
2.2. 解決策：「非同期処理」によるI/O待ちの有効活用
このブロッキング問題を解決する主要なアプローチが 「非同期処理（Asynchronous Processing）」 です。
非同期処理の基本的な考え方は、時間のかかるI/O処理を開始したら、その完了を待たずにすぐに制御をプログラムの他の部分に戻し、別のタスクを実行できるようにすることです。そして、依頼しておいたI/O処理が完了した時点で、その結果を使って中断していた作業を再開します。
重要なのは、これは複数のCPUコアで同時に計算処理を行う「並列処理（Parallel Processing）」とは異なるという点です。非同期処理は、主に単一のプロセス（またはスレッド）内で、I/O処理の「待ち時間」を他の作業に充てることで、見かけ上の並行性を実現するテクニックです。CPUの計算処理がボトルネックとなるタスク（CPUバウンド）の高速化には直接的には向きませんが、I/O待ちが多いタスク（I/Oバウンド）には処理効率と応答性の向上に大きな効果があります。
2.3. 非同期処理のイメージとPythonでの道具立て
では、この「非同期処理」というアプローチを、より具体的にイメージし、Pythonで実現するにはどうすればよいのでしょうか？ここでは、まず身近な例え話を通して非同期処理の「考え方」を掴み、その後Pythonでこれを支える「道具立て」について見ていきましょう。
一人のシェフが複数の料理を効率よく作るイメージ

非同期処理を理解するために、一人の非常に有能なシェフ（プログラムの実行主体だと考えてください）が、キッチンで同時に複数の料理を注文された場面を想像してみましょう。
注文された料理には、例えば「じっくり煮込むビーフシチュー（煮込みに1時間かかる）」「野菜をオーブンでロースト（焼き時間に30分かかる）」「手早く作れるサラダ（数分で完成）」があるとします。
もし、このシェフが「ビーフシチューの煮込みが終わるまで、鍋の前で見守り続ける」「野菜のローストが終わるまで、オーブンの前で待ち続ける」というやり方（これを同期的な調理と呼びます）をしていたら、その長い待ち時間の間、他の料理（サラダなど）は一切進みません。これではお客様を待たせてしまいますね。
非同期的な調理法とは？
そこで、賢いシェフは次のように調理を進めます。
まず「ビーフシチュー」の材料を鍋に入れ、火にかけて煮込みを開始します。煮込みには時間がかかることを知っているので、タイマーをセットしたら、その場を離れて次の料理に取り掛かります。
次に「野菜のロースト」の準備をし、オーブンに入れます。これも焼き時間をタイマーにセットし、オーブンのことは一旦忘れて別の作業に移ります。
そして、空いた時間で「サラダ」を手早く作って完成させます。
そうこうしているうちに、キッチンタイマーが「ビーフシチューの煮込みが終わりましたよ！」「野菜のローストが焼き上がりましたよ！」と知らせてくれます。
通知を受けたら、シェフはサラダの盛り付けなど、もし他にすぐにできる作業があればそれを一旦中断（どこまで進んだか覚えておきます）し、煮込み終わったビーフシチューの火を止めたり、焼きあがった野菜をオーブンから取り出したりします。
そして、その用事が済んだら、また中断していた作業（例えばサラダの最後の仕上げなど）を再開します。
このように、一つの料理の調理工程（特に待ち時間が発生する工程）の完了を待っている間に、別の料理の準備や調理を進めることで、一人のシェフでも全体の作業がスムーズに進み、より多くのお客様に早く料理を提供できるようになります。これが非同期処理の基本的な考え方です。CPU（シェフ）は一つでも、I/O待ち（煮込み待ち、焼き待ちなど）の時間を有効活用することで、見かけ上、複数の調理が並行して進んでいるように見えるのです。
Pythonでこの「賢い調理法」を実現する道具たち
Pythonでこのような非同期的な処理を実現するために、いくつかの特別な「道具」や「役割」が用意されています。
中断・再開が可能な「特別な調理工程」： コルーチン (Coroutine)
先の例で、「ビーフシチューを煮込む」「野菜をローストする」「サラダを作る」といった個々の料理の調理工程が、途中で中断したり、後で再開したりできる性質を持っていました。Pythonでは、このように中断と再開が可能な処理の単位、つまり「特別な調理法が必要な料理のレシピやその作業工程」を「コルーチン」と呼びます。
コルーチンは、その作り方（手順書）を async def というキーワードを使って関数を定義することで示します。
調理の進行を管理し、タイマーの知らせを受ける「キッチンマネージャー」： イベントループ (Event Loop)
シェフが複数の料理を効率よく切り替えながら進めるためには、「どの料理が煮込み中で、どの料理が焼き上がりを待っていて、どの料理が今すぐ対応可能か」を管理し、タイマーが鳴ったらシェフに知らせる存在が必要です。先の例では、キッチン全体を見渡し、タイマーの音を聞いてシェフに指示を出す「キッチンマネージャー」がその役割を担っていました。
Pythonの非同期処理では、この「キッチンマネージャー」の役割を「イベントループ」が果たします。イベントループは、調理可能なコルーチン（料理の工程）を見つけて実行し、コルーチンが await（後述）で「待ち」に入ると、そのコルーチンを待機状態にします。そして、待っていた処理（例：煮込み完了、焼き上がり、asyncio.sleep の時間経過など）が完了すると、それを検知して待機していたコルーチンをシェフ（CPU）に再度割り当て、調理を再開させます。
非同期調理のための道具箱と基本調味料： asyncio ライブラリ
コルーチンという特別なレシピを定義したり、イベントループというキッチンマネージャーを動かしたり、あるいは「一定時間待つ（asyncio.sleep）」といった非同期調理でよく使われる基本的な機能や道具を提供してくれるのが、Pythonの標準ライブラリである asyncio です。async や await といったPythonのキーワードは、この asyncio が提供するイベントループの仕組みの上で機能します。
このように、「コルーチン」という中断・再開可能な特別な調理工程を、「イベントループ」というキッチンマネージャーがうまく切り替えながらシェフ（CPU）に実行させることで、待ち時間を有効活用し、効率的な処理が実現されるのです。
2.4. 実践：Pythonで非同期の「賢い調理法」を指示する – async def, await, asyncio.run() の役割
前のセクションで、非同期処理の背景にある「賢いシェフの働き方」と、それを支える「コルーチン」「イベントループ」「asyncioライブラリ」というPythonの道具立てを見ました。では、これらの道具を使って、Pythonのコード上でシェフに具体的な調理指示を出すにはどうすればよいのでしょうか。それには主に async def、await、そして asyncio.run() という3つのキーワードや関数が深く関わっています。
import asyncio # まずは asyncio ライブラリ（非同期調理の道具箱）を使えるようにする

# (1) 「特別な調理法のレシピ」を作るための印： async def
# これで定義された関数（コルーチン関数）を呼び出すと、
# 「その料理を作るための具体的なレシピカード（コルーチンオブジェクト）」が手に入る。
async def prepare_ingredient_after_delay(ingredient_name, delay_seconds):
    print(f"シェフ：『{ingredient_name}』の下ごしらえ開始。{delay_seconds}秒かかります（待ち）。")
    # (2) 「特定の調理工程（ここでは待機）を開始し、終わるまで待つ」指示： await
    # asyncio.sleep(delay_seconds) もコルーチンオブジェクトを返す特別な作業。
    # await はその作業を開始させ、完了するまでこの prepare_ingredient_after_delay の進行を一時停止する。
    await asyncio.sleep(delay_seconds)
    prepared_item = f"下ごしらえ完了：{ingredient_name}"
    print(f"シェフ：『{prepared_item}』ができました！")
    return prepared_item

# これもコルーチン関数。一連の非同期な調理工程をまとめたメインレシピ。
async def main_cooking_flow():
    print("シェフ：本日の調理を開始します！")
    # prepare_ingredient_after_delay("ニンジン", 2) はコルーチンオブジェクト（レシピカード）を返す。
    # await はそのレシピカードに従って調理（下ごしらえ）を開始し、完了を待つ。
    carrot = await prepare_ingredient_after_delay("ニンジン", 2)
    # 同様に「タマネギ」の下ごしらえも行う。
    onion = await prepare_ingredient_after_delay("タマネギ", 1)
    print(f"シェフ：すべての下ごしらえが完了しました！ 材料：『{carrot}』、『{onion}』")
    # ここでこれらの材料を使って、さらに料理を進めることができる...

# (3) 「本日の調理、開始！」の号令とキッチン全体の管理： asyncio.run()
if __name__ == "__main__":
    # asyncio.run() がキッチンマネージャー（イベントループ）を起動し、
    # main_cooking_flow というメインレシピ（コルーチンオブジェクト）に従って調理を開始するよう指示する。
    asyncio.run(main_cooking_flow())
このコードがどのように連携して、シェフが賢く調理を進めるかを見ていきましょう。
async def：特別な調理法の「レシピ」を定義する
関数定義の先頭に async def を付けると、それは「これは特別な調理法（途中で中断・再開できる）で作る料理のレシピを定義するコルーチン関数ですよ」という宣言になります。
このコルーチン関数（例：prepare_ingredient_after_delay）を呼び出すと、例えば recipe_card = prepare_ingredient_after_delay("ニンジン", 2) のように書くと、関数の中身がすぐに実行される（調理が始まる）わけではありません。代わりに、recipe_card 変数には、「"ニンジン"を2秒かけて下ごしらえする」という作業を行うための具体的な 「レシピカード（コルーチンオブジェクト）」 が格納されます。これはまだ開始されていない「料理の設計図」や「注文票」のようなものです。
await：レシピカードに従って調理を開始し、特定の工程の完了を待つ
await キーワードは、手元にあるコルーチンオブジェクト（レシピカード）に書かれた調理を実際に開始させ、その特定の調理工程が完了するまで、現在の調理作業（await を書いた側のコルーチン）の進行を一時停止させるために使います。await はコルーチン関数 (async def で定義された関数) の中でしか使えません。
例えば、main_cooking_flow コルーチンの中の carrot = await prepare_ingredient_after_delay("ニンジン", 2) という行を見てみましょう。
まず prepare_ingredient_after_delay("ニンジン", 2) が呼び出され、前述の通り「ニンジンの下ごしらえレシピカード（コルーチンオブジェクト）」が作られます。
次に await がこのレシピカードを受け取り、「このレシピカードに書かれた下ごしらえを開始してください。そして、その下ごしらえが完了するまで、この main_cooking_flow の調理はここで一時停止します」とキッチンマネージャー（イベントループ）に伝えます。
prepare_ingredient_after_delay の中で await asyncio.sleep(delay_seconds) が実行される際も同様です。asyncio.sleep() も「一定時間待つ」という特別な調理工程（コルーチン）であり、その完了を await が待ちます。
重要なのは、この await による「一時停止（例：ニンジンの下ごしらえ待ち）」中に、キッチンマネージャー（イベントループ）は、もし他にすぐに着手できる調理工程（別のコルーチン）があれば、シェフ（CPU）にそちらの作業を進めさせることができるという点です。これにより、シェフは待ち時間を無駄にしません。
asyncio.run()：「本日の調理、開始！」の号令とキッチン全体の管理
async def でレシピが定義され、await で個々の調理工程の進め方が指示できるようになったら、いよいよキッチン全体を動かし始める必要があります。そのための「開始の号令」と、全体の進行管理を行うのが asyncio.run() 関数です。
コードの最後の asyncio.run(main_cooking_flow()) は、以下の処理を行います。
キッチンマネージャー（新しいイベントループ）を準備します。
引数として渡されたメインレシピ（この場合は main_cooking_flow() が返すコルーチンオブジェクト）を、そのキッチンマネージャーに渡し、「このレシピに従って調理を開始してください」と指示します。
そのメインレシピに書かれた全ての調理工程が完了するまで（そして、そこから await で呼び出される全てのサブ工程が完了するまで）、キッチンマネージャーはシェフ（CPU）の作業を管理し続けます。
全ての調理が終わったら、キッチンマネージャーは後片付けをして業務を終了します。
これは通常、Pythonの非同期プログラム（非同期な調理場）の「営業開始」の合図として使われます。
このように、async def で「非同期に実行できる特別な調理法のレシピ」を定義し、そのレシピカード（コルーチンオブジェクト）をawait で「実際に調理を開始させて特定の工程の完了を待ち」、そして asyncio.run() で「キッチン全体の非同期な調理オペレーションを開始する」 という流れで、Pythonの非同期プログラムは成り立っています。
この「一人のシェフが複数の料理を効率的に作る」というイメージと、それを実現するためのPythonの道具（async def, await, asyncio.run()）の関係性を掴むことが、MCP SDKのような asyncio をベースとしたライブラリを理解し、使いこなすための重要な一歩となるでしょう。

 3. 解決策2：使った道具の後始末を確実に行う ～コンテキストマネージャによる安全なリソース管理～AIエージェントがMCPサーバと通信を行う際には、通信路を確保したり、データを一時的に保持したりするために、コンピュータの様々な「リソース」を使います。これは、シェフが調理をする際に、特別な鍋やフライパン、あるいは一時的に借りる高価な食材といった「調理道具や材料」を使うのに似ています。
これらの道具や材料は、使い終わったら必ず元の場所に戻したり、きれいに洗浄したり、あるいは適切に廃棄したりといった「後始末」が必要です。もし後始末を忘れてしまうと、キッチンが散らかって次の調理ができなくなったり（リソース枯渇）、食材が腐ってしまったり（メモリリークによる不具合）と、大変なことになってしまいます。特に、長時間動き続けるAIエージェント（レストランのように一日中営業するキッチン）にとっては、この「後始末」が非常に重要になります。
この章では、このような「後始末」を、特に2.章で学んだ非同期的な作業（複数の調理を並行して行う状況）の中で、安全かつ確実に行うための「コンテキストマネージャ」という賢い仕組みについて見ていきましょう。
3.1. 調理道具の「後始末」忘れ：なぜ問題なのか？
シェフが特別な調理器具（例えば、滅多に使わない高性能ミキサーや、借りてきた真空調理器など）を使ったとしましょう。使い終わった後、それを洗浄せずに放置したり、借りたものを返し忘れたりしたらどうなるでしょうか？
次にその調理器具を使いたい他のシェフが使えなくなります。
器具が汚れたまま放置されれば、衛生的にも問題ですし、器具自体が傷んでしまうかもしれません。
借りたものは、返さないと次の人が使えないだけでなく、ペナルティが発生するかもしれません。
プログラムにおける「リソース」も同様です。ファイルを開いたら必ず閉じる、ネットワーク接続を確立したら必ず切断する、といった後始末は非常に重要です。もし、調理の途中で急なトラブルが発生したり（プログラムのエラー）、作業手順が複雑でうっかり後始末を忘れてしまったりすると、これらのリソースが解放されないままになり、システム全体の動作に悪影響を与えてしまいます。
特に非同期処理（シェフが複数の調理を同時にこなす状況）では、どの道具がどの料理で使われていて、いつ後始末が必要なのかを正確に把握することが、より一層難しく、かつ重要になります。
3.2. 解決策：「後始末おまかせサービス」としてのコンテキストマネージャ
この「後始末忘れ」を防ぐための非常に賢い仕組みが、Pythonにおける「コンテキストマネージャ」です。
コンテキストマネージャを例えるなら、「特別な調理器具の貸し出し・返却カウンターの親切な係員」のようなものです。

貸し出し時（前処理）: シェフが特定の調理器具（リソース）を使いたいと言うと、この係員は器具を貸し出す前に、使い方や注意事項を説明したり、必要な準備（例えば、器具の電源を入れるなど）をしてくれます。

返却時（後処理）: シェフが調理器具を使い終わると（あるいは、使っている途中で何かトラブルがあっても）、この係員が必ず器具の状態を確認し、洗浄したり、電源を切ったり、元の棚に戻したりといった後始末を代行してくれます。シェフは「使い終わりました」と伝えるだけでOKです。
この「係員」のおかげで、シェフは「後始末を絶対に忘れないようにしなければ！」と常に気を張る必要がなくなり、調理そのものに集中できます。たとえ調理中に予期せぬトラブルが発生しても、係員が後始末はちゃんとしてくれるので安心です。
非同期処理（複数の調理を並行して行う）の場合も、この「係員」はちゃんと対応してくれます。 その場合は、貸し出しや返却の手続き自体が少し時間のかかるもの（例えば、他のシェフが使っている器具が空くのを待つなど）になるかもしれませんが、その間もシェフは他の作業を進めることができます。これを実現するのが 「非同期コンテキストマネージャ」 で、Pythonでは async with という特別な書き方で利用します。
3.3. コンテキストマネージャの仕組み：調理器具を使う時の「お約束」
この「後始末おまかせサービス（コンテキストマネージャ）」は、プログラム上では、特定の調理器具（リソース）を使う際の「お約束の範囲」を明確にすることで機能します。
Pythonでは、この「お約束の範囲」を with 文（同期処理の場合）や async with 文（非同期処理の場合）を使って示します。
# 同期的な調理器具の例（例：特別なまな板）
with get_special_cutting_board() as board:
    # この "with" のブロック内が「まな板を使っているお約束の範囲」
    # board.cut_vegetables()
    # board.slice_meat()
    pass # このブロックを抜け出すと、自動的に「まな板の返却・清掃」が行われる

# 非同期的な調理器具の例（例：予約が必要な特殊オーブン）
async with reserve_special_oven() as oven:
    # この "async with" のブロック内が「オーブンを使っているお約束の範囲」
    # await oven.bake_bread()
    # await oven.roast_chicken()
    pass # このブロックを抜け出すと、自動的に「オーブンの使用終了手続き・清掃」が行われる
この with や async with の裏側では、リソースの準備（「貸し出し時の仕事」）と解放（「返却時の仕事」／後始末）が、対応する特別なメソッドによって自動的に行われます。

リソースの準備（前処理）: with や async with のブロックが始まる際に実行されます。

同期 (with): __enter__() メソッドが呼び出されます。
例：ファイルを開いて、ファイルオブジェクトを取得する。 (file = open('config.txt', 'r')) この処理は完了するまで待機する。


非同期 (async with): async def __aenter__() メソッドが await されながら呼び出されます。
例：データベースへの接続を非同期に確立し、接続オブジェクトを取得する。 (connection = await async_db_library.connect()) 接続確立を待つ間、他の処理が進む可能性がある。



リソースの解放（後処理／後片付け）: ブロックを抜ける際、処理が正常に完了した場合でも、途中で例外が発生した場合でも、必ず実行されます。これは try...finally 文の finally 節の動作に似ています。

同期 (with): __exit__() メソッドが呼び出されます。
例：開いていたファイルオブジェクトを閉じる。 (file.close())


非同期 (async with): async def __aexit__() メソッドが await されながら呼び出されます。
例：データベース接続を非同期に閉じる。 (await connection.close()) 接続クローズ処理を待つ間、他の処理が進む可能性がある。


シェフ（プログラマ）は、通常、これらの __enter__ や __exit__（あるいは __aenter__, __aexit__）を直接呼び出す必要はありません。with や async with を使うだけで、Pythonが自動的に適切なタイミングでこれらを実行してくれるのです。これにより、リソースの確保と解放のロジックをより綺麗にカプセル化でき、リソース管理が非常に簡単かつ安全になります。
3.4. Pythonコードでの実践：async with と調理道具の安全な後始末
MCP SDKおよびその利用例では、この非同期コンテキストマネージャの仕組みが積極的に活用されています。なぜなら、MCP SDKは非同期処理（シェフが複数の調理を並行して行う）を基本としているため、調理道具（リソース）の管理もその非同期の流れの中で安全に行う必要があるからです。
3.4.1. 個々の調理道具の安全な貸し出しと返却
MCP SDKの中には、例えばサーバとの通信セッション（特定の調理器具を使うための「利用権」のようなもの）を管理するクラスがあります。これらのクラスは、非同期コンテキストマネージャとして設計されており、async def __aenter__ と async def __aexit__ メソッドを持っています。
async def __aenter__ (貸し出し時の非同期手続き):
セッションを開始するための準備（例えば、サーバとの接続に必要なバックグラウンドタスクを非同期的に開始する）を行います。シェフが async with SomeSession(...) as session: と書くと、この __aenter__ が呼び出されます。
async def __aexit__ (返却時の非同期手続き):
セッションを終了する際の後片付け（例えば、開始したバックグラウンドタスクをキャンセルし、接続を非同期的に閉じる）を行います。シェフが async with ブロックを抜けると、この __aexit__ が自動的に呼び出され、後始末が確実に行われます。
これにより、プログラマは async with を使うだけで、セッションの開始処理と終了処理（後始末）が自動的かつ安全に行われることを保証され、リソースリークの心配を大幅に減らすことができます。
3.4.2. 複数の調理道具をまとめて、安全に後始末する：AsyncExitStack
AIエージェントのような複雑なアプリケーション（大規模なレストランの厨房）では、一度に複数の異なる調理道具（例えば、複数のMCPサーバへの接続セッションなど）を同時に、かつ非同期的に管理し、使い終わったら全てを確実に後始末する必要が出てくることがあります。
このような、複数の非同期コンテキストマネージャ（後始末が必要な調理道具）をまとめて、かつ安全に管理するための便利な道具として、Pythonの contextlib モジュールにある AsyncExitStack というものがあります。
これを例えるなら、 「複数の特別な調理器具を一度に借りて、使い終わったら全てをまとめて安全に返却・清掃してくれるベテランのキッチンアシスタント」 のようなものです。
シェフは、このアシスタントに「このミキサーと、あのオーブンと、そこの特殊鍋を使いたいんだけど、後でまとめてちゃんと片付けておいてね」とお願いするイメージです。
from contextlib import AsyncExitStack
# （他の必要な import 文 ... ClientSession や stdio_client など）

async def manage_multiple_kitchen_tools():
    async with AsyncExitStack() as stack: # ベテランアシスタントを呼ぶ
        # アシスタントに「このミキサー（stdio_client）を借りて、使えるようにして」とお願い
        mixer_streams = await stack.enter_async_context(stdio_client(...))
        # アシスタントに「このオーブン（ClientSession）も借りて、使えるようにして」とお願い
        oven_session = await stack.enter_async_context(ClientSession(mixer_streams[0], mixer_streams[1]))

        # これで mixer_streams や oven_session が使える状態
        # await oven_session.initialize()
        # result = await oven_session.call_tool(...)
        print("シェフ：ミキサーとオーブンを使って調理中...")

        # async with stack: のブロックを抜ける時、
        # アシスタント（AsyncExitStack）が、借りた順番と逆の順番で、
        # oven_session の後始末（__aexit__）、
        # mixer_streams の後始末（__aexit__）を自動的に行ってくれる。
    print("シェフ：すべての調理器具の後始末が完了しました。")
AsyncExitStack() で「ベテランアシスタント」を呼び出し、async with ... as stack: でアシスタントの活動範囲を決めます。
await stack.enter_async_context(...) を使って、個々の非同期コンテキストマネージャ（調理道具の利用権など）をアシスタントに登録し、利用可能な状態にしてもらいます（各道具の __aenter__ が呼び出されます）。
そして、親となる async with stack: ブロックを抜け出す際に、このアシスタント（AsyncExitStack）が、登録した全ての調理道具の「後始末係の仕事（__aexit__ メソッド）」を、登録した時とは逆の順番で、一つ一つ丁寧に呼び出してくれるのです。
AsyncExitStack を利用することで、非同期処理の中で複数のリソースの確保と解放のロジックを一箇所にまとめ、コードの見た目をスッキリさせ、後始末の安全性をさらに高めることができます。
このように、MCP SDKでは、非同期処理を効率的に行うために asyncio を基盤とし、その上で大切な調理道具（リソース）を安全に管理するために非同期コンテキストマネージャ (async with) や AsyncExitStack といった仕組みを活用しています。この2つの考え方（効率的な並行作業と、確実な後始末）は、堅牢で信頼性の高いAIエージェント（いつでも美味しい料理を提供できる素晴らしいレストラン）を構築するために、連携して機能するのです。

 4. MCPホストの構築2.章および3.章では、MCPクライアントがMCPサーバと通信し、LLMと連携するための基本的な処理フローと、それを支える非同期処理やコンテキスト管理といったコア技術について学びました。
この4.章では、これらの知識を基に、具体的なMCPホストの実装コードを通じて、MCPを用いたAIエージェントが実際にどのように機能するのか、その「概念」と「仕組み」を深く掘り下げていきます。アーキテクチャ図やシーケンス図で示した各要素や処理ステップが、コード上でどのように表現され、連携しているのかを確認しましょう。
全体のコードはここを参照にしてください。
https://github.com/suckgeun/mcp_sample.git
4.1. AIエージェントの全体像：アーキテクチャと処理の流れ
まず、サンプルコードが実現しようとしているAIエージェントの全体構造と、主要な動作の流れを再確認します。
4.1.1. アーキテクチャ：MCPにおける登場人物とその役割
MCPのアーキテクチャは「ホスト」「クライアント」「サーバ」という3つの主要な役割で構成されていました。host/src/mcp_host_tutorial_openai.py は、この中の「ホスト」としての役割を担います。
図1に、本チュートリアルで構築するAIエージェントのアーキテクチャ概要を示します
図1: AIエージェントのアーキテクチャ概要

AIエージェント ホストプロセス (mcp_host_tutorial_openai.py):
全体の司令塔です。ユーザーとの対話を受け付け、LLM（LLMServer）に判断を仰ぎます。
そして、LLMの指示に基づき、特定のMCPサーバ（MCPServer1, MCPServer2）と通信するためのMCPクライアント（図中の MCPクライアント 1, MCPクライアント 2）を実行します。
各MCPクライアントは、対応するMCPサーバとの間でMCPセッションを確立し、ツール実行などのやり取りを行います。mcp_host_tutorial_openai.py では、ClientSession クラスがこのセッション管理を担っています。

4.1.2. 主要処理フロー：AIエージェントの思考と行動
次に、このアーキテクチャの上で、AIエージェントがどのように思考し行動するのか、その一連の流れ（シーケンス）を見てみましょう。
図2に、MCPホストの典型的な処理フローを示します。
図2: AIエージェントの主要処理フロー
このシーケンスは、AIエージェントがMCPツールを利用する際の典型的な流れを示しています。

初期化フェーズ（1-3）: エージェント起動時に、接続可能なMCPサーバからツール情報を収集し、LLMが利用できる形に整えます。

LLM連携フェーズ（4-6）: ユーザーのリクエストを受け、準備したツール情報と共にLLMに判断を仰ぎます。

MCPツール実行フェーズ（7a-12a）: LLMがツール利用を指示した場合、該当するMCPサーバのツールを実行し、その結果を再度LLMにフィードバックして最終応答を得ます。
mcp_host_tutorial_openai.py のコードは、これらの概念と処理フローを具体的に実装しています。それでは、コードの主要部分がこれらの概念とどのように結びついているのかを見ていきましょう。
4.2. 初期化：MCPサーバとの接続とツールカタログの準備
AIエージェントが起動する際、まず行うべきは、連携するMCPサーバの情報を把握し、実際に接続して利用可能なツールの一覧を取得することです。これはシーケンス図のステップ1～3に相当します。
4.2.1. MCPサーバ情報の定義と準備 (RAW_CONFIG, build_servers)
mcp_host_tutorial_openai.py では、まず RAW_CONFIG という辞書で、接続対象のMCPサーバとその起動方法を定義しています。
# (mcp_host_tutorial_openai.py より抜粋)
RAW_CONFIG: Dict[str, dict] = {
    "fetch": {"command": "uvx", "args": ["mcp-server-fetch"]},
    "google_search": { # ← これがサーバ識別子 (server_name)
        "command": "python",
        "args": ["path/to/your/google_search_server.py"], # ← 起動コマンドと引数
    },
}
この RAW_CONFIG は、build_servers 関数によって、より扱いやすい MCPServer オブジェクトの辞書に変換されます。各 MCPServer オブジェクトは、サーバ名、起動コマンド、引数といった情報を保持し、さらに後述する init_servers 関数内で、確立された ClientSession （MCPセッションの実体）を session 属性として持つことになります。
4.2.2. MCPセッションの確立とツール情報収集 (init_servers 関数)
init_servers 関数が、この初期化処理の中核です。この関数は、アーキテクチャ図の「ホストプロセス」が、各「MCPクライアントロジック」を介して「MCPサーバ」と「MCPセッション」を確立し、ツール情報を収集するという一連の流れを実装しています。
# (mcp_host_tutorial_openai.py より抜粋)
async def init_servers(
    stack: AsyncExitStack, servers: Dict[str, MCPServer] # servers は build_servers の結果
) -> List[dict]:
    openai_tools: List[dict] = []
    for server_obj in servers.values(): # 各MCPServerオブジェクトに対して処理
        # 1. MCPサーバプロセスの起動と通信チャネルの確立 (シーケンス図 ステップ1の一部)
        #    stdio_client を使い、サーバプロセスを起動。
        #    AsyncExitStack (stack) でリソース管理。
        read_stream, write_stream = await stack.enter_async_context(
            stdio_client(
                StdioServerParameters(
                    command=server_obj.command, args=server_obj.args, env=server_obj.env
                )
            )
        )
        # 2. MCPセッションの確立と ClientSession オブジェクトの保存
        #    ClientSession がMCPセッションのプロトコル処理を担当。
        #    これも AsyncExitStack でリソース管理。
        #    確立したセッションを server_obj.session に保存。
        server_obj.session = await stack.enter_async_context(
            ClientSession(read_stream, write_stream)
        )
        # 3. MCPセッションの初期化 (シーケンス図 ステップ1の残り & ステップ2)
        await server_obj.session.initialize()

        # 4. 利用可能なツール情報の要求と取得 (シーケンス図 ステップ2)
        response = await server_obj.session.list_tools()
        
        # 5. ツール情報をLLM向けに整形 (シーケンス図 ステップ3)
        for mcp_tool_spec in response.tools:
            openai_tools.append(
                mcp_tool_to_openai_tool(mcp_tool_spec, server_obj.name)
            )
    return openai_tools
この関数では、2.章および3.章で学んだ非同期処理 (async/await) が全面的に活用されています。stdio_client の呼び出し、ClientSession の作成、initialize()、list_tools() といったMCPサーバとの通信を伴う処理は全て非同期に行われ、複数のサーバに対する初期化処理を効率的に進めることができます。
また、AsyncExitStack を使うことで、これらの処理で確保されるリソース（サーバプロセス、通信ストリーム、ClientSession オブジェクト）が、init_servers を呼び出している chat_loop 関数の async with stack: ブロックを抜ける際に、確実に解放されることが保証されます。
4.2.3. ツール情報のLLM向け整形 (mcp_tool_to_openai_tool 関数)
init_servers 関数内で呼び出される mcp_tool_to_openai_tool 関数は、MCPサーバから取得したツール情報（MCPTool オブジェクト）を、LLM（この場合はOpenAI）のFunction Calling (Tool Use)機能が理解できるJSONスキーマ形式に変換します。
# (mcp_host_tutorial_openai.py より抜粋)
def mcp_tool_to_openai_tool(tool: MCPTool, server_name: str) -> dict:
    # ツール名にサーバ識別子を付加して一意性を確保
    unique_name = f"{server_name}{TOOL_SEPARATOR}{tool.name}"
    return {# OpenAI Function Calling のスキーマ形式
        "type": "function",
        "name": unique_name,
        "description": tool.description,
        "parameters": tool.inputSchema,
    }
ここで重要なのは、LLMに提示するツール名 (unique_name) を、サーバ識別名 + TOOL_SEPARATOR + MCPツール本来の名前 という形式で生成している点です。これにより、異なるMCPサーバが同じ名前のツールを提供していても、LLMはそれらを区別して扱えます。
この初期化フェーズが完了すると、AIエージェントは「どのMCPサーバに、どのようなツールがあり、それをLLMにどう伝えればよいか」という情報を全て手に入れた状態になります。
4.3. LLMとの連携とMCPツールの実行：対話の中核 (chat_loop と dispatch_tool_call)
初期化が完了すると、AIエージェントはユーザーとの対話を開始します。chat_loop 関数がこの対話ループ全体の制御を担います。
4.3.1. LLMへの判断依頼 (シーケンス図 ステップ4-6)
chat_loop 関数内では、まずユーザーからの入力を受け付けます。そして、その入力と、init_servers で準備したLLM向けのツールカタログ（tools_for_llm）を、OpenAIクライアント (client) を使ってLLMに送信します。
# (mcp_host_tutorial_openai.py の chat_loop 関数より抜粋)
async def chat_loop(servers: Dict[str, MCPServer]) -> None:
    # ... (OpenAIクライアント初期化、AsyncExitStackとinit_servers呼び出し) ...
    # tools_for_llm には初期化されたLLM向けツールスキーマのリストが入る

    while True: # ユーザーとの対話ループ
        user_text = await asyncio.to_thread(input, "You: ") # ユーザー入力
        # ... (終了処理) ...

        # LLM API呼び出しのための引数準備
        call_kwargs = {
            "model": MODEL_NAME,
            "messages": [{"role": "user", "content": user_text}], # OpenAI APIの形式
            "tools": tools_for_llm, # 準備したツールカタログ
        }
        # ... (previous_id の処理) ...

        # LLMにリクエストを送信し、応答を得る
        response: Response = client.responses.create(**call_kwargs) # OpenAI API呼び出し
        # ... (応答の処理へ続く) ...
ここで client.responses.create(**call_kwargs) を呼び出すことで、シーケンス図のステップ6が実行され、LLMはユーザーのリクエストと利用可能なツール情報を基に、次のアクションを判断します。
4.3.2. LLMの応答とMCPツールの実行判断 (シーケンス図 ステップ7)
LLMからの応答 (response) には、ツールを利用すべきかどうかの指示が含まれています。mcp_host_tutorial_openai.py では、応答の output[0] が ResponseFunctionToolCall のインスタンスであるかどうかでこれを判断しています。
      # (mcp_host_tutorial_openai.py の chat_loop 関数より抜粋)
            # ... (LLM API呼び出しの後) ...
            # LLMがツール利用を指示したかどうかをチェック
            while isinstance(response.output[0], ResponseFunctionToolCall):
                tool_call_instruction = response.output[0] # ツールコール指示オブジェクト
                
                # シーケンス図 ステップ7a (ツール指示を解析) & 8a (MCPツール実行)
                # dispatch_tool_call 関数を呼び出してMCPツールを実行
                tool_output_str = await dispatch_tool_call(tool_call_instruction, servers)
                
                # シーケンス図 ステップ10a: LLMにツール結果を報告
                # ツール実行結果をLLMにフィードバックするための新しいリクエストを作成
                response = client.responses.create(
                    model=MODEL_NAME,
                    previous_response_id=response.id, # 文脈維持
                    input=[{
                        "type": "function_call_output",
                        "call_id": tool_call_instruction.call_id,
                        "output": tool_output_str,
                    }],
                    tools=tools_for_llm,
                )
            # ループを抜けたら、response.output[0] はテキスト応答のはず
            # シーケンス図 ステップ11a (最終応答) または 7b (直接応答)
            assistant_message_text = response.output[0].content[0].text
            print(f"Assistant: {assistant_message_text}\n")
LLMがツール利用を指示した場合（isinstance が True）、dispatch_tool_call 関数が呼び出されます。
4.3.3. MCPツールの実行と結果の取得 (dispatch_tool_call 関数)
dispatch_tool_call 関数は、LLMからのツールコール指示を受け取り、実際にMCPツールを実行します。これはシーケンス図のステップ8aと9aに相当します。
# (mcp_host_tutorial_openai.py より抜粋)
async def dispatch_tool_call(
    tool_call: ResponseFunctionToolCall, servers: Dict[str, MCPServer]
) -> str:
    args = json.loads(tool_call.arguments) # LLMが生成した引数をパース
    
    # LLMからのツール名 (例: "google_search__search_web") を
    # サーバ識別子 ("google_search") とツール名 ("search_web") に分割
    server_name, mcp_tool_name = tool_call.name.split(TOOL_SEPARATOR)
    
    # init_servers で保存しておいたアクティブな ClientSession を取得
    active_session = servers[server_name].session 
    
    # ClientSession を使ってMCPツールを非同期に呼び出す
    result = await active_session.call_tool(name=mcp_tool_name, arguments=args)
    
    # 結果を文字列として返す (簡略化)
    return str(result.content[0].text) if result.content else "Tool returned no content."
この関数は、まずLLMが指定した一意化されたツール名 (tool_call.name) を TOOL_SEPARATOR で分割し、対象の server_name と mcp_tool_name を特定します。そして、servers 辞書（init_servers で ClientSession が格納されたもの）から該当する active_session を取り出し、await active_session.call_tool(...) を使ってMCPツールを非同期に実行します。
ここでも、MCPサーバとの通信は await を使った非同期処理として行われます。
4.3.4. ツール実行結果のLLMへのフィードバックと最終応答の生成
dispatch_tool_call からツール実行結果が返されると、chat_loop はその結果をLLMにフィードバックします（シーケンス図 ステップ10a）。LLMはツール結果を考慮して、最終的なテキスト応答を生成するか（シーケンス図 ステップ11a, 12a）、あるいはさらに別のツール利用を指示することもあります。このやり取りは、LLMが直接テキストで応答するまで繰り返されます。
この一連の対話ループを通じて、AIエージェントはユーザーの要求を理解し、MCPツールを効果的に活用し、最終的な応答を生成します。2.章および3.章で学んだ非同期処理は、LLM API呼び出しやMCPツール呼び出しといったI/Oバウンドな処理の待ち時間を有効活用し、エージェント全体の応答性を高めるのに貢献しています。
4.4. エントリポイントと全体の実行フロー
最後に、mcp_host_tutorial_openai.py のスクリプトがどのように開始され、これまでに説明した各関数がどのように連携するのかを見てみましょう。
# (mcp_host_tutorial_openai.py より抜粋)
def build_servers(raw: Dict[str, dict]) -> Dict[str, MCPServer]:
    return {name: MCPServer(name=name, **cfg) for name, cfg in raw.items()}

def main() -> None:
    # 1. RAW_CONFIG から MCPServer オブジェクトの辞書を生成
    servers_map = build_servers(RAW_CONFIG)
    # 2. メインの非同期対話ループ chat_loop を asyncio.run で実行
    #    chat_loop 内で init_servers が呼ばれ、MCPセッションが確立される
    asyncio.run(chat_loop(servers_map))

if __name__ == "__main__":
    main()
スクリプトが実行されると、まず main() 関数が呼び出されます。
main() 関数内では、build_servers(RAW_CONFIG) が呼び出され、生のサーバ設定情報が型付けされた MCPServer オブジェクトの辞書 (servers_map) に変換されます。この時点では、まだMCPサーバとの接続は行われていません。
次に、asyncio.run(chat_loop(servers_map)) が実行されます。これがAIエージェントの非同期処理全体の開始点です。
asyncio.run() はイベントループを起動し、引数として渡されたコルーチン (chat_loop(servers_map)) の実行を開始します。
chat_loop コルーチンが実行されると、その冒頭で await init_servers(...) が呼び出され、ここで初めてMCPサーバとの接続、セッション確立、ツール情報収集が行われます。この init_servers 内の処理（stdio_client や ClientSession の利用、initialize、list_tools の呼び出し）は全て非同期に実行されます。
init_servers が完了すると、chat_loop はユーザーとの対話を開始し、必要に応じて dispatch_tool_call を介してMCPツールを非同期に実行します。

このように、mcp_host_tutorial_openai.py は、設定の読み込みから、非同期イベントループの開始、MCPサーバとの初期接続、そしてユーザーとの対話とツール実行という一連の流れを、各関数が連携して実現しています。
4.5. 環境設定と実行
gitのREADMEを参照にしてください。
4.6. まとめ
このチュートリアルでは、MCP SDK を用いてAIエージェントを構築する際の基本的な「考え方」と「実践方法」を学びました。
学んだことの振り返り:

AIエージェントにおける2つの課題: 外部サーバとの通信における「待ち時間」の効率化と、システム「リソース」の安全な管理。

非同期処理 (async/await): シェフの例えを通して、I/Oバウンドな処理の待ち時間を有効活用し、AIエージェントの応答性を高める仕組みを理解しました。async def でコルーチンを定義し、await で非同期処理の完了を待ち、asyncio.run() でイベントループを開始する方法を学びました。

コンテキストマネージャ (async with): 調理道具の後始末の例えを通して、ファイルやネットワーク接続などのリソースを確実に解放する重要性と、async with を使った安全なリソース管理方法を学びました。

AsyncExitStack: 複数の非同期コンテキストマネージャをまとめて管理し、後始末を確実に行うための便利なツールであることを理解しました。

MCP連携の基本フロー:
MCPサーバとの接続初期化とツールカタログの取得・整形。
ユーザーリクエストとツールカタログをLLMに送信し、判断を仰ぐ。
LLMの指示に基づき、MCPツールを非同期に実行。
ツール実行結果をLLMにフィードバックし、最終応答を生成。


実装コード (mcp_host_tutorial_openai.py): 上記の概念が実際のコードでどのように実装されているかを確認しました。
松尾研究所テックブログ
株式会社松尾研究所のテックブログです。
MCPホスト構築：非同期処理とリソース管理の基礎と実装

1. MCPサーバを使う時に発生する二つの課題

2. 解決策1：通信の「待ち」を効率化する～非同期処理による並行動作～

3. 解決策2：使った道具の後始末を確実に行う～コンテキストマネージャによる安全なリソース管理～

4. MCPホストの構築

Discussion

1. MCPサーバを使う時に発生する二つの課題

2. 解決策1：通信の「待ち」を効率化する ～非同期処理による並行動作～

3. 解決策2：使った道具の後始末を確実に行う ～コンテキストマネージャによる安全なリソース管理～

4. MCPホストの構築

Discussion

2. 解決策1：通信の「待ち」を効率化する～非同期処理による並行動作～

3. 解決策2：使った道具の後始末を確実に行う～コンテキストマネージャによる安全なリソース管理～
