LangChain Tool CallingとMCPプロトコル統合実践ガイド

AIアプリケーション開発において、Tool Calling（関数呼び出し）は外部システムと連携する上で不可欠な技術となりました。本稿では、LangChainにおけるTool Callingの実装と、Model Context Protocol（MCP）との統合について、HolySheep AIを活用した実践的な観点から詳細に解説します。

AI APIコスト比較：月間1000万トークン規模の分析

まず初めに、各主要LLMのAPIコストを確認しましょう。2026年最新のoutputトークン単価を基準に、月間1000万トークン利用時のコスト比較表を示します。

モデル	Output価格 ($/MTok)	10Mトークン/月 ($)	備考
Claude Sonnet 4.5	$15.00	$150.00	最高品質が必要な場合
GPT-4.1	$8.00	$80.00	汎用性に優れる
Gemini 2.5 Flash	$2.50	$25.00	コストバランス良好
DeepSeek V3.2	$0.42	$4.20	最安値・高性能

上記表中、DeepSeek V3.2はClaude Sonnet 4.5と比較して約36分の1のコストでありながら、Tool Calling能力は遜色ありません。コスト最適化を検討する場合、DeepSeek V3.2は有力な選択肢となります。

LangChain Tool Calling基礎実装

LangChainにおけるTool Callingは、モデルに外部関数の呼び出し能力を付与する仕組みです。以下に、HolySheep AIを経由した基本的な実装例を示します。

"""
LangChain Tool Calling 基礎実装
HolySheep AI API 使用例
"""
import os
from typing import Optional
from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from pydantic import BaseModel, Field

