Apidog

オールインワン協働API開発プラットフォーム

API設計

APIドキュメント

APIデバッグ

APIモック

API自動テスト

無料会員登録
ダウンロードMacとLinux版はこちら
Home

/

MCPサーバーでLangchainを使用してみた、手順は以下の通り:

MCPサーバーでLangchainを使用してみた、手順は以下の通り:

中村 拓也

中村 拓也

Updated on 4月 10, 2025

モデルコンテキストプロトコル（MCP）は、Anthropicによって開発されたオープンソースのプロトコルであり、大規模言語モデル（LLM）アプリケーションにおける基本的な課題である、外部データソースやツールからの孤立を解決します。この包括的なチュートリアルでは、LangChainを使用してMCPを実装する方法を案内し、両方の技術を効果的に活用した洗練されたアプリケーションを作成するための知識を提供します。

💡
APIベースのアプリケーションのテストを実装する際、開発者やテスターは、API開発ライフサイクルを合理化する包括的なPostmanの代替ツールである Apidog のような専門的なツールにますます頼るようになっています。 
ボタン

MCPの理解とその目的

モデルコンテキストプロトコルは、LLMベースのアプリケーションが多様な外部システムに接続する方法を標準化することを目的としています。MCPは「AIのためのUSB-C」と考えてください - LLMs/AIエージェントと外部リソースとの間でシームレスで安全かつスケーラブルなデータ交換を可能にするユニバーサルインターフェースです。

MCPはクライアントサーバーアーキテクチャを採用しています：

  • MCPホスト: 外部データにアクセスする必要があるAIアプリケーション
  • MCPサーバー: ホストに情報を提供するデータまたはツールの提供者

このプロトコルは、関心の明確な分離を促進し、開発者が堅牢なセキュリティを維持しながらモジュラーで再利用可能なコネクタを作成できるようにします。

技術アーキテクチャ

MCPのアーキテクチャは、3つの主要なコンポーネントで構成されています：

  1. サーバー: MCPサーバーは、標準化されたAPIを介してツールとデータソースを公開します。
  2. クライアント: クライアントアプリケーションはサーバーと通信してツールとデータにアクセスします。
  3. アダプター: LangChainは、MCPサーバーとLLMアプリケーション間の統合を簡素化するアダプターを提供します。

通信フローはこのパターンに従います：

  • LangChainアプリケーションがデータ/ツールの実行をリクエストします。
  • MCPアダプターがリクエストをMCPプロトコルフォーマットに変換します。
  • サーバーがリクエストを処理し、結果を返します。
  • アダプターが応答をLangChainが使用できるフォーマットに戻します。

前提条件

始める前に、以下のものを確認してください：

  • Python 3.8以上がインストールされていること
  • OpenAI APIキー（LangChainでGPTモデルを使用するため）
  • LangChainの概念に対する基本的な理解
  • ターミナルアクセス（macOSでの例を示します）

環境設定

まず、開発環境を作成して構成しましょう：

# 仮想環境を作成する
python3 -m venv MCP_Demo

# 仮想環境を有効化する
source MCP_Demo/bin/activate

# 必要なパッケージをインストールする
pip install langchain-mcp-adapters
pip install langchain-openai

# OpenAI APIキーを設定する
export OPENAI_API_KEY=your_api_key

シンプルなMCPサーバーの作成

最初に、数学演算を提供する基本的なMCPサーバーを作成します。math_server.pyという名前のファイルを作成してください：

# math_server.py
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("Math")

@mcp.tool()
def add(a: int, b: int) -> int:
    """2つの数を加算する"""
    return a + b

@mcp.tool()
def multiply(a: int, b: int) -> int:
    """2つの数を掛け算する"""
    return a * b

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

このサーバーは、addとmultiplyという2つの数学ツールを公開します。FastMCPクラスはサーバーの作成を簡素化し、プロトコルの詳細を自動的に処理します。@mcp.tool()で装飾された各関数はクライアントに利用可能となり、ドキュメンテーションはドックストリングから導出されます。

LangChainクライアントの実装

次に、MCPサーバーと対話するためのLangChainクライアントを作成します。これをclient.pyとして保存してください：

# stdio接続用のサーバーパラメータを作成する
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from langchain_mcp_adapters.tools import load_mcp_tools
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
import asyncio

model = ChatOpenAI(model="gpt-4o")

# サーバーパラメータを設定する
server_params = StdioServerParameters(
    command="python",
    # サーバーファイルへのパスを指定する
    args=["math_server.py"],
)

async def run_agent():
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # 接続を初期化する
            await session.initialize()
            
            # LangChain形式にMCPツールを読み込む
            tools = await load_mcp_tools(session)
            
            # エージェントを作成して実行する
            agent = create_react_agent(model, tools)
            agent_response = await agent.ainvoke({"messages": "What's (3 + 5) x 12?"})
            return agent_response

# 非同期関数を実行する
if __name__ == "__main__":
    result = asyncio.run(run_agent())
    print(result)

