MCPクライアントの作成方法まとめ（Python実装例付き）

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())

pip install langchain langchain-mcp-adapters langchain-ollama langgraph