Yaizu Smart City MCP Server

Instructions

Implementation Reference

• mcp/server.py:426-426 (registration)

  Registration of the execute_yaizu_api tool using the FastMCP @mcp.tool() decorator.
  @mcp.tool()
• mcp/server.py:427-595 (handler)

  Core handler function that executes the Yaizu smart city API query for the specified entity_type, handling parameters, limits, and formatting the JSON response with summary.
  async def execute_yaizu_api(entity_type: str, params: Optional[str] = None, limit: int = 10) -> str: """ 焼津市のFIWARE NGSIv2 APIからエンティティデータを取得します。 APIカタログの情報を基に適切なエンドポイント設定を自動で行います。 Args: entity_type: エンティティタイプ（例: Aed, EvacuationShelter, DisasterMail） params: 追加のクエリパラメータ（JSON文字列形式） limit: 取得件数制限（1-1000、デフォルト10） Returns: str: APIレスポンス """ import aiohttp import uuid try: # APIキーを環境変数から取得 api_key = os.getenv('YAIZU_API_KEY') if not api_key: return "❌ エラー: APIキーが設定されていません。.envファイルにYAIZU_API_KEYを設定してください。" # エンティティタイプに応じたサービスパス決定 service_paths = { "Aed": "/Aed", "EvacuationShelter": "/EvacuationShelter", "DisasterMail": "/DisasterMail", "WeatherAlert": "/WeatherAlert", "WeatherForecast": "/WeatherForecast", "FloodRiskAreaMaxScale": "/FloodRiskAreaMaxScale", "TsunamiEvacuationBuilding": "/TsunamiEvacuationBuilding", "DrinkingWaterTank": "/DrinkingWaterTank", "PrecipitationGauge": "/PrecipitationGauge", "CameraInformation": "/CameraInformation", "StreamGauge": "/StreamGauge", "FirstAidStation": "/FirstAidStation", "ReliefHospital": "/ReliefHospital" } service_path = service_paths.get(entity_type, "/") # クエリパラメータの構築 query_params = { "type": entity_type, "limit": str(min(max(1, limit), 1000)) # 1-1000の範囲に制限 } # 追加パラメータの処理 if params: try: if isinstance(params, str): additional_params = json.loads(params) else: additional_params = params query_params.update(additional_params) except json.JSONDecodeError: return "❌ エラー: パラメータはJSON形式で指定してください。" # ヘッダーの設定（curlコマンドと完全一致） headers = { "Accept": "application/json", "apikey": api_key, # 小文字のapikey "Fiware-Service": "smartcity_yaizu", "Fiware-ServicePath": service_path, "x-request-trace-id": str(uuid.uuid4()), # UUIDトレースID "User-Agent": "smartcity-service" # WAF対策 # 重要: Content-Typeヘッダーは含めない（GETリクエストでは不要、curlでも送信していない） } # エンドポイントURL（検証済みベースURL） endpoint_url = "https://api.smartcity-yaizu.jp/v2/entities" # APIリクエスト実行 async with aiohttp.ClientSession() as session: logger.info(f"焼津市API実行: {entity_type} エンティティ取得") # NGSIv2 GETリクエスト実行 async with session.get(endpoint_url, headers=headers, params=query_params, timeout=30) as response: status = response.status response_text = await response.text() # レート制限情報を取得 rate_limit = response.headers.get('x-ratelimit-remaining-minute', 'N/A') rate_limit_reset = response.headers.get('ratelimit-reset', 'N/A') # レスポンスの処理 output = f"# 焼津市API実行結果\n\n" output += f"**エンティティタイプ**: `{entity_type}`\n" output += f"**エンドポイント**: `{endpoint_url}`\n" output += f"**サービスパス**: `{service_path}`\n" output += f"**ステータスコード**: {status}\n" output += f"**レート制限残り**: {rate_limit}\n\n" if query_params: output += f"**クエリパラメータ**:\n```json\n{json.dumps(query_params, ensure_ascii=False, indent=2)}\n```\n\n" if status == 200: output += "✅ **成功**\n\n" try: # JSONレスポンスをパース json_data = json.loads(response_text) data_count = len(json_data) if isinstance(json_data, list) else 1 output += f"**取得件数**: {data_count}件\n\n" # データサマリー表示（最初の3件のみ要約） if isinstance(json_data, list) and len(json_data) > 0: output += "**データサマリー**:\n" for i, item in enumerate(json_data[:3]): name = "名称不明" address = "住所不明" position = "" # 名称取得 if 'Name' in item and 'value' in item['Name']: name = item['Name']['value'] # 住所取得 if 'EquipmentAddress' in item and 'value' in item['EquipmentAddress']: addr = item['EquipmentAddress']['value'] if isinstance(addr, dict): if 'FullAddress' in addr: if isinstance(addr['FullAddress'], dict) and 'value' in addr['FullAddress']: address = addr['FullAddress']['value'] elif isinstance(addr['FullAddress'], str): address = addr['FullAddress'] # 設置位置取得 if 'InstallationPosition' in item and 'value' in item['InstallationPosition']: position = f" ({item['InstallationPosition']['value']})" output += f"- **{i+1}. {name}**: {address}{position}\n" output += f" - ID: `{item.get('id', 'N/A')}`\n" if len(json_data) > 3: output += f"- ... 他{len(json_data) - 3}件\n" output += "\n**完全なレスポンスデータ**:\n```json\n" output += json.dumps(json_data, ensure_ascii=False, indent=2)[:4000] # 最大4000文字 if len(json.dumps(json_data, ensure_ascii=False)) > 4000: output += "\n... (データが大きすぎるため省略)" output += "\n```" except json.JSONDecodeError: output += f"**レスポンス**:\n```\n{response_text[:2000]}\n```" elif status == 401: output += "❌ **認証エラー**: APIキーが無効か、権限がありません。\n" output += f"詳細: {response_text}" elif status == 403: output += "❌ **アクセス拒否**: APIキーの権限またはFiwareサービス設定を確認してください。\n" output += f"詳細: {response_text}" elif status == 404: output += "❌ **Not Found**: エンティティタイプまたはエンドポイントが見つかりません。\n" output += f"詳細: {response_text}" elif status == 429: output += "❌ **レート制限超過**: APIリクエスト数が制限を超えました。しばらく待ってから再試行してください。\n" output += f"詳細: {response_text}" else: output += f"❌ **エラー**: {status}\n" output += f"詳細: {response_text[:1000]}" return output except asyncio.TimeoutError: return "❌ タイムアウトエラー: APIへのリクエストがタイムアウトしました。" except aiohttp.ClientError as e: return f"❌ 接続エラー: {str(e)}" except Exception as e: logger.error(f"API実行エラー: {e}") return f"❌ 予期しないエラー: {str(e)}"