このクライアントはMCPサーバーへの接続を確立し、利用可能なツールを読み込んで、これらのツールを使用して問題を解決できるLangChainエージェントを作成します。

例の実行

この例を実行するには：

  1. 1つのターミナルタブでMCPサーバーを起動します：
python3 math_server.py
  1. 別のターミナルタブでクライアントを実行します：
python3 client.py

クライアントはLangChainエージェントを呼び出し、次のことを行います：

  1. 質問「(3 + 5) x 12」を解析します。
  2. 引数3と5でaddツールを呼び出します。
  3. 結果8を取得します。
  4. 引数8と12でmultiplyツールを呼び出します。
  5. 最終的な答え：96を返します。

高度なMCPサーバー実装

より洗練されたMCPサーバーを作成するために実装を拡張しましょう。この例は、外部システムへのコネクタを構築する方法を示しています：

# db_server.py
from mcp.server.fastmcp import FastMCP
import sqlite3
from typing import List, Dict, Any

class DatabaseConnector:
    def __init__(self, db_path):
        self.conn = sqlite3.connect(db_path)
        self.cursor = self.conn.cursor()
    
    def execute_query(self, query: str) -> List[Dict[str, Any]]:
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        results = []
        for row in self.cursor.fetchall():
            results.append({columns[i]: row[i] for i in range(len(columns))})
        return results

mcp = FastMCP("DatabaseTools")
db_connector = DatabaseConnector("example.db")

@mcp.tool()
def run_sql_query(query: str) -> List[Dict[str, Any]]:
    """データベースでSQLクエリを実行し、結果を返します"""
    try:
        return db_connector.execute_query(query)
    except Exception as e:
        return {"error": str(e)}

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

LangChainによる複数のMCPサーバーの統合

より複雑なアプリケーションでは、複数のMCPサーバーを統合する必要があるかもしれません。複数のサーバーに接続するクライアントを作成する方法は次の通りです：

# multi_server_client.py
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from langchain_mcp_adapters.tools import load_mcp_tools
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
import asyncio
from typing import List, Dict

# サーバーの構成を定義します
servers = [
    {
        "name": "math",
        "params": StdioServerParameters(
            command="python", 
            args=["math_server.py"]
        )
    },
    {
        "name": "database",
        "params": StdioServerParameters(
            command="python", 
            args=["db_server.py"]
        )
    }
]

async def connect_to_server(server_config):
    """単一のMCPサーバーに接続してそのツールを読み込みます"""
    name = server_config["name"]
    params = server_config["params"]
    
    read, write = await stdio_client(params).__aenter__()
    session = ClientSession(read, write)
    await session.__aenter__()
    await session.initialize()
    
    tools = await load_mcp_tools(session)
    return {
        "name": name,
        "session": session,
        "tools": tools,
        "cleanup": lambda: asyncio.gather(
            session.__aexit__(None, None, None),
            stdio_client(params).__aexit__(None, None, None)
        )
    }

async def run_multi_server_agent():
    # すべてのサーバーに接続する
    connections = await asyncio.gather(
        *[connect_to_server(server) for server in servers]
    )
    
    try:
        # すべてのサーバーからツールを収集する
        all_tools = []
        for connection in connections:
            all_tools.extend(connection["tools"])
        
        # すべてのツールを持つエージェントを作成する
        model = ChatOpenAI(model="gpt-4o")
        agent = create_react_agent(model, all_tools)
        
        # 複数のサーバーを使用する可能性のある複雑なクエリでエージェントを実行する
        response = await agent.ainvoke({
            "messages": "平均注文価額を超えた顧客を見つけ、その合計支出を計算します。"
        })
        
        return response
    
    finally:
        # すべての接続をクリーンアップする
        for connection in connections:
            await connection["cleanup"]()

# 複数サーバーエージェントを実行する
if __name__ == "__main__":
    result = asyncio.run(run_multi_server_agent())
    print(result)

エラーハンドリングとフォールバック戦略

堅牢なMCPの実装には、エラーハンドリングが含まれるべきです。以下は、エラーハンドリングとフォールバック戦略を示すクライアントの強化版の例です：

# robust_client.py
async def run_agent_with_fallbacks():
    try:
        # プライマリ接続を試みる
        async with stdio_client(server_params) as (read, write):
            async with ClientSession(read, write) as session:
                try:
                    await session.initialize()
                    tools = await load_mcp_tools(session)
                    agent = create_react_agent(model, tools)
                    return await agent.ainvoke({"messages": "what's (3 + 5) x 12?"})
                except Exception as e:
                    print(f"MCPツールを使用中にエラーが発生しました: {e}")
                    # ツールなしで直接モデル呼び出しにフォールバックする
                    return await model.ainvoke([
                        HumanMessage(content="what's (3 + 5) x 12?")
                    ])
    except Exception as connection_error:
        print(f"接続エラー: {connection_error}")
        # 最終フォールバック
        return {"error": "MCPサーバーへの接続を確立できませんでした"}

