SP Database MCP Server

Instructions

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 []

• sp_database_mcp/api_client.py:144-170 (helper)

  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 []