AIアシスタントや外部ツール連携の現場で注目されている「MCP(Modular Command Protocol)」クライアントの作成方法を、初心者にも分かりやすく解説します。この記事では、MCPの基礎から、Pythonによる最小構成のクライアント実装例、さらにローカルLLM(Ollama)との連携まで、実践的な内容を網羅します。
MCPクライアントは、AIやアプリケーションからのリクエストを受け取り、適切なMCPサーバーへAPI呼び出しを行い、その結果をAIなどに返す役割を持ちます。つまり「AIのリクエストをAPIや外部ツールの呼び出しに変換し、結果を返す仲介役」です。
以下は、PythonでMCPクライアントを自作するシンプルな例です。AI連携なしで、ユーザーの指示に従いMCPサーバーのツールを直接呼び出します。
import asyncio
import json
from typing import Dict, Any
from contextlib import AsyncExitStack
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
class MCPClientSimple:
def __init__(self):
self.session = None
self.exit_stack = None
self.tools = []
async def connect(self, server_path: str):
if self.exit_stack:
await self.exit_stack.aclose()
self.exit_stack = AsyncExitStack()
command = "python" if server_path.endswith('.py') else "node"
server_params = StdioServerParameters(
command=command,
args=[server_path],
env={"PYTHONIOENCODING": "utf-8", "PYTHONUNBUFFERED": "1"}
)
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()
response = await self.session.list_tools()
self.tools = [tool.name for tool in response.tools]
print(f"Connected to MCP server. Available tools: {', '.join(self.tools)}")
async def call_tool(self, tool_name: str, params: Dict[str, Any]):
if not self.session:
print("Not connected to MCP server.")
return
if tool_name not in self.tools:
print(f"Tool '{tool_name}' is not available.")
return
result = await self.session.call_tool(tool_name, params)
print("Tool result:", result.content)
async def main():
server_path = input("Enter path to MCP server script (e.g., server.py): ")
client = MCPClientSimple()
await client.connect(server_path)
while True:
print("\nAvailable tools:", client.tools)
tool = input("Which tool do you want to use? (type tool name, or 'exit' to quit): ")
if tool == 'exit':
break
params_str = input("Enter parameters as JSON (e.g., {\"a\": 5, \"b\": 3}): ")
try:
params = json.loads(params_str)
except Exception as e:
print("Invalid JSON:", e)
continue
await client.call_tool(tool, params)
if client.exit_stack:
await client.exit_stack.aclose()
if __name__ == "__main__":
asyncio.run(main())
このコードを使えば、MCPサーバーに接続し、提供されているツール(例: add, subtractなど)を手動で選択・実行できます。
LangChainやOllamaと連携することで、AIエージェントが自動でMCPサーバーのツールを活用できるようになります。複数のMCPサーバーを統合し、LLMエージェントから一元的に利用するには、langchain-mcp-adapters
のMultiServerMCPClient
が便利です。
pip install langchain langchain-mcp-adapters langchain-ollama langgraph
MCPクライアントは、AIやアプリケーションからのリクエストを外部ツールやAPIに橋渡しする重要な役割を担います。Pythonによる最小構成から、LangChainやOllamaを使った高度な連携まで、用途や規模に応じた実装が可能です。まずはシンプルなクライアントから始めて、徐々に拡張していくのがおすすめです。