LoginSignup
5
3

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@junkmdin株式会社ディーバ

【Python】MCPサーバーを実装する際のテスト戦略案: インメモリクライアントセッション経由で自動テストしてサーバーを走らせるより前にバグに気づこう

Posted at

はじめに

近年、生成AI/LLMは、強力なテキスト生成能力や推論能力を持つようになりました。一方で、現実世界や外部システムとインタラクションするためには、「ツール」を呼び出して実行する能力が不可欠です。

2024年11月にAnthropicが発表したModel Context Protocol (MCP)は、このようなAI/LLMが外部のツールを利用するための標準的なインターフェース(オープンプロトコル)として注目を集めています。
Python用のSDKであるmodelcontextprotocol/python-sdk(以下「MCP Python SDK」)もAnthropicから提供され、AI/LLMからの指示に応じて別々のデータソースやツールに接続する機能があるMCPサーバーを効率的に実装できるようになっています。

生成AI/LLMは、その本質において確率的な要素を含んでおり、常に完全に予測可能で安定した出力を保証することは難しい側面があります。しかし、MCPサーバーについては従来のプログラミングによって実装される要素が大きいため、せめてこの部分だけでもその動作が予測可能で入力に対して常に同じ結果を返す再現性のあるものであることを厳密に担保したいと考える私のような開発者がいると思います。

コードの正しさを確認する古典的かつ基本的な手段として、個々の関数やメソッドを隔離してテストする「単体テスト」が有効だと考えられています。この考え方をMCPサーバーのテストにも適用し、各機能を単体テストすることで予測可能性と再現性を保証したい、というのは自然な流れだと考えています。

しかし、MCP Python SDKを用いて開発するMCPサーバーについて、テスト手法はまだ完全に確立されておらず、広く利用されるベストプラクティスや支援機能が整備されている段階とは言えません。 例えば、Django開発においてパッケージ内にdjango.test.Clientのようなテスト用クライアントが提供されていたり、pytest-djangoがclientフィクスチャを提供しているような、手軽で包括的なテスト基盤がMCP Python SDKにおいてはまだ十分ではありません。

現状、MCPサーバー開発者はテスト用の仕組みをある程度自前で構築する必要に迫られています。

本稿では、MCPサーバー開発におけるテストの課題を具体的な例と共に示し、MCP Python SDKが提供する関数を利用してテストコード側で必要な仕組みを構築することで、より実践的かつ信頼性の高いテストを実現する手法を紹介します。

MCPサーバーのプロダクションコード例

今回テストの対象とするのは、以下のような、MCP Python SDK用いて同期ツールgreetと非同期ツールaddを定義したコードであると想定します。同期ツールとして定義されたgreet関数では内部で非同期関数say_helloをasyncio.runを使用して呼び出していす。

mcpserver.py(オリジナル)
import asyncio

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("myserver", log_level="ERROR")


async def say_hello(name: str) -> str:
    await asyncio.sleep(0.05)
    return f"Hello, {name}!"


@mcp.tool()
def greet(name: str) -> str:
    """挨拶します。（同期ツール）"""
    return asyncio.run(say_hello(name))


@mcp.tool()
async def add(a: int, b: int) -> int:
    """足し算します。（非同期ツール）"""
    await asyncio.sleep(0.05)
    return a + b


if __name__ == "__main__":
    mcp.run()

関数単体テストの落とし穴

テストコードの例

これらのツール関数greetやaddはMCPフレームワークのコンテキストから切り離して、下記のようにPythonの関数として単体テストすることが可能です。

tests/test_mcpserver.py(関数版)
import asyncio

import mcpserver


class Test_Functions:
    def test_greet(self):
        assert mcpserver.greet("you") == "Hello, you!"

    def test_add(self):
        assert asyncio.run(mcpserver.add(1, 2)) == 3

このテストを実行すると問題なく成功します。

関数版テスト結果(成功) 
> uv run pytest tests\test_mcpserver.py -s -v
myprj\.venv\Lib\site-packages\pytest_asyncio\plugin.py:217: PytestDeprecationWarning: The configuration option "asyncio_default_fixture_loop_scope" is unset.
The event loop scope for asynchronous fixtures will default to the fixture caching scope. Future versions of pytest-asyncio will default the loop scope for asynchronous fixtures to function scope. Set the default fixture loop scope explicitly in order to avoid unexpected behavior in the future. Valid fixture loop scopes are: "function", "class", "module", "package", "session"

  warnings.warn(PytestDeprecationWarning(_DEFAULT_FIXTURE_LOOP_SCOPE_UNSET))
