Yaizu Smart City MCP Server

Instructions

Implementation Reference

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

  Primary handler function for the 'scrape_api_docs' tool, registered via @mcp.tool() decorator. Initializes APIDocumentManager's scraper and invokes scrape_and_save_all() to perform the scraping, then formats and returns a 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 scraping logic in YaizuAPIScraper.scrape_and_save_all(): handles authentication, fetches API catalog and disaster-related APIs, parses data, and saves multiple JSON files to data/api_specs/ directory.
  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:131-167 (registration)

  The @mcp.tool() decorator on scrape_api_docs function registers it as an MCP tool.
  @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/server.py:43-120 (helper)

  APIDocumentManager class manages scraper lifecycle and provides methods to load, list, and search saved API documents. Used by scrape_api_docs 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