HolySheep API設定
os.environ["HOLYSHEEP_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"
os.environ["HOLYSHEEP_BASE_URL"] = "https://api.holysheep.ai/v1"

ツール定義
@tool
def get_weather(city: str) -> str:
    """指定された都市の天気を取得する"""
    weather_data = {
        "東京": "晴れ、25°C",
        "ニューヨーク": "曇り、18°C",
        "ロンドン": "雨、12°C"
    }
    return weather_data.get(city, "データなし")

@tool
def calculate(expression: str) -> float:
    """数式を計算する（安全な評価のみ）"""
    allowed_chars = set("0123456789+-*/.() ")
    if all(c in allowed_chars for c in expression):
        return eval(expression)
    raise ValueError("許可されていない文字が含まれています")

モデル初期化（DeepSeek V3.2）
model = init_chat_model(
    model="deepseek/deepseek-chat-v3.2",
    model_provider="openai",
    api_key=os.environ["HOLYSHEEP_API_KEY"],
    base_url=os.environ["HOLYSHEEP_BASE_URL"],
    temperature=0.7
)

ツールバインディング
tools = [get_weather, calculate]
model_with_tools = model.bind_tools(tools)

実行例
query = "東京の天気を教えていただき、138 * 27を計算してください"
response = model_with_tools.invoke(query)

print(f"Query: {query}")
print(f"Response: {response}")
print(f"Tool Calls: {response.tool_calls}")

MCPプロトコルとの統合

Model Context Protocol（MCP）は、LLMと外部データソース・ツールを統一的に接続するプロトコルです。LangChainではmcplibを通じてMCPサーバーへの接続がサポートされています。

"""
LangChain + MCP統合実装
外部MCPサーバーに接続してツールを呼び出す
"""
import asyncio
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, SystemMessage

HolySheep API設定
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"

MCPサーバー設定
mcp_servers = {
    "filesystem": {
        "command": "npx",
        "args": ["-y", "@modelcontextprotocol/server-filesystem", "/tmp"],
        "transport": "stdio"
    },
    "brave-search": {
        "command": "python",
        "args": ["-m", "mcp_servers.brave_search"],
        "env": {"BRAVE_API_KEY": "your-brave-api-key"},
        "transport": "stdio"
    }
}

async def main():
    # MCPクライアント初期化
    async with MultiServerMCPClient(mcp_servers) as client:
        # 利用可能なツール取得
        tools = client.get_tools()
        print(f"利用可能なMCPツール: {[t.name for t in tools]}")
        
        # DeepSeek V3.2でモデル初期化
        model = init_chat_model(
            model="deepseek/deepseek-chat-v3.2",
            model_provider="openai",
            api_key=HOLYSHEEP_API_KEY,
            base_url=HOLYSHEEP_BASE_URL
        )
        
        # システムプロンプト設定
        system_msg = SystemMessage(
            content="あなたはMCPプロトコルを通じて外部ツールにアクセスできる助手です。"
        )
        
        # ユーザークエリ
        user_query = HumanMessage(
            content="/tmp/test.txt ファイルの最終更新日時を教えてください"
        )
        
        # ツール呼び出しを含むチェーン実行
        from langchain_core.output_parsers import StrOutputParser
        from langchain import RunnablePassthrough
        
        chain = (
            {"context": RunnablePassthrough()}
            | model.bind_tools(tools)
        )
        
        result = await chain.ainvoke([system_msg, user_query])
        print(f"MCP呼び出し結果: {result}")

if __name__ == "__main__":
    asyncio.run(main())

Tool Callingの結果処理とストリーミング

実際のアプリケーションでは、Tool Callingの結果を適切に処理し、ユーザーへフィードバックを返す必要があります。以下に結果処理とストリーミング出力の実装例を示します。

"""
Tool Calling 結果処理とストリーミング実装
LangChain Expression Language (LCEL) 使用
"""
from langchain.chat_models import init_chat_model
from langchain_core.output_parsers import StrOutputParser
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, AIMessage
from langchain_core.runnables import RunnablePassthrough
import json

ツール定義
@tool
def query_database(sql: str) -> str:
    """データベースにSQLクエリを実行する（読み取り専用）"""
    # 実際の実装では接続池を使用
    return json.dumps([
        {"id": 1, "name": "製品A", "price": 2980},
        {"id": 2, "name": "製品B", "price": 4500}
    ])

@tool  
def send_notification(message: str, channel: str = "email") -> str:
    """指定チャンネルに通知を送信する"""
    channels = {"email": "mailgun", "slack": "webhook", "line": "api"}
    return f"通知送信完了: {channels.get(channel, 'unknown')} via {message[:20]}..."

HolySheep初期化
model = init_chat_model(
    model="deepseek/deepseek-chat-v3.2",
    model_provider="openai",
    api_key="YOUR_HOLYSHEEP_API_KEY",
    base_url="https://api.holysheep.ai/v1"
)

tools = [query_database, send_notification]

LCELによるチェーン構築
def create_tool_calling_chain():
    # ツールバインディング
    model_with_tools = model.bind_tools(tools, parallel_tool_calls=True)
    
    # チェーン定義
    chain = model_with_tools | StrOutputParser()
    
    return chain

ストリーミング実行
def stream_tool_calls():
    chain = create_tool_calling_chain()
    
    query = (
        "売上上位5件の製品を取得し、結果をSlackチャンネルに通知してください"
    )
    
    print(f"クエリ: {query}")
    print("-" * 50)
    
    # ストリーミング出力
    for chunk in chain.stream([HumanMessage(content=query)]):
        print(chunk, end="", flush=True)
    
    print("\n" + "-" * 50)

結果処理クラス
class ToolCallHandler:
    def __init__(self):
        self.call_history = []
    
    def process_tool_result(self, ai_message):
        """Tool Calling結果を処理"""
        if hasattr(ai_message, 'tool_calls') and ai_message.tool_calls:
            for call in ai_message.tool_calls:
                print(f"ツール呼び出し: {call['name']}")
                print(f"引数: {json.dumps(call['args'], ensure_ascii=False)}")
                self.call_history.append(call)
        
        return ai_message

if __name__ == "__main__":
    handler = ToolCallHandler()
    stream_tool_calls()

LangChain Agentによる自律的Tool Calling

より高度な用途では、Agentパターンを採用することで、モデルに自律的にツール選択・実行させることができます。以下にReAct Agentの実装例を示します。

"""
LangChain ReAct Agent 実装
自律的なTool Callingと実行
"""
from langchain.agents import create_react_agent, AgentExecutor
from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from langchain import hub

@tool
def get_stock_price(symbol: str) -> float:
    """株式の現在価格を取得（記号: AAPL, GOOGL, MSFT等）"""
    prices = {"AAPL": 178.50, "GOOGL": 142.30, "MSFT": 378.90}
    return prices.get(symbol.upper(), 0.0)

@tool
def calculate_portfolio_value(symbols: list, shares: list) -> float:
    """ポートフォリオの合計価値を計算"""
    total = 0.0
    for symbol, share_count in zip(symbols, shares):
        price = get_stock_price(symbol)
        total += price * share_count
    return total

@tool
def get_company_info(symbol: str) -> str:
    """企業の基本情報を取得"""
    info = {
        "AAPL": "Apple Inc. - テクノロジーセクター",
        "GOOGL": "Alphabet Inc. - テクノロジーセクター",
        "MSFT": "Microsoft Corp. - テクノロジーセクター"
    }
    return info.get(symbol.upper(), "企業情報なし")

HolySheep設定
model = init_chat_model(
    model="deepseek/deepseek-chat-v3.2",
    model_provider="openai",
    api_key="YOUR_HOLYSHEEP_API_KEY",
    base_url="https://api.holysheep.ai/v1",
    temperature=0.2  # 決定論的な応答
)

tools = [get_stock_price, calculate_portfolio_value, get_company_info]

ReAct Agent作成
prompt = hub.pull("hwchase17/react-chat")

agent = create_react_agent(model, tools, prompt)

Agent Executor設定
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=10,
    handle_parsing_errors=True
)

