Yaizu Smart City MCP Server

Instructions

Implementation Reference

• mcp/server.py:427-595 (handler)

  The primary implementation of the 'execute_yaizu_api' MCP tool. This async function, decorated with @mcp.tool(), handles API requests to the Yaizu smart city FIWARE NGSIv2 endpoint. It constructs headers with API key, Fiware-Service/Path, and executes GET requests with query parameters for entity retrieval, processes responses, and formats output with summaries and JSON data.
  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)}"
• mcp/server.py:427-427 (registration)

  The @mcp.tool() decorator registers the execute_yaizu_api function as an MCP tool.
  async def execute_yaizu_api(entity_type: str, params: Optional[str] = None, limit: int = 10) -> str: