Go to Qiita Advent Calendar Top
LoginSignup
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@555hamano(Akr)

Model Context Protocol完全解説 30日間シリーズ - Day 12【MCP実装 #12】Toolsの実装方法：電卓機能をMCP経由で提供してみよう

Last updated at Posted at 2025-09-15

はじめに

この記事は、QiitaのModel Context Protocol（以下、MCP）解説シリーズの第12回です。

今回は、MCPのもう一つの主要機能であるToolsの実装方法を学びます。具体的な例として、LLMに計算をさせるための電卓機能をMCP経由で提供してみましょう。

🛠️ なぜToolsが必要なのか？

LLMは言語処理に優れていますが、以下のようなタスクには限界があります：

LLMが苦手なタスク

  • 精密な数値計算: 大きな数値や複雑な計算で誤りが生じやすい
  • 外部システムへのアクセス: データベース操作、API呼び出しなど
  • リアルタイム情報取得: 現在の時刻、最新のデータなど
  • ファイル操作: 実際のファイル読み書き、処理など

Toolsの役割

Toolsは、LLMが苦手なタスクを外部の専門的な機能に委譲するための仕組みです。これにより、LLMは：

  1. 正確性の向上: 専門的なツールを使用することで、より正確な結果を得られる
  2. 機能拡張: 元々持たない機能を外部ツールで補完
  3. 効率性: 得意な言語処理に集中し、他のタスクはツールに任せる

今回の例では、以下の計算機能を持つToolsを実装します：

  • add: 2つの数値の加算
  • subtract: 2つの数値の減算
  • multiply: 2つの数値の乗算
  • divide: 2つの数値の除算（ゼロ除算エラー処理含む）
  • calculate: 数式文字列の評価

📝 ステップ1：プロジェクトのセットアップ

Pythonを使って実装します。TypeScript版も後ほど紹介します。

# プロジェクトディレクトリを作成
mkdir mcp-calculator
cd mcp-calculator

# 仮想環境を作成・有効化
python -m venv mcp-env
source mcp-env/bin/activate  # Windows: mcp-env\Scripts\activate

# 必要なパッケージをインストール
pip install mcp

📝 ステップ2：Python版Calculator Toolsの実装

calculator_server.pyファイルを作成し、以下のコードを実装してください：

import asyncio
import logging
import math
import re
from typing import List, Union, Any

from mcp.server.models import InitializationOptions
import mcp.types as types
from mcp.server import NotificationOptions, Server

# ログ設定
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# MCPサーバーを作成
server = Server("calculator-server")

class CalculatorTools:
    """計算機能を提供するクラス"""
    
    @staticmethod
    def safe_eval(expression: str) -> float:
        """安全な数式評価"""
        # 許可された文字のみ（数字、演算子、括弧、小数点、空白）
        allowed_pattern = r'^[\d+\-*/().\s]+$'
        if not re.match(allowed_pattern, expression):
            raise ValueError("Invalid characters in expression")
            
        # 危険な関数名を除外
        dangerous_names = ['__', 'eval', 'exec', 'import', 'open', 'input']
        for name in dangerous_names:
            if name in expression:
                raise ValueError(f"Dangerous function '{name}' not allowed")
        
        try:
            # 安全な環境で評価
            safe_dict = {
                "__builtins__": {},
                "abs": abs,
                "round": round,
                "min": min,
                "max": max,
                "pow": pow,
                "sqrt": math.sqrt,
                "sin": math.sin,
                "cos": math.cos,
                "tan": math.tan,
                "pi": math.pi,
                "e": math.e,
            }
            
            result = eval(expression, safe_dict, safe_dict)
            return float(result)
            
        except ZeroDivisionError:
            raise ValueError("Division by zero")
        except Exception as e:
            raise ValueError(f"Invalid expression: {str(e)}")

calculator = CalculatorTools()

