💨

APIサービスをMCP Toolとして実装するワークフロー

に公開
Model Context Protocol
Cursor
Claude
tech

1. 要件定義

  • 使用するAPIサービスを特定する
  • 必要なAPIエンドポイントを特定する
  • ツールの入力パラメータと出力形式を決定する

2. APIドキュメントの調査とエンドポイントの特定

APIドキュメントの探し方

  • 公式サイトの「Developers」「API」「Documentation」セクションを探す
  • Google検索で「[サービス名] API documentation」と検索する
  • GitHub上の公式SDKやライブラリを調査する

OpenAPIドキュメントの活用

  • 多くのAPIはOpenAPI/Swagger形式のドキュメントを提供している
  • APIエンドポイント、必要なパラメータ、レスポンス形式が体系的に記述されている
  • SwaggerUIなどの対話型ドキュメントがあれば、それを使ってAPIをテストできる

3. APIのテストと動作確認

APIエンドポイントのテスト方法

コマンドラインツール

# curlを使用した例（天気API）
curl "https://api.openweathermap.org/data/2.5/weather?lat=35.6895&lon=139.6917&appid=YOUR_API_KEY&units=metric&lang=ja"

# PowerShellの場合
Invoke-RestMethod -Uri "https://api.openweathermap.org/data/2.5/weather?lat=35.6895&lon=139.6917&appid=YOUR_API_KEY&units=metric&lang=ja" | ConvertTo-Json -Depth 5

GUIツール

  • Postman: API呼び出しの作成、テスト、共有が可能
  • Insomnia: シンプルで使いやすいREST APIクライアント
  • Thunder Client: VSCode拡張、エディタ内でAPIテスト可能

ブラウザ拡張

  • JSONView: JSONレスポンスを整形して表示
  • REST Client: VSCodeの拡張機能、コード内でAPIリクエストをテスト

API応答の分析

  1. レスポンスのJSON構造を把握する
  2. 必要なデータがどのパスに存在するかをメモする
  3. レスポンスサンプルをコピーしておく
  4. エラーレスポンスのパターンも確認しておく

オンラインツール

  • RequestBin: APIのリクエスト/レスポンスを検査
  • JSON Formatter: 応答を整形して読みやすくする

サンプルコード

# 対話的にAPIのレスポンスを調べるためのPythonスクリプト
import httpx
import json
from rich.console import Console
from rich.syntax import Syntax

console = Console()

async def test_api(url, method="GET", headers=None, params=None, json_data=None):
    """APIをテストして応答を整形して表示する"""
    async with httpx.AsyncClient() as client:
        request_kwargs = {"url": url}
        if headers:
            request_kwargs["headers"] = headers
        if params:
            request_kwargs["params"] = params
        if json_data and method.upper() in ["POST", "PUT", "PATCH"]:
            request_kwargs["json"] = json_data
        
        response = await getattr(client, method.lower())(**request_kwargs)
        status_code = response.status_code
        
        try:
            data = response.json()
            formatted_json = json.dumps(data, indent=2, ensure_ascii=False)
            console.print(f"Status Code: {status_code}")
            console.print(Syntax(formatted_json, "json", theme="monokai"))
            
            return data
        except:
            console.print(f"Status Code: {status_code}")
            console.print(response.text)
            return response.text

# 使用例
# await test_api("https://api.openweathermap.org/data/2.5/weather", 
#               params={"lat": 35.6895, "lon": 139.6917, "appid": "YOUR_API_KEY", "units": "metric", "lang": "ja"})

4. 環境準備

  • 必要なライブラリをインストールする
pip install fastmcp httpx
  • APIキーの取得と管理方法を決める

5. ベーステンプレートの作成

from fastmcp import FastMCP
import httpx
import os
from typing import Optional, Dict, Any

mcp = FastMCP("api-tools")

# ここにツールを追加していく

if __name__ == "__main__":
    mcp.run(transport="stdio")

6. API接続のユーティリティ関数作成

async def make_api_request(url: str, method: str = "GET", headers: Dict[str, str] = None, 
                         params: Dict[str, Any] = None, json_data: Dict[str, Any] = None) -> Dict:
    """API呼び出しの共通処理"""
    async with httpx.AsyncClient() as client:
        request_kwargs = {"url": url, "headers": headers}
        if params:
            request_kwargs["params"] = params
        if json_data and method.upper() in ["POST", "PUT", "PATCH"]:
            request_kwargs["json"] = json_data
        
        response = await getattr(client, method.lower())(**request_kwargs)
        response.raise_for_status()
        return response.json()

7. 共通例外処理とエラーハンドリング

