e-Stat MCP

Instructions

Implementation Reference

• e_stats_mcp/tools/stats.py:373-384 (handler)

  The main async handler function that executes the tool logic: fetches meta information for a given stats_data_id from the e-Stat API 'getSimpleMetaInfo' endpoint in CSV format using the shared _make_request helper.

  async def get_meta_info_csv(stats_data_id: str) -> str:
      """統計表のメタ情報をCSVで取得する."""
      params = {
          "lang": "J",
          "statsDataId": stats_data_id,
      }
      response = await _make_request(
          "getSimpleMetaInfo",
          params,
          format="csv",
      )
      return cast(str, response)

• e_stats_mcp/main.py:49-49 (registration)

  MCP tool registration using FastMCP's mcp.tool() decorator, which likely generates schema from type hints.

  mcp.tool()(get_meta_info_csv)

• e_stats_mcp/tools/stats.py:17-66 (helper)

  Shared helper function _make_request used by the handler to perform the actual HTTP request to the e-Stat API with proper parameters and CSV format handling.

  async def _make_request(
      endpoint: str,
      params: dict | None = None,
      *,
      method: str = "GET",
      format: str = "json",
      data: dict | None = None,
  ) -> dict[str, Any] | str:
      """e-Stat APIへのリクエストを実行する.
  
      Args:
          endpoint: APIエンドポイント（例: \"json/getStatsList\"）
          params: クエリパラメータ
          method: HTTPメソッド（GET/POST）
          format: レスポンス形式（\"json\" または \"csv\"）
          data: POST時のフォームデータ
  
      Returns:
          JSONの場合はdict、CSVの場合は文字列
  
      Raises:
          ValueError: APIキーが設定されていない場合
          httpx.HTTPError: API通信エラー
      """
      settings = get_settings()
      if not settings.E_STAT_APP_ID:
          raise ValueError(
              "E_STAT_APP_ID環境変数が設定されていません。"
              "https://www.e-stat.go.jp/api/ からアプリケーションIDを取得してください。"
          )
  
      url = f"{E_STAT_API_BASE}/{endpoint}"
      request_params = {"appId": settings.E_STAT_APP_ID, **(params or {})}
      http_method = method.upper()
  
      async with httpx.AsyncClient() as client:
          if http_method == "POST":
              response = await client.post(
                  url, params=request_params, data=data, timeout=DEFAULT_TIMEOUT
              )
          else:
              response = await client.get(
                  url, params=request_params, timeout=DEFAULT_TIMEOUT
              )
  
          response.raise_for_status()
          if format == "csv":
              return response.text
          return response.json()