MLIT Data Platform MCP Server

Instructions

            使い方:
            - 大量件数をバッチで取得します。内部的に GraphQL `getAllData` を使用し、返却された `nextDataRequestToken` を用いて次バッチを自動で取得します。
            - 絞り込みは `term` / `phrase_match` と、属性（`catalog_id`, `dataset_id`, `prefecture_code`, `municipality_code`, `address`）や矩形範囲を組み合わせて指定できます。
            - 1回のバッチ件数は `size`（API上限は1000）。本ツールの既定は `size=1000`（最大値）で、`max_batches` または `max_items` で総取得量を制御します。
            - メタデータが不要な場合は `include_metadata=False` で転送量を削減できます。

            例:
            - データセット単位で全件取得（メタデータ付き）:
            term="", dataset_id="mlit-001", size=1000, max_batches=10, include_metadata=True

            - カタログIDと矩形で範囲取得（東京都心部の例）:
            term="", catalog_id="dimaps",
            location_rectangle_top_left_lat=35.80,  location_rectangle_top_left_lon=139.55,
            location_rectangle_bottom_right_lat=35.60, location_rectangle_bottom_right_lon=139.85,
            size=1000, max_batches=5

            - 都道府県コードのみで全件走査:
            term="", prefecture_code="13", size=1000, max_items=5000

            注意:
            - API仕様上、`locationFilter`（矩形など）**単独では検索不可**です。必ず `term` または `attributeFilter`（本ツールでは `catalog_id` / `dataset_id` / `prefecture_code` / `municipality_code` / `address` に相当）を併用してください。
            - 次バッチ取得時は `nextDataRequestToken` を使用し、**他の条件は無視**されます（ツール側で自動処理）。データが空になった時点で取得を停止します。
            - `size` のAPI上限は1000です（本ツールの既定値は1000）。大量取得時は `max_batches` / `max_items` を併用して制御してください。
            - 座標は WGS84。矩形は「北西（top_left）→南東（bottom_right）」の順で指定してください。
            - `include_metadata=False` にすると `id`/`title` 中心の軽量レスポンスになります。

Implementation Reference

• src/client.py:891-914 (handler)

  The handler for collecting all data using the `get_all_data_iter` async iterator, managing batch counts, and returning a dictionary of items.

  async def get_all_data_collect(
      self,
      params: GetAllDataInput,
      *,
      max_items: Optional[int] = None
  ) -> Dict[str, Any]:
      items: List[Dict[str, Any]] = []
      count = 0
      batches = 0
  
      async for batch in self.get_all_data_iter(params):
          rows = [x.dict() for x in batch]
          items.extend(rows)
          count += len(rows)
          batches += 1
  
          if max_items is not None and count >= max_items:
              break
  
          if len(str(items).encode("utf-8")) > 900_000:
              logger.warning("get_all_data_collect: response approaching size cap, truncating further collection")
              break
  
      return {"batches": batches, "count": count, "items": items}

• src/client.py:823-889 (handler)

  The iterator function that performs the actual API requests for `get_all_data`, handling batching, next_token logic, and filters.

  async def get_all_data_iter(self, params: GetAllDataInput) -> AsyncIterator[List[GetAllDataItem]]:
      token: Optional[str] = None
      batches = 0
  
      attr_filter_str = self.make_attribute_filter_strict_for_get_all_data(
          prefecture_code=params.prefecture_code,
          municipality_code=params.municipality_code,
          address=params.address,
          catalog_id=params.catalog_id,
          dataset_id=params.dataset_id,
      )
  
      loc_filter_str: Optional[str] = None
      if all(v is not None for v in [
          params.location_rectangle_top_left_lat,
          params.location_rectangle_top_left_lon,
          params.location_rectangle_bottom_right_lat,
          params.location_rectangle_bottom_right_lon
      ]):
          tl_lat = float(params.location_rectangle_top_left_lat)  # type: ignore
          tl_lon = float(params.location_rectangle_top_left_lon)  # type: ignore
          br_lat = float(params.location_rectangle_bottom_right_lat)  # type: ignore
          br_lon = float(params.location_rectangle_bottom_right_lon)  # type: ignore
          if not (-90 <= tl_lat <= 90 and -180 <= tl_lon <= 180 and -90 <= br_lat <= 90 and -180 <= br_lon <= 180):
              raise ValueError("Invalid rectangle coordinates")
          loc_filter_str = self.make_rectangle_filter(tl_lat, tl_lon, br_lat, br_lon)
  
      effective_first_term: Optional[str] = params.term
      if effective_first_term is None:
          if attr_filter_str is not None or loc_filter_str is not None:
              effective_first_term = ""
  
      while True:
          if batches >= params.max_batches:
              break
  
          if token:
              q = self.build_get_all_data(size=params.size, next_token=token)
          else:
              q = self.build_get_all_data(
                  size=params.size,
                  term=effective_first_term,
                  phrase_match=params.phrase_match,
                  attribute_filter=attr_filter_str,
                  location_filter=loc_filter_str,
                  next_token=None,
              )
  
          data = await self.post_query(q)
          node = data.get("getAllData") or {}
          raw_items = node.get("data") or []
          token = node.get("nextDataRequestToken")
  
          batch: List[GetAllDataItem] = []
          for it in raw_items:
              item = GetAllDataItem(
                  id=str(it.get("id")),
                  title=it.get("title"),
                  metadata=(it.get("metadata") if params.include_metadata else None),
              )
              batch.append(item)
  
          yield batch
  
          batches += 1
          if not batch or not token:
              break

• src/server.py:1338-1358 (registration)

  The registration point in `server.py` where the `get_all_data` tool is called and routed to the `client.get_all_data_collect` method.

  elif name == "get_all_data":
      arguments = await _auto_normalize_region_args(arguments, client)
      p = GetAllDataInput.model_validate({
          "size": arguments.get("size", 1000),
          "term": arguments.get("term"),
          "phrase_match": arguments.get("phrase_match"),
          "prefecture_code": arguments.get("prefecture_code"),
          "municipality_code": arguments.get("municipality_code"),
          "address": arguments.get("address"),
          "catalog_id": arguments.get("catalog_id"),
          "dataset_id": arguments.get("dataset_id"),
          "location_rectangle_top_left_lat": arguments.get("location_rectangle_top_left_lat"),
          "location_rectangle_top_left_lon": arguments.get("location_rectangle_top_left_lon"),
          "location_rectangle_bottom_right_lat": arguments.get("location_rectangle_bottom_right_lat"),
          "location_rectangle_bottom_right_lon": arguments.get("location_rectangle_bottom_right_lon"),
          "max_batches": arguments.get("max_batches", 20),
          "include_metadata": arguments.get("include_metadata", True),
      })
      max_items = arguments.get("max_items")
      data = await client.get_all_data_collect(p, max_items=max_items)