def handle_api_errors(func):
    """API呼び出しの共通エラーハンドリング用デコレータ"""
    async def wrapper(*args, **kwargs):
        try:
            return await func(*args, **kwargs)
        except httpx.HTTPStatusError as e:
            status_code = e.response.status_code
            if status_code == 401:
                return "認証エラー: APIキーが無効または期限切れです"
            elif status_code == 403:
                return "アクセス拒否: このリソースにアクセスする権限がありません"
            elif status_code == 404:
                return "リソースが見つかりません"
            elif status_code == 429:
                return "リクエスト制限に達しました。しばらく待ってから再試行してください"
            else:
                return f"APIリクエスト中にエラーが発生しました: {status_code} - {e.response.text}"
        except httpx.RequestError as e:
            return f"ネットワークエラー: {str(e)}"
        except Exception as e:
            return f"予期しないエラーが発生しました: {str(e)}"
    return wrapper

8. 各APIサービス用のツール作成テンプレート

@mcp.tool()
@handle_api_errors
async def api_service_tool(param1: str, param2: int, api_key: Optional[str] = None) -> str:
    """
    APIサービスの機能を利用するツール
    
    Args:
        param1: 最初のパラメータの説明
        param2: 2番目のパラメータの説明
        api_key: APIキー（指定しない場合は環境変数から取得）
        
    Returns:
        処理結果
    """
    # APIキーの取得（引数で指定されていない場合は環境変数から）
    api_key = api_key or os.environ.get("SERVICE_API_KEY")
    
    if not api_key:
        return "APIキーが指定されていません。引数でapi_keyを指定するか、環境変数SERVICE_API_KEYを設定してください。"
    
    # APIリクエストのパラメータ準備
    headers = {"Authorization": f"Bearer {api_key}"}
    params = {"param1": param1, "param2": param2}
    
    # APIリクエスト実行
    result = await make_api_request(
        url="https://api.service.com/endpoint",
        headers=headers,
        params=params
    )
    
    # 結果の整形と返却
    return format_result(result)

9. レスポンス整形関数の作成

def format_result(data: Dict) -> str:
    """APIレスポンスを読みやすい形式に整形する"""
    formatted_text = f"""
タイトル: {data.get('title', '不明')}
内容: {data.get('description', '説明なし')}
その他情報: {data.get('additional_info', 'なし')}
    """
    return formatted_text.strip()

10. 実装例: 天気情報API

@mcp.tool()
@handle_api_errors
async def get_weather(latitude: float, longitude: float, api_key: Optional[str] = None) -> str:
    """
    指定された緯度と経度の場所の現在の天気情報を取得する
    
    Args:
        latitude: 場所の緯度
        longitude: 場所の経度
        api_key: OpenWeatherMap APIキー（指定しない場合は環境変数から取得）
        
    Returns:
        天気情報の文字列
    """
    api_key = api_key or os.environ.get("OPENWEATHERMAP_API_KEY")
    
    if not api_key:
        return "APIキーが指定されていません。"
    
    url = f"https://api.openweathermap.org/data/2.5/weather"
    params = {
        "lat": latitude,
        "lon": longitude,
        "appid": api_key,
        "units": "metric",
        "lang": "ja"
    }
    
    data = await make_api_request(url=url, params=params)
    
    return f"""
場所: {data.get('name', '不明な場所')}
天気: {data['weather'][0]['description']}
気温: {data['main']['temp']}°C（体感温度: {data['main']['feels_like']}°C）
湿度: {data['main']['humidity']}%
風速: {data['wind']['speed']}m/s
    """.strip()

11. 実装例: ニュースAPI

@mcp.tool()
@handle_api_errors
async def get_news(query: str, api_key: Optional[str] = None) -> str:
    """
    指定されたキーワードに関するニュース記事を取得する
    
    Args:
        query: 検索キーワード
        api_key: NewsAPI APIキー（指定しない場合は環境変数から取得）
        
    Returns:
        ニュース記事のリスト
    """
    api_key = api_key or os.environ.get("NEWSAPI_API_KEY")
    
    if not api_key:
        return "APIキーが指定されていません。"
    
    url = "https://newsapi.org/v2/everything"
    params = {
        "q": query,
        "sortBy": "publishedAt",
        "language": "ja",
        "pageSize": 5
    }
    headers = {"X-Api-Key": api_key}
    
    data = await make_api_request(url=url, params=params, headers=headers)
    
    articles = data.get("articles", [])
    if not articles:
        return "記事が見つかりませんでした。"
    
    result = "【ニュース記事】\n\n"
    for i, article in enumerate(articles, 1):
        result += f"{i}. {article['title']}\n"
        result += f"   出典: {article['source']['name']}\n"
        result += f"   URL: {article['url']}\n"
        result += f"   公開日: {article['publishedAt']}\n\n"
    
    return result.strip()