platform win32 -- Python 3.11.2, pytest-8.3.5, pluggy-1.5.0 -- myprj\.venv\Scripts\python.exe
cachedir: .pytest_cache
rootdir: myprj
configfile: pyproject.toml
plugins: anyio-4.9.0, asyncio-0.26.0
asyncio: mode=Mode.STRICT, asyncio_default_fixture_loop_scope=None, asyncio_default_test_loop_scope=function
collected 2 items                                                                                                                                                  

tests/test_mcpserver.py::Test_Functions::test_greet PASSED
tests/test_mcpserver.py::Test_Functions::test_add PASSED

======================================================================= 2 passed in 1.33s ========================================================================

しかし、このテストが検証しているのは、純粋な関数としての入力/出力のロジックのみです。
MCPサーバーとしての動作は、このテストでは全く検証されていません。

サーバー実行時に問題が発覚する

実際にツールが呼び出されることを想定して、先ほどのコード(mcpserver.py)をテスト&デバッグ用開発者ツールMCP Inspectorで起動します。
開発モードでMCPサーバーを走らせるには以下のコマンドを使用します。

mcp dev mcpserver.py

コマンド実行後表示される🔍 MCP Inspector is up and running at ... 🚀にブラウザ上でアクセスすると、テストツールUIが表示されます。
(fastapiにおけるSwagger UIの関係性と相似だと考えています)

Inspectorからgreetツールを呼び出すと、以下のようなエラーが発生してツールが正常に実行できません。

"Error executing tool greet: asyncio.run() cannot be called from a running event loop"

image.png

これはまさに「テストは通るのに本番で動かない」という問題の典型的な例です。
MCP Python SDKで動作するMCPサーバーは、クライアントからのリクエストを処理するために自身で非同期イベントループを内部で実行しています。
@mcp.tool()でデコレートされた関数は、このサーバーが管理するイベントループ上で実行されます。greet関数の中でasyncio.run()が呼び出された際、既にイベントループが稼働している状態であったため、「既に実行中のイベントループがあるのに、さらに別のイベントループを開始しようとした」という非同期実装の扱いに起因するエラーが発生したのです。
生成AIの不確実性に加え、ツールが正常に動作するかまでサーバーを実際に動かさないとわからなくなってしまうのは避けたい事態です。

クライアントセッション版テストで落とし穴を回避する

テストコードの例

このような、単体テストでは見つけられず実際のサーバーを動かしたとき初めて明らかになる非同期関連の問題を検出するためには、より現実に近い状況をエミュレートする実践的なテスト手法が必要です。MCPサーバーで言えばMCPツール関数を「単なるPython関数」としてではなく「MCPサーバー上でクライアントからのリクエストを受けて実行されるもの」としてテストすることがそれに当たります。

MCP Python SDKはMCPサーバーとのクライアントセッションをインメモリで張る関数が実装されています。それらを活用しつつテストコード側で必要な「テスト用の仕組み」を構築することで、実際のサーバーの振る舞いを模倣したテストが実現できます。以下のテストコードはこの手法を用いたものです。

非同期関数(コルーチン)をテストするためpytest-asyncioが必要になります。

テスト結果に表示されているPytestDeprecationWarning: The configuration option "asyncio_default_fixture_loop_scope" is unset.を抑制するには下記をpyroject.tomlに追加するか相応の設定が必要です。

pyproject.toml
[tool.pytest.ini_options]
asyncio_default_fixture_loop_scope = "function"
tests/test_mcpserver.py(クライアントセッション版)
import contextlib
from collections.abc import AsyncIterator

import pytest
from mcp.client.session import ClientSession
from mcp.shared.memory import (
    create_connected_server_and_client_session as create_client_session,
)
from mcp.types import TextContent

import mcpserver


@contextlib.asynccontextmanager
async def create_client() -> AsyncIterator[ClientSession]:
    """テスト用のクライアントセッション生成をDRY化するためのasync context manager"""
    # MCP Python SDKが提供する関数を使用してテストする
    # これは本家のテストでも使われているインメモリでサーバーとクライアントセッションを接続する手法。ユースケースは下記を参照のこと。
    # https://github.com/modelcontextprotocol/python-sdk/blob/main/tests/server/fastmcp/test_server.py
    async with create_client_session(mcpserver.mcp._mcp_server) as client:
        yield client


class Test_MCPServer:
    @pytest.mark.asyncio
    async def test_greet(self):
        # 自前で定義したcontext managerを使用してクライアントセッションを取得する
        async with create_client() as client:  # クライアントセッション経由でツールを呼び出す
            # 元を同期関数として実装していても、awaitを使って呼び出す
            res = await client.call_tool("greet", {"name": "you"})
            cnt = res.content[0]
            assert isinstance(cnt, TextContent)
            assert cnt.text == "Hello, you!"

    @pytest.mark.asyncio
    async def test_add(self):
        async with create_client() as client:
            # 非同期でも同様にawaitを利用して呼び出す。結果的に元の実装が同期か非同期かに関わらず同じテスト手法を利用できる。
            res = await client.call_tool("add", {"a": 1, "b": 2})
            cnt = res.content[0]
            assert isinstance(cnt, TextContent)
            assert cnt.text == "3"

このクライアントセッション版テストを実行してみるとtest_greetが下記のように失敗します。

プロダクションコード修正前クライアントセッション版テスト結果(失敗) 
> uv run pytest tests\test_mcpserver.py -s -v
myprj\.venv\Lib\site-packages\pytest_asyncio\plugin.py:217: PytestDeprecationWarning: The configuration option "asyncio_default_fixture_loop_scope" is unset.
The event loop scope for asynchronous fixtures will default to the fixture caching scope. Future versions of pytest-asyncio will default the loop scope for asynchronous fixtures to function scope. Set the default fixture loop scope explicitly in order to avoid unexpected behavior in the future. Valid fixture loop scopes are: "function", "class", "module", "package", "session"

  warnings.warn(PytestDeprecationWarning(_DEFAULT_FIXTURE_LOOP_SCOPE_UNSET))
====================================================================== test session starts =======================================================================
platform win32 -- Python 3.11.2, pytest-8.3.5, pluggy-1.5.0 -- myprj\.venv\Scripts\python.exe
cachedir: .pytest_cache
rootdir: myprj
configfile: pyproject.toml
plugins: anyio-4.9.0, asyncio-0.26.0
asyncio: mode=Mode.STRICT, asyncio_default_fixture_loop_scope=None, asyncio_default_test_loop_scope=function
collected 2 items

tests/test_mcpserver.py::Test_MCPServer::test_greet FAILED
tests/test_mcpserver.py::Test_MCPServer::test_add PASSED

