e-Stat MCP

Instructions

Implementation Reference

• e_stats_mcp/tools/catalog.py:8-60 (handler)

  The primary handler function implementing the get_data_catalog tool. It constructs query parameters based on input arguments and calls the _make_request helper to fetch data catalog information from the e-Stat API in JSON format.

  async def get_data_catalog(
      search_word: str | None = None,
      stats_field: str | None = None,
      stats_code: str | None = None,
      gov_code: str | None = None,
      survey_years: str | None = None,
      open_years: str | None = None,
      stats_name_list: str | None = None,
      updated_date: str | None = None,
      start_position: int | None = None,
      limit: int | None = None,
  ) -> dict:
      """データカタログ情報を取得する.
  
      Args:
          search_word: 検索キーワード
          stats_field: 統計分野コード
          stats_code: 政府統計コード
          gov_code: 政府機関コード
          survey_years: 調査年
          open_years: 公開年
          stats_name_list: 調査・集計の種類
          updated_date: 更新日
          start_position: データ取得開始位置
          limit: 取得件数
  
      Returns:
          データカタログ情報
      """
      params: dict = {}
      if search_word:
          params["searchWord"] = search_word
      if stats_field:
          params["statsField"] = stats_field
      if stats_code:
          params["statsCode"] = stats_code
      if gov_code:
          params["governmentCode"] = gov_code
      if survey_years:
          params["surveyYears"] = survey_years
      if open_years:
          params["openYears"] = open_years
      if stats_name_list:
          params["statsNameList"] = stats_name_list
      if updated_date:
          params["updatedDate"] = updated_date
      if start_position:
          params["startPosition"] = str(start_position)
      if limit:
          params["limit"] = str(limit)
  
      response = await _make_request("json/getDataCatalog", params)
      return cast(dict[str, Any], response)

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

  Registers the get_data_catalog handler as an MCP tool using the FastMCP decorator.

  mcp.tool()(get_data_catalog)

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

  Shared helper function that performs HTTP requests to the e-Stat API. Used by the get_data_catalog handler to execute the actual API call to 'json/getDataCatalog' endpoint.

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