@server.list_tools()
async def handle_list_tools() -> List[types.Tool]:
    """利用可能なツールの一覧を返す"""
    return [
        types.Tool(
            name="add",
            description="Add two numbers together",
            inputSchema={
                "type": "object",
                "properties": {
                    "a": {
                        "type": "number",
                        "description": "The first number to add",
                    },
                    "b": {
                        "type": "number", 
                        "description": "The second number to add",
                    },
                },
                "required": ["a", "b"],
            },
        ),
        types.Tool(
            name="subtract",
            description="Subtract one number from another",
            inputSchema={
                "type": "object",
                "properties": {
                    "a": {
                        "type": "number",
                        "description": "The number to subtract from",
                    },
                    "b": {
                        "type": "number",
                        "description": "The number to subtract",
                    },
                },
                "required": ["a", "b"],
            },
        ),
        types.Tool(
            name="multiply",
            description="Multiply two numbers together",
            inputSchema={
                "type": "object",
                "properties": {
                    "a": {
                        "type": "number",
                        "description": "The first number to multiply",
                    },
                    "b": {
                        "type": "number",
                        "description": "The second number to multiply", 
                    },
                },
                "required": ["a", "b"],
            },
        ),
        types.Tool(
            name="divide",
            description="Divide one number by another",
            inputSchema={
                "type": "object",
                "properties": {
                    "a": {
                        "type": "number",
                        "description": "The dividend (number to be divided)",
                    },
                    "b": {
                        "type": "number",
                        "description": "The divisor (number to divide by)",
                    },
                },
                "required": ["a", "b"],
            },
        ),
        types.Tool(
            name="calculate",
            description="Evaluate a mathematical expression safely. Supports basic arithmetic operations, parentheses, and common math functions like sqrt, sin, cos, tan, abs, round, min, max, pow. Also provides constants pi and e.",
            inputSchema={
                "type": "object",
                "properties": {
                    "expression": {
                        "type": "string",
                        "description": "Mathematical expression to evaluate (e.g., '2 + 3 * 4', 'sqrt(16)', 'sin(pi/2)')",
                    },
                },
                "required": ["expression"],
            },
        ),
        types.Tool(
            name="power",
            description="Calculate the power of a number",
            inputSchema={
                "type": "object",
                "properties": {
                    "base": {
                        "type": "number",
                        "description": "The base number",
                    },
                    "exponent": {
                        "type": "number",
                        "description": "The exponent",
                    },
                },
                "required": ["base", "exponent"],
            },
        ),
        types.Tool(
            name="sqrt",
            description="Calculate the square root of a number",
            inputSchema={
                "type": "object",
                "properties": {
                    "number": {
                        "type": "number",
                        "description": "The number to calculate square root of (must be non-negative)",
                    },
                },
                "required": ["number"],
            },
        ),
    ]

@server.call_tool()
async def handle_call_tool(name: str, arguments: dict | None) -> List[types.TextContent]:
    """ツールの実行を処理"""
    if not arguments:
        return [types.TextContent(type="text", text="Error: No arguments provided")]
        
    try:
        if name == "add":
            result = arguments["a"] + arguments["b"]
            return [types.TextContent(
                type="text", 
                text=f"{arguments['a']} + {arguments['b']} = {result}"
            )]
            
        elif name == "subtract":
            result = arguments["a"] - arguments["b"]
            return [types.TextContent(
                type="text", 
                text=f"{arguments['a']} - {arguments['b']} = {result}"
            )]
            
        elif name == "multiply":
            result = arguments["a"] * arguments["b"]
            return [types.TextContent(
                type="text", 
                text=f"{arguments['a']} × {arguments['b']} = {result}"
            )]
            
        elif name == "divide":
            if arguments["b"] == 0:
                return [types.TextContent(
                    type="text", 
                    text="Error: Division by zero is not allowed"
                )]
            result = arguments["a"] / arguments["b"]
            return [types.TextContent(
                type="text", 
                text=f"{arguments['a']} ÷ {arguments['b']} = {result}"
            )]
            
        elif name == "calculate":
            try:
                result = calculator.safe_eval(arguments["expression"])
                return [types.TextContent(
                    type="text", 
                    text=f"{arguments['expression']} = {result}"
                )]
            except ValueError as e:
                return [types.TextContent(
                    type="text", 
                    text=f"Calculation error: {str(e)}"
                )]
                
        elif name == "power":
            try:
                result = pow(arguments["base"], arguments["exponent"])
                return [types.TextContent(
                    type="text", 
                    text=f"{arguments['base']}^{arguments['exponent']} = {result}"
                )]
            except (OverflowError, ValueError) as e:
                return [types.TextContent(
                    type="text", 
                    text=f"Power calculation error: {str(e)}"
                )]
                
        elif name == "sqrt":
            if arguments["number"] < 0:
                return [types.TextContent(
                    type="text", 
                    text="Error: Cannot calculate square root of negative number"
                )]
            result = math.sqrt(arguments["number"])
            return [types.TextContent(
                type="text", 
                text=f"√{arguments['number']} = {result}"
            )]
            
        else:
            return [types.TextContent(
                type="text", 
                text=f"Error: Unknown tool '{name}'"
            )]
            
    except KeyError as e:
        return [types.TextContent(
            type="text", 
            text=f"Error: Missing required argument {str(e)}"
        )]
    except Exception as e:
        logger.error(f"Error in tool {name}: {e}")
        return [types.TextContent(
            type="text", 
            text=f"Error: {str(e)}"
        )]

