e-Stat MCP

Instructions

Implementation Reference

• e_stats_mcp/tools/stats.py:387-447 (handler)

  The handler function that implements the get_stats_data_csv tool. It builds parameters using _build_stats_data_params and makes an API request to 'getSimpleStatsData' endpoint in CSV format via _make_request.

  async def get_stats_data_csv(
      stats_data_id: str,
      cdcat01: str | None = None,
      cdcat02: str | None = None,
      cdcat03: str | None = None,
      lvcat01: str | None = None,
      lvcat02: str | None = None,
      lvcat03: str | None = None,
      lvcat04: str | None = None,
      lvcat05: str | None = None,
      lvcat06: str | None = None,
      lvcat07: str | None = None,
      lvcat08: str | None = None,
      lvcat09: str | None = None,
      lvcat10: str | None = None,
      lvcat11: str | None = None,
      lvcat12: str | None = None,
      lvcat13: str | None = None,
      lvcat14: str | None = None,
      lvcat15: str | None = None,
      cdtime: str | None = None,
      cdarea: str | None = None,
      start_position: int | None = None,
      section_header_flg: bool = False,
      cnt_get_flg: bool = False,
      limit: int = 100,
  ) -> str:
      """統計データをCSVで取得する."""
      params = _build_stats_data_params(
          stats_data_id=stats_data_id,
          cdcat01=cdcat01,
          cdcat02=cdcat02,
          cdcat03=cdcat03,
          lvcat01=lvcat01,
          lvcat02=lvcat02,
          lvcat03=lvcat03,
          lvcat04=lvcat04,
          lvcat05=lvcat05,
          lvcat06=lvcat06,
          lvcat07=lvcat07,
          lvcat08=lvcat08,
          lvcat09=lvcat09,
          lvcat10=lvcat10,
          lvcat11=lvcat11,
          lvcat12=lvcat12,
          lvcat13=lvcat13,
          lvcat14=lvcat14,
          lvcat15=lvcat15,
          cdtime=cdtime,
          cdarea=cdarea,
          start_position=start_position,
          section_header_flg=section_header_flg,
          cnt_get_flg=cnt_get_flg,
          limit=limit,
      )
      response = await _make_request(
          "getSimpleStatsData",
          params,
          format="csv",
      )
      return cast(str, response)

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

  Registers the get_stats_data_csv function as an MCP tool using FastMCP's tool decorator.

  mcp.tool()(get_stats_data_csv)

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

  Imports the get_stats_data_csv function for use in MCP registration.

  get_stats_data_csv,

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

  Helper function used by get_stats_data_csv to make HTTP requests to e-Stat API, handling CSV format specifically.

  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()

• e_stats_mcp/tools/stats.py:250-322 (helper)

  Helper function that constructs the query parameters for the stats data CSV request.

  def _build_stats_data_params(
      *,
      stats_data_id: str,
      cdcat01: str | None = None,
      cdcat02: str | None = None,
      cdcat03: str | None = None,
      lvcat01: str | None = None,
      lvcat02: str | None = None,
      lvcat03: str | None = None,
      lvcat04: str | None = None,
      lvcat05: str | None = None,
      lvcat06: str | None = None,
      lvcat07: str | None = None,
      lvcat08: str | None = None,
      lvcat09: str | None = None,
      lvcat10: str | None = None,
      lvcat11: str | None = None,
      lvcat12: str | None = None,
      lvcat13: str | None = None,
      lvcat14: str | None = None,
      lvcat15: str | None = None,
      cdtime: str | None = None,
      cdarea: str | None = None,
      start_position: int | None = None,
      section_header_flg: bool = False,
      cnt_get_flg: bool = False,
      limit: int = 100,
  ) -> dict:
      """getStatsData系の共通パラメータを組み立てる."""
      params = {
          "lang": "J",
          "statsDataId": stats_data_id,
          "limit": str(limit),
      }
  
      if cdcat01:
          params["cdCat01"] = cdcat01
      if cdcat02:
          params["cdCat02"] = cdcat02
      if cdcat03:
          params["cdCat03"] = cdcat03
      lv_params = [
          lvcat01,
          lvcat02,
          lvcat03,
          lvcat04,
          lvcat05,
          lvcat06,
          lvcat07,
          lvcat08,
          lvcat09,
          lvcat10,
          lvcat11,
          lvcat12,
          lvcat13,
          lvcat14,
          lvcat15,
      ]
      for idx, value in enumerate(lv_params, start=1):
          if value:
              params[f"lvCat{idx:02d}"] = value
      if cdtime:
          params["cdTime"] = cdtime
      if cdarea:
          params["cdArea"] = cdarea
      if start_position:
          params["startPosition"] = str(start_position)
      if section_header_flg:
          params["sectionHeaderFlg"] = "1"
      if cnt_get_flg:
          params["cntGetFlg"] = "1"
  
      return params