============================================================================ FAILURES ============================================================================ 
___________________________________________________________________ Test_MCPServer.test_greet ____________________________________________________________________ 
  + Exception Group Traceback (most recent call last):
  |   File "myprj\.venv\Lib\site-packages\_pytest\runner.py", line 341, in from_call
  |     result: TResult | None = func()
  |                              ^^^^^^
  |   File "myprj\.venv\Lib\site-packages\_pytest\runner.py", line 242, in <lambda>
  |     lambda: runtest_hook(item=item, **kwds), when=when, reraise=reraise
  |             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  |   File "myprj\.venv\Lib\site-packages\pluggy\_hooks.py", line 513, in __call__
  |     return self._hookexec(self.name, self._hookimpls.copy(), kwargs, firstresult)
  |            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  |   File "myprj\.venv\Lib\site-packages\pluggy\_manager.py", line 120, in _hookexec
  |     return self._inner_hookexec(hook_name, methods, kwargs, firstresult)
  |            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  |   File "myprj\.venv\Lib\site-packages\pluggy\_callers.py", line 139, in _multicall
  |     raise exception.with_traceback(exception.__traceback__)
  |   File "myprj\.venv\Lib\site-packages\pluggy\_callers.py", line 122, in _multicall
  |     teardown.throw(exception)  # type: ignore[union-attr]
  |     ^^^^^^^^^^^^^^^^^^^^^^^^^
  |   File "myprj\.venv\Lib\site-packages\_pytest\threadexception.py", line 92, in pytest_runtest_call
  |     yield from thread_exception_runtest_hook()
  |   File "myprj\.venv\Lib\site-packages\_pytest\threadexception.py", line 68, in thread_exception_runtest_hook
  |     yield
  |   File "myprj\.venv\Lib\site-packages\pluggy\_callers.py", line 122, in _multicall
  |     teardown.throw(exception)  # type: ignore[union-attr]
  |     ^^^^^^^^^^^^^^^^^^^^^^^^^
  |   File "myprj\.venv\Lib\site-packages\_pytest\unraisableexception.py", line 95, in pytest_runtest_call
  |     yield from unraisable_exception_runtest_hook()
  |   File "myprj\.venv\Lib\site-packages\_pytest\unraisableexception.py", line 70, in unraisable_exception_runtest_hook
  |     yield
  |   File "myprj\.venv\Lib\site-packages\pluggy\_callers.py", line 122, in _multicall
  |     teardown.throw(exception)  # type: ignore[union-attr]
  |     ^^^^^^^^^^^^^^^^^^^^^^^^^
  |   File "myprj\.venv\Lib\site-packages\_pytest\logging.py", line 846, in pytest_runtest_call
  |     yield from self._runtest_for(item, "call")
  |   File "myprj\.venv\Lib\site-packages\_pytest\logging.py", line 829, in _runtest_for
  |     yield
  |   File "myprj\.venv\Lib\site-packages\pluggy\_callers.py", line 122, in _multicall
  |     teardown.throw(exception)  # type: ignore[union-attr]
  |     ^^^^^^^^^^^^^^^^^^^^^^^^^
  |   File "myprj\.venv\Lib\site-packages\_pytest\capture.py", line 898, in pytest_runtest_call
  |     return (yield)
  |             ^^^^^
  |   File "myprj\.venv\Lib\site-packages\pluggy\_callers.py", line 122, in _multicall
  |     teardown.throw(exception)  # type: ignore[union-attr]
  |     ^^^^^^^^^^^^^^^^^^^^^^^^^
  |   File "myprj\.venv\Lib\site-packages\_pytest\skipping.py", line 257, in pytest_runtest_call
  |     return (yield)
  |             ^^^^^
  |   File "myprj\.venv\Lib\site-packages\pluggy\_callers.py", line 103, in _multicall
  |     res = hook_impl.function(*args)
  |           ^^^^^^^^^^^^^^^^^^^^^^^^^
  |   File "myprj\.venv\Lib\site-packages\_pytest\runner.py", line 174, in pytest_runtest_call
  |     item.runtest()
  |   File "myprj\.venv\Lib\site-packages\pytest_asyncio\plugin.py", line 549, in runtest
  |     super().runtest()
  |   File "myprj\.venv\Lib\site-packages\_pytest\python.py", line 1627, in runtest
  |     self.ihook.pytest_pyfunc_call(pyfuncitem=self)
  |   File "myprj\.venv\Lib\site-packages\pluggy\_hooks.py", line 513, in __call__
  |     return self._hookexec(self.name, self._hookimpls.copy(), kwargs, firstresult)
  |            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  |   File "myprj\.venv\Lib\site-packages\pluggy\_manager.py", line 120, in _hookexec
  |     return self._inner_hookexec(hook_name, methods, kwargs, firstresult)
  |            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  |   File "myprj\.venv\Lib\site-packages\pluggy\_callers.py", line 182, in _multicall
  |     return outcome.get_result()
  |            ^^^^^^^^^^^^^^^^^^^^
  |   File "myprj\.venv\Lib\site-packages\pluggy\_result.py", line 100, in get_result
  |     raise exc.with_traceback(exc.__traceback__)
  |   File "myprj\.venv\Lib\site-packages\pluggy\_callers.py", line 103, in _multicall
  |     res = hook_impl.function(*args)
  |           ^^^^^^^^^^^^^^^^^^^^^^^^^
  |   File "myprj\.venv\Lib\site-packages\_pytest\python.py", line 159, in pytest_pyfunc_call
  |     result = testfunction(**testargs)
  |              ^^^^^^^^^^^^^^^^^^^^^^^^
  |   File "myprj\.venv\Lib\site-packages\pytest_asyncio\plugin.py", line 1069, in inner
  |     _loop.run_until_complete(task)
  |   File "mylocal\Python\Python311\Lib\asyncio\base_events.py", line 653, in run_until_complete
  |     return future.result()
  |            ^^^^^^^^^^^^^^^
  |   File "myprj\tests\test_mcpserver.py", line 23, in test_greet
  |     async with create_client() as client:
  |   File "mylocal\Python\Python311\Lib\contextlib.py", line 222, in __aexit__
  |     await self.gen.athrow(typ, value, traceback)
  |   File "myprj\tests\test_mcpserver.py", line 16, in create_client
  |     async with create_client_session(mcpserver.mcp._mcp_server) as client:
  |   File "mylocal\Python\Python311\Lib\contextlib.py", line 222, in __aexit__
  |     await self.gen.athrow(typ, value, traceback)
  |   File "myprj\.venv\Lib\site-packages\mcp\shared\memory.py", line 79, in create_connected_server_and_client_session
  |     async with anyio.create_task_group() as tg:
  |   File "myprj\.venv\Lib\site-packages\anyio\_backends\_asyncio.py", line 772, in __aexit__
  |     raise BaseExceptionGroup(
  | ExceptionGroup: unhandled errors in a TaskGroup (1 sub-exception)
  +-+---------------- 1 ----------------
    | Exception Group Traceback (most recent call last):
    |   File "myprj\.venv\Lib\site-packages\mcp\shared\memory.py", line 90, in create_connected_server_and_client_session
    |     async with ClientSession(
    |   File "myprj\.venv\Lib\site-packages\anyio\_backends\_asyncio.py", line 772, in __aexit__
    |     raise BaseExceptionGroup(
    | ExceptionGroup: unhandled errors in a TaskGroup (1 sub-exception)
    +-+---------------- 1 ----------------
      | Traceback (most recent call last):
      |   File "myprj\.venv\Lib\site-packages\mcp\shared\memory.py", line 100, in create_connected_server_and_client_session
      |     yield client_session
      |   File "myprj\tests\test_mcpserver.py", line 17, in create_client
      |     yield client
      |   File "myprj\tests\test_mcpserver.py", line 27, in test_greet
      |     assert cnt.text == "Hello, you!"
      | AssertionError: assert 'Error execut...ng event loop' == 'Hello, you!'
      |
      |   - Hello, you!
      |   + Error executing tool greet: asyncio.run() cannot be called from a running event loop
      +------------------------------------
==================================================================== short test summary info =====================================================================
FAILED tests/test_mcpserver.py::Test_MCPServer::test_greet - ExceptionGroup: unhandled errors in a TaskGroup (1 sub-exception)
================================================================== 1 failed, 1 passed in 1.63s ===================================================================

失敗した際のツールからの返り値を見ると、サーバーを起動してツールを呼び出した際に発生したエラーメッセージと同一内容が出力されていることがわかります。
この結果から、クライアントセッションを使ったテストが関数単体テストでは見逃された非同期コンテキストの問題を正確に捉えていることを明確に示しています。
これは、MCP Python SDKが提供する関数を基盤としつつ、テストコード側で必要な仕組みを自前で構築したことによる成果と言えます。

上記の方法以外にも、エラーかそうでないかを把握するために、MCPツールの呼び出し結果として返るCallToolResultインスタンスからisErrorフィールドを参照することができます。

    @pytest.mark.asyncio
    async def test_something(self):
        async with create_client() as client:
            # mcp.types.CallToolResultインスタンスが返る
            res = await client.call_tool("something", {"arg": "foo"})
            assert not res.isError  # エラーでないことを確認する

問題の解決策とテストによる確認

greetツールの実装を修正します。解決策としては、主に以下の2つのアプローチが考えられます。

  1. ツールを非同期化し、呼び出される非同期関数にawaitを使用する:
    エラーの原因はgreetツール内でasyncio.run(say_hello(name))を呼び出している点です。
    同じイベントループ上で協調して、あるコルーチンから別のコルーチン(async defで定義された関数)を呼び出す際には、その呼び出し元もasync defで定義されている必要があります(ブロッキングしないように別スレッド等で実行する方法もありますが、awaitを使用して協調的に処理を委譲するのがオーソドックスな方法です)。
    say_helloが非同期関数であるため、greetツール自身も非同期ツール (async def)とし、内部でawait say_hello(name)と呼び出すようにします。

    mcpserver.py(非同期ツール化による問題解決)
    import asyncio

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("myserver", log_level="ERROR")


async def say_hello(name: str) -> str:
    await asyncio.sleep(0.05)
    return f"Hello, {name}!"


@mcp.tool()
async def greet(name: str) -> str:
    """挨拶します。（非同期ツール）"""
    return await say_hello(name)


@mcp.tool()
async def add(a: int, b: int) -> int:
    """足し算します。（非同期ツール）"""
    await asyncio.sleep(0.05)
    return a + b


if __name__ == "__main__":
    mcp.run()

    この修正後のコードに対してクライアントセッション版テストを実行すると今度は両方のテストが成功します。非同期イベントループ上での適切なawaitによるコルーチン連携が検証された結果です。

    プロダクションコード修正後クライアントセッション版テスト結果(成功) 
    > uv run pytest tests\test_mcpserver.py -s -v
myprj\.venv\Lib\site-packages\pytest_asyncio\plugin.py:217: PytestDeprecationWarning: The configuration option "asyncio_default_fixture_loop_scope" is unset.
The event loop scope for asynchronous fixtures will default to the fixture caching scope. Future versions of pytest-asyncio will default the loop scope for asynchronous fixtures to function scope. Set the default fixture loop scope explicitly in order to avoid unexpected behavior in the future. Valid fixture loop scopes are: "function", "class", "module", "package", "session"

warnings.warn(PytestDeprecationWarning(_DEFAULT_FIXTURE_LOOP_SCOPE_UNSET))
====================================================================== test session starts =======================================================================
platform win32 -- Python 3.11.2, pytest-8.3.5, pluggy-1.5.0 -- myprj\.venv\Scripts\python.exe
cachedir: .pytest_cache
rootdir: myprj
configfile: pyproject.toml
plugins: anyio-4.9.0, asyncio-0.26.0
asyncio: mode=Mode.STRICT, asyncio_default_fixture_loop_scope=None, asyncio_default_test_loop_scope=function
collected 2 items

tests/test_mcpserver.py::Test_MCPServer::test_greet PASSED
tests/test_mcpserver.py::Test_MCPServer::test_add PASSED

======================================================================= 2 passed in 1.38s ========================================================================

    また、MCP Inspectorでツールを実行してもエラーにはなりません。
    image.png

  2. 同期ツールから呼び出される非同期関数を同期関数にする:
    もしsay_helloが実際には非同期IOを含まないか同期的な処理で要件を十分に満たすのであれば、say_hello自体を同期関数にして、greetツールも同期のままにします。

    mcpserver.py(asyncio.runを利用せず同期ツールにして問題解決)
    import asyncio

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("myserver", log_level="ERROR")


def say_hello(name: str) -> str:
    return f"Hello, {name}!"


@mcp.tool()
def greet(name: str) -> str:
    """挨拶します。（同期ツール）"""
    return say_hello(name)


@mcp.tool()
async def add(a: int, b: int) -> int:
    """足し算します。（非同期ツール）"""
    await asyncio.sleep(0.05)
    return a + b


if __name__ == "__main__":
    mcp.run()

    この場合も、greetツールが非同期イベントループのコンテキスト内で不適切なasyncio.run()を呼び出すことがなくなるため、クライアントセッション版テストは成功します。

    もちろん、MCP Inspectorでツールを実行してもエラーにはなりません。

    image.png

どちらの解決策を選ぶかは、say_helloが本来的に非同期処理である必要があるかどうかに依存します。重要なのは、既にイベントループが動いているMCPサーバーのツール関数内で、安易にasyncio.run()を使って別のコルーチンを実行しようとしないことです。

クライアントセッション版テスト手法の利点

このクライアントセッション版テスト手法は、MCP Python SDKが提供する関数を基盤としつつ、テストコード側で実践的な仕組みを構築します。
これによって単なる関数単体をテストするだけでは見過ごしてしまうような、非同期イベントループのコンテキストに依存する問題や、MCP Python SDKの内部的な処理との連携に起因する問題を効果的に検出できます。
また、プロダクションコードのツール実装が同期関数であろうと非同期関数であろうと、await client.call_tool(...)という非同期呼び出しでツールを呼び出せることはテストのメンテナビリティにとって重要です。

結論

AI/LLMに「手足」を与えるためのMCPサーバー開発において、その信頼性確保は極めて重要です。生成AIの確率的な性質とは別に、MCPサーバー自体の動作は予測可能で再現性があるものとして担保されるべきです。

本稿で示したように、このような自前で構築したテスト環境でのクライアントセッションを使ったテストは、MCPサーバーがクライアントからのリクエストを受けてツールを実行する際の振る舞いを忠実に再現し、非同期イベントループのコンテキストを含めた検証を可能にします。
単なる関数ロジックのテストにとどまらず、MCPサーバーがサーバー上でどのように実行されるかを含めた実践的なテストを行うことが、AI/LLMとMCPサーバー連携の信頼性を高める鍵となると考えています。

テスト手段が完全に確立されていない現状であっても、MCP Python SDKが提供する構成要素を活用し、必要なテストの仕組みを自前で構築することで、非同期コードの潜在的な問題を早期に検出し、より堅牢なMCPサーバーを開発することができます。あなたのMCPサーバー開発の品質向上に、このテスト戦略の導入を検討してみてください。

Appendix

この記事内容で紹介したテストをGitHub Actionsで再現するリポジトリを作成しました。

5
3
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
5
3

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?