実行例
if __name__ == "__main__":
    query = """
    私のポートフォリオはAAPLを100株、GOOGLを50株持っています。
    各企業の情報を確認し、合計投資額を計算してください。
    """
    
    print("クエリ:", query)
    print("=" * 60)
    
    result = agent_executor.invoke({"input": query})
    
    print("=" * 60)
    print(f"最終回答:\n{result['output']}")

HolySheep AI活用の具体例

私は実際にHolySheep AIを使用してTool Callingアプリケーションを構築していますが、特に以下の点で有用性を実感しています。

• コスト効率：DeepSeek V3.2をHolySheep経由で利用率$0.42/MTok。月間1000万トークン利用时、公式API比で約85%のコスト削減を実現
• 低レイテンシ：<50msの応答速度で、Tool Callingの対話型体験が快適
• 柔軟な決済：WeChat Pay・Alipay対応で、日本在住の開発者でも簡単にチャージ可能
• 互換性：OpenAI互換APIのため、LangChain等他ライブラリとの統合がスムーズ

よくあるエラーと対処法

エラー1：Tool Callingが実行されない（bind_tools未設定）

# ❌ 誤り：バインディングなしで呼び出し
model = init_chat_model(
    model="deepseek/deepseek-chat-v3.2",
    model_provider="openai",
    api_key=HOLYSHEEP_API_KEY,
    base_url=HOLYSHEEP_BASE_URL
)
response = model.invoke("東京の天気を教えて")  # ツールが呼ばれない

✅ 修正：bind_toolsでツールをバインディング
model_with_tools = model.bind_tools(tools)
response = model_with_tools.invoke("東京の天気を教えて")

原因：モデルにツールの存在を認識させていないため、JSON形式でツール呼び出しを生成しない。

エラー2：MCPサーバー接続タイムアウト

# ❌ 誤り：タイムアウト設定なし
async with MultiServerMCPClient(servers) as client:
    ...

✅ 修正：タイムアウトとエラーハンドリング追加
from contextlib import asynccontextmanager

@asynccontextmanager
async def safe_mcp_client(servers, timeout=30.0):
    try:
        async with asyncio.timeout(timeout):
            async with MultiServerMCPClient(servers) as client:
                yield client
    except asyncio.TimeoutError:
        print(f"MCPサーバー接続が{timeout}秒以内に完了しませんでした")
        raise
    except Exception as e:
        print(f"MCP接続エラー: {e}")
        raise

使用例
async with safe_mcp_client(mcp_servers, timeout=30.0) as client:
    tools = client.get_tools()

原因：MCPサーバーが起動していない、またはネットワーク問題で接続できない。

エラー3：Tool引数の型不一致

from pydantic import BaseModel, Field
from typing import Optional

❌ 誤り：型ヒントなしで定義
@tool
def search_products(keyword, limit=10):
    ...

✅ 修正：Pydanticモデルで引数を定義
class SearchInput(BaseModel):
    keyword: str = Field(description="検索キーワード")
    limit: Optional[int] = Field(default=10, description="取得件数上限")
    category: Optional[str] = Field(default=None, description="商品カテゴリ")

@tool(args_schema=SearchInput)
def search_products(keyword: str, limit: int = 10, category: str = None):
    """商品を検索する"""
    # 実装
    return results

原因：LLMが生成するJSON引数の型がPython関数と不一致。

エラー4：レート制限（Rate Limit）エラー

import time
import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential

❌ 誤り：リトライなしでの連続呼び出し
result = model.invoke(query)
result2 = model.invoke(query2)

✅ 修正：エクスポネンシャルバックオフでリトライ
@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
async def call_model_with_retry(messages):
    try:
        return await model.ainvoke(messages)
    except Exception as e:
        if "rate_limit" in str(e).lower():
            print("レート制限を検出、リトライ中...")
            raise
        return None

使用
result = await call_model_with_retry(query)

原因：API呼び出し頻度がHolySheepの制限を超過。

パフォーマンス最適化Tips

• parallel_tool_calls=True：独立したツール呼び出しを並列実行可能
• Streaming対応：real-timeフィードバックでUX向上
• キャッシュ活用：同じクエリへの応答をキャッシュしてコスト削減
• モデル選択：Tool Calling用途にはDeepSeek V3.2で十分（Claude比95%コスト削減）

まとめ

本稿では、LangChainにおけるTool Callingの基礎実装からMCPプロトコル統合、Agentパターンの活用まで解説しました。コスト面での優位性（DeepSeek V3.2利用時、月間1000万トークンで$4.20）とHolySheep AIの柔軟な決済手段（WeChat Pay/Alipay対応）を組み合わせることで、商用レベルのAIアプリケーションを経済的に構築できます。

是非[今すぐ登録]して、あなた만의Tool Callingアプリケーション開発を始めてみてください。登録者には無料クレジットが付与されます。

👉 HolySheep AI に登録して無料クレジットを獲得