12. テスト方法

単体テスト: APIモックを使用

# テストコード例
import pytest
from unittest.mock import patch, AsyncMock

@pytest.mark.asyncio
async def test_get_weather():
    mock_response = {
        "name": "東京",
        "weather": [{"description": "晴れ"}],
        "main": {"temp": 25.5, "feels_like": 26.2, "humidity": 60},
        "wind": {"speed": 2.1}
    }
    
    with patch('httpx.AsyncClient.get', new_callable=AsyncMock) as mock_get:
        mock_get.return_value.json.return_value = mock_response
        mock_get.return_value.raise_for_status = AsyncMock()
        
        result = await get_weather(35.6895, 139.6917, "test_api_key")
        assert "東京" in result
        assert "晴れ" in result

手動テスト: MCPインスペクタを使用

mcp dev path/to/your/file.py

13. APIレスポンスの探索と分析

APIレスポンスが複雑な場合、以下のようなコードでJSONデータをインタラクティブに探索できます：

import json
from rich.console import Console
from rich.tree import Tree
from rich.syntax import Syntax

def explore_json(data, max_depth=3, current_depth=0):
    """JSONデータを再帰的に探索して構造を表示する"""
    console = Console()
    
    def build_tree(data, tree, depth=0):
        if depth >= max_depth:
            return
        
        if isinstance(data, dict):
            for key, value in data.items():
                if isinstance(value, (dict, list)) and value:
                    branch = tree.add(f"[bold blue]{key}[/]")
                    build_tree(value, branch, depth + 1)
                else:
                    if isinstance(value, str) and len(value) > 50:
                        value = value[:50] + "..."
                    tree.add(f"[bold green]{key}[/]: {value}")
        elif isinstance(data, list):
            if data:
                if len(data) > 5:
                    # サンプルとして最初の数項目のみ表示
                    for i, item in enumerate(data[:3]):
                        branch = tree.add(f"[bold magenta]{i}[/]")
                        build_tree(item, branch, depth + 1)
                    tree.add(f"... ({len(data) - 3} more items)")
                else:
                    for i, item in enumerate(data):
                        branch = tree.add(f"[bold magenta]{i}[/]")
                        build_tree(item, branch, depth + 1)
    
    root = Tree("[bold red]Root[/]")
    build_tree(data, root)
    console.print(root)

# 使用例
# with open('api_response.json', 'r', encoding='utf-8') as f:
#     data = json.load(f)
#     explore_json(data)

データパスの抽出

特定のデータパスを特定する関数：

def extract_paths(data, prefix=""):
    """JSONデータ内の全てのパスを抽出する"""
    paths = []
    
    def _extract(obj, current_path):
        if isinstance(obj, dict):
            for k, v in obj.items():
                new_path = f"{current_path}.{k}" if current_path else k
                paths.append(new_path)
                _extract(v, new_path)
        elif isinstance(obj, list) and obj:
            # 最初の要素のみを探索（リスト内の構造が同じと仮定）
            _extract(obj[0], f"{current_path}[0]")
    
    _extract(data, prefix)
    return paths

# 使用例
# paths = extract_paths(api_response)
# for path in paths:
#     print(path)

14. デプロイとセキュリティ

  • APIキーを環境変数として設定
  • 本番環境では.envファイルや環境変数管理サービスを使用
  • 適切なエラーハンドリングを確保
  • レート制限の考慮
  • 必要に応じてAPIキーの使用量をモニタリング

15. ドキュメント作成

  • 各ツールの使用方法を記述
  • 必要なAPIキーの取得方法を説明
  • 実装例やサンプルコードを提供
  • 一般的なエラーとその解決策を記載

参考: 主要なAPIサービスと公式ドキュメント

サービス名 概要 公式ドキュメントURL
OpenWeatherMap 気象データAPI https://openweathermap.org/api
NewsAPI ニュースデータAPI https://newsapi.org/docs
Google Maps 地図・位置情報API https://developers.google.com/maps/documentation
Twitter API ツイートデータAPI https://developer.twitter.com/en/docs
GitHub API GitHubデータAPI https://docs.github.com/en/rest
DeepL API 翻訳API https://www.deepl.com/docs-api
Currency Exchange 為替レートAPI https://exchangeratesapi.io/documentation/
Spotify API 音楽データAPI https://developer.spotify.com/documentation/web-api
TMDB 映画データAPI https://developers.themoviedb.org/3/getting-started/introduction
NASA API 宇宙データAPI https://api.nasa.gov/

Discussion