async def main():
    """サーバーのメイン関数"""
    logger.info("Starting calculator MCP server...")
    
    # stdio transport
    from mcp.server.stdio import stdio_server
    
    async with stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            InitializationOptions(
                server_name="calculator-server",
                server_version="1.0.0",
                capabilities=server.get_capabilities(
                    notification_options=NotificationOptions(),
                    experimental_capabilities={},
                ),
            ),
        )

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

👆 ポイント解説

  • セキュリティ対策: safe_eval関数で危険な関数の実行を防止
  • エラーハンドリング: ゼロ除算、負数の平方根など適切なエラー処理
  • 豊富な数学機能: 基本四則演算から三角関数、定数まで対応
  • 明確な説明: 各ツールの詳細な説明とパラメータ定義

📝 ステップ3：TypeScript版の実装例

TypeScript版も参考として紹介します：

import { Server } from '@modelcontextprotocol/sdk/server/index.js';
import { StdioServerTransport } from '@modelcontextprotocol/sdk/server/stdio.js';
import {
  CallToolRequestSchema,
  ListToolsRequestSchema,
} from '@modelcontextprotocol/sdk/types.js';

class CalculatorServer {
  private server: Server;

  constructor() {
    this.server = new Server(
      {
        name: 'calculator-server',
        version: '1.0.0',
      },
      {
        capabilities: {
          tools: {},
        },
      }
    );

    this.setupHandlers();
  }

  private setupHandlers() {
    this.server.setRequestHandler(ListToolsRequestSchema, async () => {
      return {
        tools: [
          {
            name: 'add',
            description: 'Add two numbers together',
            inputSchema: {
              type: 'object',
              properties: {
                a: { type: 'number', description: 'First number' },
                b: { type: 'number', description: 'Second number' },
              },
              required: ['a', 'b'],
            },
          },
          {
            name: 'calculate',
            description: 'Evaluate a mathematical expression',
            inputSchema: {
              type: 'object',
              properties: {
                expression: { 
                  type: 'string', 
                  description: 'Mathematical expression to evaluate' 
                },
              },
              required: ['expression'],
            },
          },
        ],
      };
    });

    this.server.setRequestHandler(CallToolRequestSchema, async (request) => {
      const { name, arguments: args } = request.params;

      try {
        switch (name) {
          case 'add':
            const sum = args.a + args.b;
            return {
              content: [{ type: 'text', text: `${args.a} + ${args.b} = ${sum}` }],
            };

          case 'calculate':
            // 簡単な例（実際にはより安全な実装が必要）
            const result = Function('"use strict"; return (' + args.expression + ')')();
            return {
              content: [{ type: 'text', text: `${args.expression} = ${result}` }],
            };

          default:
            throw new Error(`Unknown tool: ${name}`);
        }
      } catch (error) {
        return {
          content: [{ type: 'text', text: `Error: ${error}` }],
          isError: true,
        };
      }
    });
  }

  async start() {
    const transport = new StdioServerTransport();
    await this.server.connect(transport);
    console.error('Calculator MCP server running on stdio');
  }
}

async function main() {
  const server = new CalculatorServer();
  await server.start();
}

main().catch((error) => {
  console.error('Server error:', error);
  process.exit(1);
});

🚀 ステップ3：サーバーの実行とテスト

1. サーバーの起動

# Python版
python calculator_server.py

2. Claude Desktopでの設定

claude_desktop_config.jsonに以下を追加：

{
  "mcpServers": {
    "calculator": {
      "command": "python",
      "args": ["/path/to/your/project/calculator_server.py"]
    }
  }
}

3. Claude Desktopでのテスト

以下のような質問でテストしてみましょう：

基本的な計算

23 + 45 を計算してください

1234 × 5678 を計算してください

複雑な数式

次の数式を計算してください: (2 + 3) * 4 - 10 / 2

2の10乗を計算してください

三角関数

sin(π/2) の値を計算してください

4. 期待される応答例

質問: "複雑な計算をお願いします。(15 + 25) × 3 - sqrt(144) を計算してください"

Claudeの応答:

複雑な計算を実行します。

(15 + 25) × 3 - sqrt(144) = 108

計算過程：
- 15 + 25 = 40
- 40 × 3 = 120  
- sqrt(144) = 12
- 120 - 12 = 108

答えは 108 です。

🧠 高度なTools実装パターン

1. 状態を持つツール

class StatefulCalculator:
    """計算履歴を保持する電卓"""
    
    def __init__(self):
        self.history: List[str] = []
        self.memory: float = 0.0
    
    def add_to_history(self, operation: str, result: float):
        self.history.append(f"{operation} = {result}")
        
    def get_history(self) -> List[str]:
        return self.history.copy()
        
    def clear_history(self):
        self.history.clear()
        
    def store_in_memory(self, value: float):
        self.memory = value
        
    def recall_memory(self) -> float:
        return self.memory

