Yaizu Smart City MCP Server

Instructions

Implementation Reference

• mcp/server.py:131-167 (handler)

  Primary handler function for the 'scrape_api_docs' MCP tool. Decorated with @mcp.tool() for registration. Orchestrates scraping by initializing the scraper via doc_manager and calling scrape_and_save_all(), then returns a formatted summary.

  @mcp.tool()
  async def scrape_api_docs() -> str:
      """
      焼津市APIカタログから最新のAPIドキュメントをスクレイピングして保存します。
      認証情報は.envファイルから自動的に読み込まれます。
      
      Returns:
          str: スクレイピング結果のサマリー
      """
      try:
          scraper = await doc_manager.initialize_scraper()
          
          # スクレイピングと保存を実行
          result = await scraper.scrape_and_save_all()
          
          if result['success']:
              summary = f"""# スクレイピング完了
  
  ## 結果サマリー
  - **総API数**: {result['total_apis']}
  - **防災関連API数**: {result['disaster_apis']}
  - **保存されたファイル数**: {len(result['saved_files'])}
  
  ## 保存されたファイル
  """
              for filename in result['saved_files']:
                  summary += f"- {filename}\n"
              
              summary += "\n✅ APIドキュメントの更新が完了しました。"
              return summary
          else:
              return "❌ スクレイピングに失敗しました。認証情報を確認してください。"
              
      except Exception as e:
          logger.error(f"スクレイピングエラー: {e}")
          return f"❌ エラーが発生しました: {str(e)}"

• mcp/scraper.py:475-521 (helper)

  Core helper method in YaizuAPIScraper class that performs the actual scraping: logs in, fetches full API catalog, extracts disaster-related APIs, saves multiple JSON files locally, and returns summary statistics.

  async def scrape_and_save_all(self) -> Dict[str, Any]:
      """すべてのAPIデータをスクレイピングして保存"""
      result = {
          'success': False,
          'total_apis': 0,
          'disaster_apis': 0,
          'saved_files': []
      }
      
      # ログイン
      if not await self.login():
          logger.error("ログインに失敗しました")
          return result
      
      # APIカタログを取得
      catalog = await self.fetch_api_catalog()
      if catalog:
          # 全体のカタログを保存
          if await self.save_api_data(catalog, 'api_catalog'):
              result['saved_files'].append('api_catalog.json')
          
          result['total_apis'] = len(catalog.get('apis', []))
      
      # 防災関連APIを取得
      disaster_apis = await self.fetch_disaster_apis()
      if disaster_apis:
          disaster_catalog = {
              'title': '焼津市防災関連API',
              'apis': disaster_apis,
              'last_updated': datetime.now().isoformat()
          }
          
          if await self.save_api_data(disaster_catalog, 'disaster_apis'):
              result['saved_files'].append('disaster_apis.json')
          
          result['disaster_apis'] = len(disaster_apis)
          
          # 個別のAPI詳細も保存
          for api in disaster_apis:
              api_name = api.get('name', 'unknown').replace(' ', '_').replace('/', '_')
              if await self.save_api_data(api, f"api_{api_name}"):
                  result['saved_files'].append(f"api_{api_name}.json")
      
      result['success'] = True
      logger.info(f"スクレイピング完了: {result}")
      
      return result

• mcp/server.py:43-120 (helper)

  APIDocumentManager class provides document management, scraper initialization (key for tool), loading/saving/searching of scraped API docs. Global instance doc_manager used in handler.

  class APIDocumentManager:
      """APIドキュメント管理クラス"""
      
      def __init__(self):
          self.data_dir = Path("data/api_specs")
          self.api_docs_dir = Path("data/api_specs")
          self.data_dir.mkdir(parents=True, exist_ok=True)
          self.api_docs_dir.mkdir(parents=True, exist_ok=True)
          self.scraper = None
      
      async def initialize_scraper(self) -> YaizuAPIScraper:
          """スクレイパーの初期化"""
          if not self.scraper:
              self.scraper = YaizuAPIScraper()
              await self.scraper.__aenter__()
          return self.scraper
      
      async def cleanup(self):
          """リソースのクリーンアップ"""
          if self.scraper:
              await self.scraper.__aexit__(None, None, None)
      
      def load_api_docs(self, filename: str = "api_catalog") -> Optional[Dict[str, Any]]:
          """保存済みのAPIドキュメントを読み込む"""
          try:
              # data/api_specsディレクトリを確認
              specs_filepath = self.data_dir / f"{filename}.json"
              if specs_filepath.exists():
                  with open(specs_filepath, 'r', encoding='utf-8') as f:
                      return json.load(f)
              data_filepath = self.data_dir / f"{filename}.json"
              if data_filepath.exists():
                  with open(data_filepath, 'r', encoding='utf-8') as f:
                      return json.load(f)
              
              logger.warning(f"APIドキュメントが見つかりません: {filename}")
              return None
                  
          except Exception as e:
              logger.error(f"APIドキュメント読み込みエラー: {e}")
              return None
      
      def list_available_docs(self) -> List[str]:
          """利用可能なドキュメントファイルの一覧"""
          try:
              files = set()
              # data/api_specsディレクトリのファイル
              if self.data_dir.exists():
                  files.update(f.stem for f in self.data_dir.glob('*.json'))
              return sorted(list(files))
          except Exception as e:
              logger.error(f"ドキュメント一覧取得エラー: {e}")
              return []
      
      def search_apis(self, keyword: str) -> List[Dict[str, Any]]:
          """キーワードでAPIを検索"""
          results = []
          
          # すべてのドキュメントを検索
          for doc_file in self.list_available_docs():
              doc = self.load_api_docs(doc_file)
              if not doc:
                  continue
              
              # APIs配列がある場合
              if 'apis' in doc:
                  for api in doc['apis']:
                      # 名前と説明で検索
                      if (keyword.lower() in api.get('name', '').lower() or 
                          keyword.lower() in api.get('description', '').lower() or
                          keyword.lower() in api.get('category', '').lower()):
                          results.append({
                              'source_file': doc_file,
                              'api': api
                          })
          
          return results