list_all_tables
Retrieve all database tables to understand schema structure and available data sources for query planning and analysis.
Instructions
列出所有数据库表
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| source | No | 数据源类型 | auto |
Implementation Reference
- sp_database_mcp/server.py:224-235 (handler)Handler logic in the main call_tool function for the list_all_tables tool: parses arguments, delegates to _list_all_tables helper, formats and returns markdown list of table names.elif name == "list_all_tables": source = arguments.get("source", "auto") tables = await _list_all_tables(source) if tables: output = f"数据库中共有 {len(tables)} 个表：\n\n" for table_name in sorted(tables): output += f"- {table_name}\n" return [TextContent(type="text", text=output)] else: return [TextContent(type="text", text="未找到任何表")]
- sp_database_mcp/server.py:153-167 (registration)Registration of the list_all_tables tool in list_tools(), including description and input schema allowing optional 'source' parameter.Tool( name="list_all_tables", description="列出所有数据库表", inputSchema={ "type": "object", "properties": { "source": { "type": "string", "enum": ["database", "api", "auto"], "description": "数据源类型", "default": "auto", } }, }, ),
- sp_database_mcp/server.py:304-319 (helper)Server-level helper function that dispatches list_all_tables request to appropriate client (database or API) based on source parameter, prioritizing database for 'auto'.async def _list_all_tables(source: str) -> List[str]: """列出所有表的内部方法""" if source == "database" and db_client: return db_client.get_all_tables() elif source == "api" and api_client: return await api_client.get_all_tables() elif source == "auto": # 优先使用数据库直连 if db_client: result = db_client.get_all_tables() if result: return result if api_client: return await api_client.get_all_tables() return []
- sp_database_mcp/database.py:197-208 (helper)DatabaseClient.get_all_tables(): Core implementation for direct database connection using SQLAlchemy's metadata.reflect() to retrieve all table names.def get_all_tables(self) -> List[str]: """获取所有表名""" if not self.engine: return [] try: metadata = MetaData() metadata.reflect(bind=self.engine) return list(metadata.tables.keys()) except SQLAlchemyError as e: print(f"Error getting table list: {e}") return []
- APIClient.get_all_tables(): Asynchronous HTTP client call to backend API endpoint /api/database/tables to fetch list of all table names.async def get_all_tables(self) -> List[str]: """通过 API 获取所有表名""" try: async with httpx.AsyncClient() as client: response = await client.get( f"{self.base_url}/api/database/tables", headers=self.headers, timeout=30.0, ) if response.status_code == 200: data = response.json() if isinstance(data, dict) and "tables" in data: return data["tables"] elif isinstance(data, list): return data else: return [] else: print( f"API request failed: {response.status_code} - {response.text}" ) return [] except httpx.RequestError as e: print(f"API request error: {e}") return []