e-Stat MCP

Instructions

Implementation Reference

• e_stats_mcp/tools/stats.py:155-170 (handler)

  The core handler function implementing the get_meta_info tool. It makes an API request to e-Stat's getMetaInfo endpoint using the provided stats_data_id and returns the metadata as a dictionary.

  async def get_meta_info(stats_data_id: str) -> dict:
      """統計表のメタ情報を取得する.
  
      Args:
          stats_data_id: 統計表ID
  
      Returns:
          メタ情報（分類事項、時間軸など）
      """
      params = {
          "lang": "J",
          "statsDataId": stats_data_id,
      }
  
      response = await _make_request("json/getMetaInfo", params)
      return cast(dict[str, Any], response)

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

  Registration of the get_meta_info tool using the FastMCP decorator, making it available in the MCP server.

  mcp.tool()(get_meta_info)

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

  Helper function used by get_meta_info to make HTTP requests to the e-Stat API, handling authentication, parameters, and response parsing.

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