# グローバルインスタンス
stateful_calc = StatefulCalculator()

# 履歴管理ツールを追加
types.Tool(
    name="get_calculation_history",
    description="Get the history of recent calculations",
    inputSchema={"type": "object", "properties": {}},
),
types.Tool(
    name="clear_history", 
    description="Clear the calculation history",
    inputSchema={"type": "object", "properties": {}},
),

2. 外部サービス連携

import requests

async def handle_currency_conversion(arguments: dict) -> List[types.TextContent]:
    """通貨変換ツール"""
    try:
        from_currency = arguments["from_currency"].upper()
        to_currency = arguments["to_currency"].upper()
        amount = arguments["amount"]
        
        # 外部API（例：exchangerate-api.com）を呼び出し
        api_url = f"https://api.exchangerate-api.com/v4/latest/{from_currency}"
        response = requests.get(api_url, timeout=10)
        data = response.json()
        
        if to_currency not in data["rates"]:
            return [types.TextContent(
                type="text", 
                text=f"Error: Currency {to_currency} not supported"
            )]
            
        exchange_rate = data["rates"][to_currency]
        converted_amount = amount * exchange_rate
        
        return [types.TextContent(
            type="text", 
            text=f"{amount} {from_currency} = {converted_amount:.2f} {to_currency} (Rate: {exchange_rate})"
        )]
        
    except Exception as e:
        return [types.TextContent(
            type="text", 
            text=f"Currency conversion error: {str(e)}"
        )]

3. バリデーション強化

from pydantic import BaseModel, Field, validator

class CalculationInput(BaseModel):
    """計算入力のバリデーションモデル"""
    a: float = Field(..., description="First number")
    b: float = Field(..., description="Second number")
    
    @validator('a', 'b')
    def validate_finite(cls, v):
        if not math.isfinite(v):
            raise ValueError("Number must be finite")
        return v
        
    @validator('b')
    def validate_not_zero_for_division(cls, v, values):
        # この例では、除算専用のバリデーションが必要
        return v

class ExpressionInput(BaseModel):
    """数式入力のバリデーションモデル"""
    expression: str = Field(..., min_length=1, max_length=1000)
    
    @validator('expression')
    def validate_safe_expression(cls, v):
        # 危険な文字列をチェック
        dangerous_patterns = ['__', 'import', 'exec', 'eval', 'open']
        for pattern in dangerous_patterns:
            if pattern in v.lower():
                raise ValueError(f"Expression contains dangerous pattern: {pattern}")
        return v

🔐 セキュリティとベストプラクティス

セキュリティ考慮事項

  1. 入力検証: すべての入力を厳密に検証
  2. 実行環境の制限: 危険な関数や操作の制限
  3. リソース制限: 計算時間やメモリ使用量の制限
  4. ログ記録: すべての操作をログに記録

パフォーマンス最適化

  1. キャッシング: 計算結果のキャッシュ
  2. 非同期処理: 重い計算の非同期実行
  3. タイムアウト: 長時間実行の防止
import functools
import asyncio

# LRUキャッシュで計算結果をキャッシュ
@functools.lru_cache(maxsize=1000)
def cached_calculation(expression: str) -> float:
    """キャッシュ付き計算"""
    return calculator.safe_eval(expression)

# タイムアウト付き実行
async def safe_calculate_with_timeout(expression: str, timeout: float = 5.0) -> float:
    """タイムアウト付き安全な計算"""
    try:
        return await asyncio.wait_for(
            asyncio.get_event_loop().run_in_executor(
                None, cached_calculation, expression
            ),
            timeout=timeout
        )
    except asyncio.TimeoutError:
        raise ValueError("Calculation timeout: expression too complex")

🎯 まとめ

今回学んだ内容：

Tools実装の基本パターン

  1. ツール設計: 明確な名前と詳細な説明
  2. スキーマ定義: 厳密な入力パラメータの定義
  3. エラーハンドリング: 適切なエラー処理とメッセージ
  4. セキュリティ: 安全な実行環境の構築

高度な実装テクニック

  • 状態管理: 複数のツール呼び出し間での状態保持
  • 外部連携: API呼び出しや外部サービス統合
  • パフォーマンス: キャッシング、非同期処理、タイムアウト
  • バリデーション: Pydanticを使った強力な入力検証

実用的な応用例

この計算ツールの実装パターンを応用して、以下のような高度なツールセットを構築できます：

  • データ分析ツール: 統計計算、グラフ生成
  • API連携ツール: 外部サービスとの連携
  • ファイル処理ツール: CSV処理、データ変換
  • 通知ツール: メール送信、Slack投稿

次回は、MCPのPrompts機能について深く掘り下げ、動的なプロンプト生成とテンプレート活用について解説します。お楽しみに！

📚 参考リンク

0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?