背景
近年、AIエージェント技術の進化に伴い、企業のデジタル変革を支援するツールとしての期待が高まっています。AIエージェントは自律的に判断し行動できる特性を持ちますが、その能力を最大限に発揮するためには、様々な外部ツールと効果的に連携する必要があります。
しかし、現状ではAIエージェントと外部ツールの連携方法が標準化されておらず、各プラットフォームやLLMが独自の手法を提供しています。例えば、OpenAIではFunction Calling、LangChainではTool Callingといった形で実装が異なります。結果として、ツールの互換性の担保が難しく、AIエージェントの開発効率を大きく低下する要因となっています。
MCP（Model Context Protocol）はこのような課題を解決するために、2024年にAnthropicを中心としたAI企業コンソーシアムによって提唱されました。MCPは、LLMと外部ツールの間の標準化されたコミュニケーションプロトコルとして、AIエージェントのエコシステムを簡素化し、より効率的で強力なAIエージェント開発を可能にすることを目指しています。
複雑で大規模なAIエージェント開発のニーズが高まるとともにMCPの真価は高く評価されるようになり、2025年2月ごろからは”MCP Hype”（MCPの過熱なお祭り騒ぎ）とささやかれるまでになりました。
本記事ではAIエージェントの縁の下の力持ちとして注目を集めるMCPについて、主に実装コストの側面から解説することで、適切な技術選択の判断材料を提供します。
MCPの概要
MCPとは
MCPはLLMと外部ツールの間での通信を標準化するためのプロトコルです。AIエージェントが多様なツールを統一的な方法で利用できるようにすることを目的としています。
MCPは、USBのような役割を担っていると考えるとわかりやすいです。USBは異なるデバイスを同じポートに接続できるように標準化されたインターフェースを提供しています。同様に、MCPはLLMと外部ツールの間に標準化されたインターフェースを提供することで、ツールの種類やLLMの種類に関わらず一貫した方法での連携を可能にします。
MCPプロトコルに従って実装することで、LLMはMCPクライアントとして、各ツールはMCPサーバーとして機能し、統一された方法でツールの機能を呼び出すことができます。
（Model Context Protocolをベースに筆者が作成）
その他外部ツール連携方法との違い
その他の主な外部ツール連携としてはOpenAIのFunction CallingやLangChainのTool Callingがあります。これらの仕組みでは、ツール自体の構築とは別に、ツールの仕様情報（関数名、パラメータ、説明など）を文字列で定義し、LLMのプロンプトに含めて連携します。この方法ではツールと仕様情報を別々に管理できるため開発自由度が高いメリットがあります。一方でプログラムによってツールの連携方法が異なったり、ツールの更新時には仕様情報も同時に更新する必要があったり、統一性の欠如が課題となることがあります。
一方、MCPではツール仕様の記述方法や通信プロセス自体が標準化されています。ツールの提供者はMCPの規格に従ってツールを実装する必要があり、ツールの利用者もMCPの規格に従って連携する必要があります。ツールの構築やシステムとの連携方法に、作成者による違いが生じにくくなるため、統合や更新にも堅牢なシステムが構築できます。
MCPが解決する問題
MCPによるAPI仕様の標準化によって以下の問題が解決されます。
- LLMプラットフォーム変更の困難性
- 各プラットフォームで独自のツール連携手法が乱立していたため、異なるLLMプラットフォーム間でのツール共有が困難でした。MCPはこの問題を解決し、一度実装したツールを複数のLLMで利用可能にします。
- ツールの再利用の困難性
- MCPに準拠したツールはMCPに準拠した様々なAIエージェントで再利用できます。これにより、自身が過去に作成したツールや、他者が作成して公開したツールも調整なしにすぐ連携することができるため、開発コストの削減と機能拡張の効率化が実現します。
MCPの実装方法
MCP サーバーの実装
まずはMCPサーバーの実装を公式チュートリアルをベースに理解してみます。なお、このサーバーはアメリカの天気情報を返答するツールを提供します。
実際に実装例を確認することで、MCPサーバーの実装は実はかなりシンプルだということが実感できると思います。単純な関数の開発やTool Callingと比較しても、必要なオーバーヘッドは最小限です。
まず、ライブラリのインストールや初期化などの事前準備をします。
ここではFastMCPサーバーの初期化以外は特に変わったことはしていません。
from typing import Any
import httpx
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("weather")
NWS_API_BASE = "https://api.weather.gov"
USER_AGENT = "weather-app/1.0"次に、weather.gov APIを呼び出すためのヘルパー関数を定義します。
非同期処理（async/await）を使用してリクエストを非同期に処理する以外は特に変わったことはしていません。なお、非同期処理はMCP独自のものではなく、Tool Callingなど他のツール連携でも必要となる技法です。
async def make_nws_request(url: str) -> dict[str, Any] | None:
"""Make a request to the NWS API with proper error handling."""
headers = {
"User-Agent": USER_AGENT,
"Accept": "application/geo+json"
}
async with httpx.AsyncClient() as client:
try:
response = await client.get(url, headers=headers, timeout=30.0)
response.raise_for_status()
return response.json()
except Exception:
return None次に、メインパートであるMCPツールの定義をします。
特筆すべき内容は以下の二点です。
- @mcp.tool()デコレータの追加
- これによりツール登録を自動で処理できます。この実装方針はLangChainのTool Callingとも類似しています。
- DocStringとType Hint
- これによりホストがツールを理解するためのデータを自動で生成します。DocStringには入力スキーマとツールの説明を含みます。この方針もLangChainのTool Calling同様です。
@mcp.tool()
async def get_alerts(state: str) -> str:
"""Get weather alerts for a US state.
Args:
state: Two-letter US state code (e.g. CA, NY)
"""
def _format_alert(feature: dict) -> str:
props = feature["properties"]
return f"""
Event: {props.get('event', 'Unknown')}
Area: {props.get('areaDesc', 'Unknown')}
Severity: {props.get('severity', 'Unknown')}
Description: {props.get('description', 'No description available')}
Instructions: {props.get('instruction', 'No specific instructions provided')}
"""
url = f"{NWS_API_BASE}/alerts/active/area/{state}"
data = await make_nws_request(url)
if not data or "features" not in data:
return "Unable to fetch alerts or no alerts found."
if not data["features"]:
return "No active alerts for this state."
alerts = [_format_alert(feature) for feature in data["features"]]
return "\n---\n".join(alerts)
@mcp.tool()
async def get_forecast(latitude: float, longitude: float) -> str:
"""Get weather forecast for a location.
Args:
latitude: Latitude of the location
longitude: Longitude of the location
"""
points_url = f"{NWS_API_BASE}/points/{latitude},{longitude}"
points_data = await make_nws_request(points_url)
if not points_data:
return "Unable to fetch forecast data for this location."
forecast_url = points_data["properties"]["forecast"]
forecast_data = await make_nws_request(forecast_url)
if not forecast_data:
return "Unable to fetch detailed forecast."
periods = forecast_data["properties"]["periods"]
forecasts = []
for period in periods[:5]:
forecast = f"""
{period['name']}:
Temperature: {period['temperature']}°{period['temperatureUnit']}
Wind: {period['windSpeed']} {period['windDirection']}
Forecast: {period['detailedForecast']}
"""
forecasts.append(forecast)
return "\n---\n".join(forecasts)最後に、プログラム実行時にサーバーを起動するパートです。
通信方法としてstdioを定義しており、サーバーとクライアントは同じネットワーク内に存在し、標準入出力（stdio）を通じて通信することが期待されています。
なお、他にもSSE Transportという通信方式も選択できますが、2025年4月現在、実装が完了していない部分もあるようなのでstdioを選んでおくとよいでしょう。
if __name__ == "__main__":
mcp.run(transport='stdio')MCPクライアントの実装
前項で実装したMCPサーバーはMCPクライアントから呼び出しすることができます。
MCPクライアントの実装も公式チュートリアルをベースに理解してみます。
実際に実装例を確認することで、MCPクライアントの実装も非常に簡単であることが理解できると思います。唯一の追加作業はMCPサーバーとの通信の確立ですが、非常に軽微な作業です。
まず、ライブラリのインストールや初期化などの事前準備をします。
ここでは特に変わったことはしていません。
import asyncio
from typing import Optional
from contextlib import AsyncExitStack
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from anthropic import Anthropic
from dotenv import load_dotenv
load_dotenv() 次にMCPクライアントを定義します。
コンストラクタ内では、MCPサーバーとの非同期接続に必要なセッションと終了スタックを初期化しています。また、LLMのインスタンスを初期化しています。
なお、ここではAnthropicインスタンスを用いていますが、OpenAIインスタンスでも問題ありません。また、LangChainのインスタンスを利用する場合、ツール連携の部分で若干の修正が必要となりますが、大きな問題はありません。
class MCPClient:
def __init__(self):
self.session: Optional[ClientSession] = None
self.exit_stack = AsyncExitStack()
self.anthropic = Anthropic()
async def cleanup(self):
"""Clean up resources"""
await self.exit_stack.aclose()次に、MCPサーバーへの接続を行うメソッドを定義します。
MCPサーバーへの接続は以下の流れとなっています。
- サーバースクリプトのパス文字列（.pyまたは.js）を取得
- 通信のためのサーバーパラメータを初期化
- サーバーを起動し、サーバーとクライアント間のハンドシェイクを確立
（MCPクライアント定義の続き）
async def connect_to_server(self, server_script_path: str):
"""Connect to an MCP server
Args:
server_script_path: Path to the server script (.py or .js)
"""
is_python = server_script_path.endswith('.py')
is_js = server_script_path.endswith('.js')
if not (is_python or is_js):
raise ValueError("Server script must be a .py or .js file")
command = "python" if is_python else "node"
server_params = StdioServerParameters(
command=command,
args=[server_script_path],
env=None
)
stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))
self.stdio, self.write = stdio_transport
self.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))
await self.session.initialize()続いて、メインパートである、LLMとツールの連携および実行を行います。
手順は以下の通りです。
- MCPプロトコルで利用可能なツールリストの取得
- プロンプトを通じて利用可能なツールをLLMへ連携
- LLMのレスポンスを処理してツール実行の要求を確認
- MCPプロトコルでツールを実行
「1.MCPプロトコルで利用可能なツールリストの取得」と「4.MCPプロトコルでツールを実行」ではMCPプロトコルを利用して処理を簡単にしていますが、全体の流れとしてはTool Callingとそれほど差がないことがわかります。
MCPプロトコルを利用するパートについて少しだけ補足します。
1.MCPプロトコルで利用可能なツールリストの取得
利用可能なツールとそのスキーマは、セッションが確立されたMCPサーバーのインスタンスに対して以下のメソッドを実行することで簡単に取得できます。
session.list_tools()利用可能なツールのスキーマはname, description, inputSchemaのattributesを持つToolインスタンスで返答されるので、これをdictionaryなどに展開すればtool callingなどと同じように扱うことができます。
これは対応するLLMがClaudeだけに限定されるものではなく、OpenAIのFunction CallingやLangChainのTool Callingにも互換性があることを示しています。
4.MCPプロトコルでツールを実行
MCPではツールの実行もMCPサーバーインスタンスを通じて実行することができます。ツール名や引数などは隠蔽されており、統一的な記述ができます。
session.call_tool(tool_name, tool_args)（MCPクライアント定義の続き）
async def process_query(self, query: str) -> str:
"""Process a query using Claude and available tools"""
messages = [
{
"role": "user",
"content": query
}
]
response = await self.session.list_tools()
available_tools = [{
"name": tool.name,
"description": tool.description,
"input_schema": tool.inputSchema
} for tool in response.tools]
response = self.anthropic.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=1000,
messages=messages,
tools=available_tools
)
final_text = []
for content in response.content:
if content.type == 'text':
final_text.append(content.text)
elif content.type == 'tool_use':
tool_name = content.name
tool_args = content.input
result = await self.session.call_tool(tool_name, tool_args)
final_text.append(f"[Calling tool {tool_name} with args {tool_args}]")
if hasattr(content, 'text') and content.text:
messages.append({
"role": "assistant",
"content": content.text
})
messages.append({
"role": "user",
"content": result.content
})
response = self.anthropic.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=1000,
messages=messages,
)
final_text.append(response.content[0].text)
return "\n".join(final_text)続いて、以下メソッドではLLMとのチャットを終了の合図が入力されるまで継続する処理が記載されています。一般的なチャットボット形式のLLMの実装です。
（MCPクライアント定義の続き）
async def chat_loop(self):
"""Run an interactive chat loop"""
print("\nMCP Client Started!")
print("Type your queries or 'quit' to exit.")
while True:
try:
query = input("\nQuery: ").strip()
if query.lower() == 'quit':
break
response = await self.process_query(query)
print("\n" + response)
except Exception as e:
print(f"\nError: {str(e)}")メイン関数内ではすべての実行を行います。
クライアントを初期化、サーバー接続を確立し、チャットループを開始しています。
async def main():
if len(sys.argv) < 2:
print("Usage: python client.py <path_to_server_script>")
sys.exit(1)
client = MCPClient()
try:
await client.connect_to_server(sys.argv[1])
await client.chat_loop()
finally:
await client.cleanup()最後に、メイン関数を実行しています。
if __name__ == "__main__":
import sys
asyncio.run(main())MCPサーバーの実行
ここまでMCPサーバーの実装とMCPクライアントの実装を見てきました。ここでは実際にそれらを連携して実行します。
実はサーバーとクライアントの実行もとても簡単です。 サーバーへのパスを指定してクライアントを起動することで実行できます。 なお、この際にサーバーを個別に起動する必要はなく、クライアントが接続を確立する前にサーバーを起動する仕組みになっています。
uv run client.py path/to/server.pyMCPサーバーとLangChainの統合
MCPクライアントの実装では、MCPサーバーのインスタンスへlist_toolsメソッドを実行することでツールのスキーマが簡単に取得できることを確認しました。
しかし、LangChainのTool Callingはスキーマだけでなく関数自体もToolとして登録することを要求するため、MCPによるスキーマの連携だけでは統合が行えません。対応策としてはLangChainのBaseToolを継承したToolを定義して、スキーマだけlist_toolsメソッドの返り値をパースして、_runにはダミーの処理を登録する方法があります。ただしこれは保守性を下げるアプローチなので望ましくありません。
おすすめのMCPサーバーとLangChainの統合方法は、LangChain MCP Adaptersを利用することです。LangChain MCP Adaptersを利用することで、たったコード1行の追加によってMCPサーバーとLangChain/LangGraphコードベースの間をシームレスに接続することができます。
tools = await load_mcp_tools(session)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
model = ChatOpenAI(model="gpt-4o")
server_params = StdioServerParameters(
command="python",
args=["/path/to/math_server.py"],
)
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
tools = await load_mcp_tools(session)
agent = create_react_agent(model, tools)
agent_response = await agent.ainvoke({"messages": "what's (3 + 5) x 12?"})まとめ
AIエージェントのツール統合における利便性向上を背景に注目が高まってきているMCPについて、その実装コストを実装例に沿って解説しました。MCPサーバーおよびMCPクライアントの開発は、Tool Callingなど他の手法と比較してもそれほど大きな追加工数がないことがお伝えできたかと思います。同時に、このちょっとした追加開発を行うことで、数千にものぼるOSSのMCPサーバーが即座に利用可能となるメリットもご理解いただけたかと思います。
MCPとそのエコシステムは発展途上であり、今後その重要性が増すことに疑いの余地はありません。この記事がMCPへの入門として役立てば幸いです。