セキュリティ上の考慮事項

LangChainでMCPを実装する際は、以下のセキュリティベストプラクティスを考慮してください：

  1. 入力検証: 注入攻撃を防ぐために、MCPツールへの入力を常に検証する。
  2. ツールの権限: 各ツールに対して微 granularな権限を実装する。
  3. レート制限: ツールの乱用を防ぐためにレート制限を適用する。
  4. 認証: クライアントとサーバー間の適切な認証を実装する。

以下は、ツールの権限を実装する例です：

from mcp.server.fastmcp import FastMCP, Permission

mcp = FastMCP("SecureTools")

# 権限レベルを定義する
READ_PERMISSION = Permission("read", "データを読み取ることができる")
WRITE_PERMISSION = Permission("write", "データを変更することができる")

@mcp.tool(permissions=[READ_PERMISSION])
def get_data(key: str) -> str:
    """キーによるデータ取得（読取権限が必要）"""
    # 実装...
    return f"{key}のデータ"

@mcp.tool(permissions=[WRITE_PERMISSION])
def update_data(key: str, value: str) -> bool:
    """データを更新する（書込権限が必要）"""
    # 実装...
    return True

パフォーマンス最適化

本番デプロイメントでは、以下のパフォーマンス最適化を考慮してください：

  1. 接続プーリング: 各リクエストごとに新しい接続を作成するのではなく、MCP接続を再利用する。
  2. バッチ処理: 可能な場合は複数のツール呼び出しをグループ化する。
  3. 非同期処理: asyncioを使用して複数のリクエストを同時に処理する。

接続プーリングの例：

class MCPConnectionPool:
    def __init__(self, max_connections=10):
        self.available_connections = asyncio.Queue(max_connections)
        self.max_connections = max_connections
        self.current_connections = 0
    
    async def initialize(self):
        # いくつかの接続を事前に作成する
        for _ in range(3):  # 3つの接続から始める
            await self._create_connection()
    
    async def _create_connection(self):
        if self.current_connections >= self.max_connections:
            raise Exception("最大接続数に達しました")
        
        read, write = await stdio_client(server_params).__aenter__()
        session = await ClientSession(read, write).__aenter__()
        await session.initialize()
        
        self.current_connections += 1
        await self.available_connections.put(session)
    
    async def get_connection(self):
        if self.available_connections.empty() and self.current_connections < self.max_connections:
            await self._create_connection()
        
        return await self.available_connections.get()
    
    async def release_connection(self, connection):
        await self.available_connections.put(connection)

MCP実装のテスト

徹底的なテストは信頼性のあるMCP実装には欠かせません。以下はpytestを使用したテストアプローチの例です：

# test_mcp.py
import pytest
import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from langchain_mcp_adapters.tools import load_mcp_tools

@pytest.fixture
async def mcp_session():
    server_params = StdioServerParameters(
        command="python",
        args=["math_server.py"],
    )
    
    read, write = await stdio_client(server_params).__aenter__()
    session = ClientSession(read, write)
    await session.__aenter__()
    await session.initialize()
    
    yield session
    
    await session.__aexit__(None, None, None)
    await stdio_client(server_params).__aexit__(None, None, None)

@pytest.mark.asyncio
async def test_add_tool(mcp_session):
    tools = await load_mcp_tools(mcp_session)
    add_tool = next(tool for tool in tools if tool.name == "add")
    
    result = await add_tool.invoke({"a": 5, "b": 7})
    assert result == 12

@pytest.mark.asyncio
async def test_multiply_tool(mcp_session):
    tools = await load_mcp_tools(mcp_session)
    multiply_tool = next(tool for tool in tools if tool.name == "multiply")
    
    result = await multiply_tool.invoke({"a": 6, "b": 8})
    assert result == 48

結論

モデルコンテキストプロトコルは、LangChainアプリケーションと外部ツールおよびデータソースを接続するための強力なフレームワークを提供します。これらの接続を標準化することにより、MCPは開発者が洗練されたAIエージェントを作成し、環境とシームレスに対話できるようにします。

LangChainのエージェント機能とMCPの接続性の組み合わせは、本当に強力でコンテキストを意識したアプリケーションを構築するための基盤を作ります。MCPエコシステムが成長を続ける中で、より多くの事前構築されたサーバーやツールが登場し、開発プロセスをさらに簡素化することが期待できます。

このチュートリアルでは、基本的なセットアップから接続プーリングやエラーハンドリングのような高度なパターンまで、LangChainでMCPを使用する際の基本的な概念と実装詳細を取り上げました。これらのプラクティスに従うことで、両方の技術の最良の点を活用した堅牢で本番用のアプリケーションを作成できます。

さらに探求するために、GitHubにあるMCPサーバーの成長するエコシステムを調査するか、コミュニティに自分のサーバーを貢献することを検討してください。AIエージェントの未来は、外部ツールやデータを効果的に活用する能力にあります。MCPはそのビジョンを実現するための重要なステップを提供します。