はじめに：LLMと外部ツールの連携
大規模言語モデル（LLM）は、それ自体が強力な対話能力を持ちますが、外部ツールと連携することでその可能性は飛躍的に拡大します。インターネットアクセス、地図サービス、API、データベース、あるいはロボットアームといったツールを操作できるようになると、LLMは単なるチャットボットから、より高度な「エージェント」へと進化し、人間の複雑なタスクを支援できるようになります。
従来、LLMに外部ツールを利用させるためには、プロンプトに複雑な指示を記述する「関数呼び出し」の手法が用いられていましたが、これは効率的ではなく、開発者にとって負担となっていました。
2024年11月、Anthropic社（Claude LLMの開発元）は、Model Context Protocol（MCP）を発表しました。MCPは、LLMと外部ツール間の標準化されたインターフェースを提供するプロトコルであり、Type-Cドックのように、多種多様なソフトウェアやツールをLLMに接続し、利用可能にします。要するに、MCPはLLMが外部ツールを効果的に利用するためのフレームワークです。
MCPプロトコルの概要
MCPは、LLMとリソース間の通信をクライアント-サーバーの分散アーキテクチャで構成します。主なコンポーネントは以下の通りです。
- MCPホスト (Host): LLMが接続を開始するアプリケーション（例: Cursor、Claude、Desktopなど）。
- MCPクライアント (Client): ホストアプリケーション内でサーバーとの1対1の接続を維持します。1つのホストアプリケーション内で複数のMCPクライアントを動作させ、複数の異なるサーバーに同時に接続できます。
- MCPサーバー (Server): 独立して動作する軽量なプログラムで、標準化されたプロトコルを通じてクライアントにコンテキスト、ツール、プロンプトを提供します。MCPサービスの中心となります。
MCPサービスの設定モード
MCPサービスは主に2つのモードで設定されます。
- Stdioモード: 主にローカルPC上のソフトウェアやファイルに接続するために使用されます。例えば、オンラインサービスを持たないBlenderのようなソフトウェアをAIに操作させる場合に利用し、設定は比較的複雑です。
- SSEモード: オンラインで提供され、元々APIを持つサービスに接続するために使用されます（例: Googleメール、Googleカレンダー）。設定は非常に簡単で、基本的にURLを指定するだけです。
SSE (Server-Sent Events) モードは、LLMのトークン生成などのストリーミング応答をサポートするために設計されており、以下の特徴を持ちます。
- サーバープッシュ: サーバーは生成されたトークンを継続的に送信し、クライアントはリアルタイムでそれを受信・表示します。
- Chat Completionsインターフェース互換: 通常、OpenAIの
stream=Trueインターフェースと互換性があります。 - 高性能: 全ての生成が完了してから一度に返すよりも、ストリーミング応答の方がユーザー体験と応答速度が向上します。
SSEモードでは、その「ストリーミング」通信の性質上、非同期関数（async/await）の使用が推奨されます。これは、データを継続的に待機する必要があり、同期関数ではスレッドをブロックしてしまい、他の処理やUIを停止させてしまうためです。非同期I/Oは、データが利用可能になるまでタスクを中断し、他のコルーチンに実行権を解放することで、リソースを効率的に利用し、特にチャットボットやWebサービスにおいて大幅な性能向上をもたらします。
MCPツール実践: クラウドLLMとの連携
MCPツールの基本的な動作を理解するために、fastmcpライブラリを用いたシンプルなツールと、クラウド上のLLM（OpenAI互換APIを使用）との連携デモを示します。
1. MCPツールサーバーの実装
まず、都市の天気を問い合わせるツールを定義するMCPサーバーを作成します。
# mcp_weather_tool.py
from fastmcp import FastMCP
# "weather_app"という名前でFastMCPアプリケーションのインスタンスを作成
# これが全てのツールの一元的なエントリポイントとなる
app_server = FastMCP("weather_app")
# "get_city_weather"という名前でツールを定義し、都市の天気を問い合わせる
# このツールは文字列型の都市名をパラメータとして受け取る
@app_server.tool(name="get_city_weather", description="指定された都市の現在の天気を取得します")
def get_city_weather(city_name: str):
# 特定の都市の天気情報を含む辞書を定義
# 実際のアプリケーションでは、ここに実際の天気API呼び出しを実装する
local_weather_data = {
"東京": {"temperature": 20, "condition": "晴れ", "humidity": 60},
"大阪": {"temperature": 22, "condition": "曇り", "humidity": 75},
"京都": {"temperature": 18, "condition": "雨", "humidity": 80}
}
# 対応する都市の天気情報を返す。都市が見つからない場合はエラーメッセージを返す
return local_weather_data.get(city_name, {"error": "指定された都市の天気情報は見つかりませんでした"})
if __name__ == "__main__":
# アプリケーションをStdioトランスポートで起動
# これにより、コマンドラインを介してツールと対話できるようになる
app_server.run(transport="stdio")
run(transport="stdio") は、クライアントからの標準入出力を介したツール呼び出しコマンドを子プロセスとして待ち受けます。
2. LLMクライアントとMCPツールの連携デモ
次に、LLMが上記で作成したローカルMCPツールを呼び出すクライアント側のコードを示します。ここでは、阿里云のDashScopeサービス（Qwenモデル）をOpenAI互換API経由で利用します。
# llm_mcp_client.py
import asyncio
import json
from openai import OpenAI
from mcp.client.stdio import stdio_client
from mcp import ClientSession, StdioServerParameters
# OpenAI API互換サービス（例: 阿里云DashScope）の設定
# YOUR_API_KEYは実際のAPIキーに置き換えてください
DASH_SCOPE_API_KEY = "sk-xxxxxxxxxxxxxxxxxxxxxxxxxx"
DASH_SCOPE_BASE_URL = "https://dashscope.aliyuncs.com/compatible-mode/v1"
class LLMToolIntegrator:
def __init__(self, mcp_script_path: str):
"""
LLMとMCPツールを統合するクライアントを初期化します。
:param mcp_script_path: MCPツールサーバーのPythonスクリプトパス
"""
self.mcp_script_path = mcp_script_path
# OpenAIクライアントを初期化し、互換API経由でDashScopeサービスに接続
self.llm_api_client = OpenAI(api_key=DASH_SCOPE_API_KEY, base_url=DASH_SCOPE_BASE_URL)
async def execute_query(self, user_prompt: str):
"""
ユーザーのプロンプトを実行し、ツール使用の有無による結果を比較します。
:param user_prompt: ユーザーからの質問
:return: 比較結果を含む辞書
"""
# 標準I/O通信用のMCPサーバーパラメータを設定
server_config = StdioServerParameters(command="python", args=[self.mcp_script_path])
async with stdio_client(server=server_config) as (input_stream, output_stream):
# クライアントセッションの確立
async with ClientSession(input_stream, output_stream) as session:
await session.initialize()
# サーバーに登録されている全てのツール情報を取得
registered_tools = (await session.list_tools()).tools
# MCPツール形式をOpenAIの関数呼び出し形式に変換
openai_functions = []
for tool in registered_tools:
openai_functions.append({
"name": tool.name,
"description": tool.description or "",
"parameters": tool.inputSchema or {
"type": "object",
"properties": {
"param1": {"type": "string", "description": "任意のパラメータ"}
},
"required": []
}
})
# --- ツール使用時のLLM呼び出し ---
llm_response_with_tool = self.llm_api_client.chat.completions.create(
model="qwen-max",
messages=[{"role": "user", "content": user_prompt}],
functions=openai_functions,
function_call="auto"
)
message_with_tool = llm_response_with_tool.choices[0].message
tool_usage_details = {
"model_reply": message_with_tool.content,
"invoked_tool": None,
"tool_output": None
}
# モデルがツール呼び出しを決定した場合
if message_with_tool.function_call:
tool_name_called = message_with_tool.function_call.name
tool_args = json.loads(message_with_tool.function_call.arguments)
# MCPセッションを介して実際のツールを呼び出す
tool_execution_result = await session.call_tool(tool_name_called, tool_args)
tool_usage_details.update({
"invoked_tool": tool_name_called,
"tool_arguments": tool_args,
"tool_output": tool_execution_result
})
# --- ツール不使用時のLLM呼び出し ---
llm_response_no_tool = self.llm_api_client.chat.completions.create(
model="qwen-max",
messages=[{"role": "user", "content": user_prompt}],
# functionsパラメータを渡さないため、モデルはツールを使用しない
)
message_no_tool = llm_response_no_tool.choices[0].message
no_tool_reply = {
"model_reply": message_no_tool.content
}
return {
"user_query": user_prompt,
"with_mcp_tool": tool_usage_details,
"without_tool": no_tool_reply
}
async def main_demonstration():
"""ツールの使用有無を比較するメインデモ関数"""
# MCPクライアントを初期化し、ツールサーバーのスクリプトパスを指定
client_integrator = LLMToolIntegrator(mcp_script_path="./mcp_weather_tool.py")
# 天気問い合わせの例を実行
comparison_results = await client_integrator.execute_query("東京の天気は？")
# 結果を整形して出力
print(">>> ユーザーの質問:", comparison_results["user_query"])
print("\n--- 【MCPツールを使用した場合】 ---")
print("モデルの返答:", comparison_results["with_mcp_tool"]["model_reply"])
if comparison_results["with_mcp_tool"]["invoked_tool"]:
print("呼び出されたツール:", comparison_results["with_mcp_tool"]["invoked_tool"])
print("ツールの引数:", comparison_results["with_mcp_tool"]["tool_arguments"])
print("ツールの結果:", comparison_results["with_mcp_tool"]["tool_output"])
else:
print("ツールは呼び出されませんでした。")
print("\n--- 【ツールを使用しない場合】 ---")
print("モデルの返答:", comparison_results["without_tool"]["model_reply"])
if __name__ == "__main__":
asyncio.run(main_demonstration())
このデモでは、LLMがユーザーの質問を解析し、「東京の天気は？」という質問に対してMCPのget_city_weatherツールを正確に識別し、呼び出し、その結果（例: {"temperature": 20, "condition": "晴れ", "humidity": 60}）を取得できることを示します。これにより、LLMの機能が大幅に拡張されることがわかります。
ローカル環境でのMCP統合実践
実際のアプリケーションでは、様々なツールを開発し、それらをローカルサーバー上で提供したい場合があります。Stdioモードは開発には便利ですが、本番環境ではHTTPなどの永続的なサービスが望ましいです。ここでは、vLLMでローカルにデプロイされたQwenモデルと、ローカルで稼働するMCP HTTPサーバーを連携させる方法を解説します。
1. ローカルMCP HTTPサーバーのデプロイ
複数のツール（天気と株価）を提供するMCPサーバーをローカルの4200番ポートで起動します。
# local_mcp_service.py
from fastmcp import FastMCP
# "local_tool_service"という名前でFastMCPアプリケーションのインスタンスを作成
app_service = FastMCP("local_tool_service")
# "fetch_weather"という名前で天気問い合わせツールを登録
@app_service.tool(name="fetch_weather", description="指定された都市の現在の気象情報を取得します")
def fetch_weather(location: str):
# プリセットの天気データ（本番では外部APIと連携）
available_weather = {
"福岡": {"temp": 26, "cond": "晴れ", "wind": "3m/s"},
"札幌": {"temp": 19, "cond": "曇り", "wind": "5m/s"},
"那覇": {"temp": 30, "cond": "快晴", "wind": "2m/s"}
}
return available_weather.get(location, {"error": "該当都市の気象データが見つかりません"})
# "get_stock_quote"という名前で株価問い合わせツールを登録
@app_service.tool(name="get_stock_quote", description="指定された証券コードの株価情報を取得します")
def get_stock_quote(ticker_code: str):
# プリセットの株価データ（本番では外部APIと連携）
available_stocks = {
"7203": {"company": "トヨタ自動車", "price": 2800.5, "currency": "JPY"},
"9984": {"company": "ソフトバンクグループ", "price": 9500.0, "currency": "JPY"}
}
return available_stocks.get(ticker_code, {"error": "該当証券コードの株価情報が見つかりません"})
if __name__ == "__main__":
# ストリーミング対応HTTPプロトコルでサービスを起動
app_service.run(
transport="streamable-http", # ストリーミングHTTPプロトコルを使用
host="127.0.0.1", # ローカルアドレスをリッスン
port=4200, # サービスポート
path="/api/mcp", # サービスパスのプレフィックス
log_level="info", # ログレベルを情報に設定
)
2. MCPサービス動作確認
MCPサービスが正しく起動し、ツールが利用可能かを確認するためのテストコードです。
# test_local_mcp.py
import asyncio
import httpx
from fastmcp import Client
from fastmcp.client.transports import StreamableHttpTransport
async def check_mcp_functionality():
"""FastMCPサービスの非同期テスト関数"""
# サービスURLを定義（サーバー設定と一致させる）
MCP_SERVICE_URL = "http://127.0.0.1:4200/api/mcp"
print("="*50)
print("FastMCPローカルサービス テストスクリプト")
print("="*50)
try:
# HTTPベースのストリームトランスポートクライアントを作成
http_transport = StreamableHttpTransport(url=MCP_SERVICE_URL)
# コンテキストマネージャを使用してクライアントセッションを作成
async with Client(http_transport) as mcp_client:
print(f"MCPサービスへの接続成功: {MCP_SERVICE_URL}")
# サービス連通性をテストするためのpingリクエストを送信
await mcp_client.ping()
print("サービスヘルスチェック成功")
# サーバーに登録されている全てのツールを取得
available_tools = await mcp_client.list_tools()
tool_names_list = [t.name for t in available_tools]
print(f"利用可能なツール: {', '.join(tool_names_list)}")
# ==== ツール呼び出しの例 ====
# 1. 天気ツールを呼び出して福岡の天気を問い合わせ
weather_query_results = await mcp_client.call_tool("fetch_weather", {"location": "福岡"})
weather_data = weather_query_results[0].text # 最初の結果の辞書データを抽出
print(f"福岡の天気: 気温={weather_data['temp']}℃, 状態={weather_data['cond']}")
# 2. 株価ツールを呼び出してトヨタ自動車の株価を問い合わせ
stock_query_results = await mcp_client.call_tool("get_stock_quote", {"ticker_code": "7203"})
stock_data = stock_query_results[0].text
print(f"株価情報: 会社名={stock_data['company']}, 価格={stock_data['price']} {stock_data['currency']}")
# 3. エラー処理のテスト（存在しない都市を問い合わせ）
try:
error_response = await mcp_client.call_tool("fetch_weather", {"location": "ニューヨーク"})
# エラーメッセージが期待通りか確認
if error_response and hasattr(error_response[0], 'error'):
print(f"エラー処理テスト: {error_response[0].error} - 期待される動作")
except Exception as e:
print(f"予期せぬエラー発生: {str(e)}")
except httpx.ConnectError:
print(f"接続失敗！サービスが {MCP_SERVICE_URL} で実行されているか確認してください。")
except Exception as e:
print(f"テスト中にエラーが発生しました: {str(e)}")
if __name__ == "__main__":
asyncio.run(check_mcp_functionality())
このスクリプトを実行し、エラーがなければ、MCPサービスが正しく機能していることを確認できます。
3. vLLMを用いたローカルモデルのデプロイ
vLLMを使用してQwen系列のモデルをローカルでデプロイし、OpenAI互換APIエンドポイントとして8000番ポートで公開します。
python -m vllm.entrypoints.openai.api_server \
--model ./qwen3-1.7b/ \
--served-model-name "qwen3-1.7b-local" \
--port 8000 \
--trust-remote-code \
--enable-auto-tool-choice \
--tool-call-parser hermes
上記のコマンドを実行することで、ローカルPCの8000番ポートでLLMサービスが利用可能になります。
4. ローカルLLMとローカルMCPの連携
最後に、ローカルでデプロイしたLLMとローカルMCPサービスを連携させ、ユーザーの質問に応じてツールを呼び出すデモコードを示します。
# local_llm_mcp_integration.py
import asyncio
import json
from openai import AsyncOpenAI
from fastmcp import Client
from fastmcp.client.transports import StreamableHttpTransport
# MCPサービスのエンドポイントURL
LOCAL_MCP_SERVER_URL = "http://127.0.0.1:4200/api/mcp"
# ローカルvLLMサービスのエンドポイントURL
LOCAL_LLM_SERVER_URL = "http://localhost:8000/v1"
async def invoke_mcp_tool(tool_name: str, params: dict):
"""
指定されたMCPツールを呼び出す共通関数
:param tool_name: 呼び出すツールの名前
:param params: ツールに渡すパラメータ辞書
:return: ツール実行結果
"""
mcp_transport = StreamableHttpTransport(url=LOCAL_MCP_SERVER_URL)
async with Client(mcp_transport) as mcp_client:
tool_results = await mcp_client.call_tool(tool_name, params)
# 結果はリストで返される場合があるので、最初の要素のテキストを返す
return tool_results[0].text if tool_results else {}
async def conversation_with_local_tools():
"""
ローカルツール呼び出しに対応したチャット機能の実装
1. ローカルvLLMサービスに接続
2. MCPサービスから利用可能なツールリストを取得し、OpenAI関数呼び出し形式に変換
3. ユーザーの質問に応じて適切なツールを呼び出し
4. ツール結果を統合して最終的な応答を生成
"""
# ローカルvLLMサービス（OpenAI API互換）に接続
llm_async_client = AsyncOpenAI(
base_url=LOCAL_LLM_SERVER_URL,
api_key="EMPTY" # ローカルサービスではAPIキーは不要
)
# 動的にMCPサービスからツールリストを取得
mcp_transport_for_llm = StreamableHttpTransport(url=LOCAL_MCP_SERVER_URL)
async with Client(mcp_transport_for_llm) as mcp_session_for_llm:
available_mcp_tools = await mcp_session_for_llm.list_tools()
# MCPツールスキーマをOpenAI関数呼び出し形式に変換
openai_tool_definitions = []
for tool in available_mcp_tools:
# MCPのinputSchemaがOpenAPIの形式に合致していることを前提
openai_tool_definitions.append({
"type": "function",
"function": {
"name": tool.name,
"description": tool.description,
"parameters": tool.inputSchema
}
})
# ユーザーからの質問例
user_request = "福岡の天気とトヨタ自動車の株価を教えてください"
# LLMを最初に呼び出し、ツール呼び出しが必要か判断させる
initial_response = await llm_async_client.chat.completions.create(
model="qwen3-1.7b-local",
messages=[{"role": "user", "content": user_request}],
tools=openai_tool_definitions,
tool_choice="auto" # モデルに自動でツールを選択させる
)
llm_message = initial_response.choices[0].message
print("--- モデルからの初期応答 (ツール呼び出し計画) ---")
print(llm_message.tool_calls)
if llm_message.tool_calls:
print("\n--- ツール呼び出しを検出 ---")
tool_call_results = []
# モデルが要求した全てのツールを順次実行
for call_req in llm_message.tool_calls:
print(f"ツール '{call_req.function.name}' を実行中...")
# 引数文字列を辞書に変換してツールを呼び出す
parsed_arguments = json.loads(call_req.function.arguments)
tool_output = await invoke_mcp_tool(
call_req.function.name,
parsed_arguments
)
print(f"ツール '{call_req.function.name}' の結果: {tool_output}")
tool_call_results.append({
"role": "tool",
"name": call_req.function.name,
"content": json.dumps(tool_output) # ツール結果をJSON文字列として保存
})
# ツール結果をLLMにフィードバックし、最終応答を生成させる
final_conversation_history = [
{"role": "user", "content": user_request}, # 元の質問
llm_message, # モデルのツール呼び出し計画
*tool_call_results # 各ツールの実行結果
]
final_llm_response = await llm_async_client.chat.completions.create(
model="qwen3-1.7b-local",
messages=final_conversation_history,
)
print("\n--- 最終的なモデルの応答 ---")
print(final_llm_response.choices[0].message.content)
else:
# モデルがツール不要と判断した場合、直接モデルの応答を返す
print("\n--- ツール呼び出しなし ---")
print("直接応答:", llm_message.content)
if __name__ == "__main__":
asyncio.run(conversation_with_local_tools())
この最終的なコードを実行すると、LLMがユーザーの質問「福岡の天気とトヨタ自動車の株価を教えてください」を解析し、ローカルのMCPサーバーに登録されたfetch_weatherとget_stock_quoteツールを適切に呼び出し、その結果に基づいて最終的な回答を生成する一連のプロセスを確認できます。
例として、以下のようなモデルの応答が得られます。
--- モデルからの初期応答 (ツール呼び出し計画) ---
[
ChatCompletionToolMessageParam(function=Function(arguments='{"location": "福岡"}', name='fetch_weather'), id='call_xxxxxxxxx', type='function'),
ChatCompletionToolMessageParam(function=Function(arguments='{"ticker_code": "7203"}', name='get_stock_quote'), id='call_yyyyyyyyy', type='function')
]
--- ツール呼び出しを検出 ---
ツール 'fetch_weather' を実行中...
ツール 'fetch_weather' の結果: {'temp': 26, 'cond': '晴れ', 'wind': '3m/s'}
ツール 'get_stock_quote' を実行中...
ツール 'get_stock_quote' の結果: {'company': 'トヨタ自動車', 'price': 2800.5, 'currency': 'JPY'}
--- 最終的なモデルの応答 ---
福岡の天気は晴れで、気温は26℃、風速は3m/sです。トヨタ自動車の現在の株価は2800.5円です。
このように、ローカルでデプロイされたLLMとMCPツールがシームレスに連携し、複雑なクエリにも対応できるようになります。