• tests/test_mcp_integration.py:95-147 (handler)

  Test/mock implementation of execute_yaizu_api used in integration tests.
  async def execute_yaizu_api(entity_type: str, operation: str = "list", **params) -> str: """API実行の実装""" if not API_KEY: return "Error: YAIZU_API_KEY not found in environment variables" spec_file = DATA_DIR / "api_specs" / f"{entity_type}.json" if not spec_file.exists(): return f"Error: API仕様ファイルが見つかりません: {entity_type}" with open(spec_file, 'r', encoding='utf-8') as f: spec = json.load(f) base_url = spec['api_specification']['base_url'] headers = spec['api_specification']['required_headers'].copy() # APIキーをヘッダーに追加（小文字のapikey） headers['apikey'] = API_KEY headers['Accept'] = 'application/json' headers['User-Agent'] = 'smartcity-service' headers['x-request-trace-id'] = str(uuid.uuid4()) # 必須のトレースID # Content-TypeはGETリクエストでは不要 if 'Content-Type' in headers: del headers['Content-Type'] # URL構築 if operation == "list": url = f"{base_url}/v2/entities?type={entity_type}" if params.get('limit'): url += f"&limit={params['limit']}" if params.get('offset'): url += f"&offset={params['offset']}" # HTTP リクエスト実行 try: print(f"🔗 API Request: {url}") print(f"📋 Headers: {dict(headers)}") async with aiohttp.ClientSession() as session: async with session.get(url, headers=headers, timeout=30) as response: response_text = await response.text() if response.status == 200: return response_text else: return f"Error: HTTP {response.status} - {response_text}" except asyncio.TimeoutError: return "Error: API request timeout" except Exception as e: return f"Error: {str(e)}"