mysql_mcp_server

Instructions

Implementation Reference

• src/mysql_mcp_server_pro/handles/get_table_index.py:34-67 (handler)

  The run_tool method implements the core execution logic: splits table names from 'text' input, builds SQL query to select indexes from information_schema.STATISTICS, executes it using the ExecuteSQL tool, and returns results or error.

  async def run_tool(self, arguments: Dict[str, Any]) -> Sequence[TextContent]:
      """获取指定表的索引信息
  
      参数:
          text (str): 要查询的表名，多个表名以逗号分隔
  
      返回:
          list[TextContent]: 包含查询结果的TextContent列表
          - 返回表的索引名、索引字段、索引类型等信息
          - 结果按表名、索引名和索引顺序排序
          - 结果以CSV格式返回，包含列名和数据
      """
      try:
          if "text" not in arguments:
              raise ValueError("缺少查询语句")
  
          text = arguments["text"]
  
          config = get_db_config()
          execute_sql = ExecuteSQL()
  
          # 将输入的表名按逗号分割成列表
          table_names = [name.strip() for name in text.split(',')]
          # 构建IN条件
          table_condition = "','".join(table_names)
  
          sql = "SELECT TABLE_NAME, INDEX_NAME, COLUMN_NAME, SEQ_IN_INDEX, NON_UNIQUE, INDEX_TYPE "
          sql += f"FROM information_schema.STATISTICS WHERE TABLE_SCHEMA = '{config['database']}' "
          sql += f"AND TABLE_NAME IN ('{table_condition}') ORDER BY TABLE_NAME, INDEX_NAME, SEQ_IN_INDEX;"
  
          return await execute_sql.run_tool({"query": sql})
  
      except Exception as e:
          return [TextContent(type="text", text=f"执行查询时出错: {str(e)}")]

• src/mysql_mcp_server_pro/handles/get_table_index.py:18-33 (schema)

  Defines the tool's input schema: an object requiring a 'text' string parameter for the table name(s) to search.

  def get_tool_description(self) -> Tool:
      return Tool(
          name=self.name,
          description=self.description,
          inputSchema={
              "type": "object",
              "properties": {
                  "text": {
                      "type": "string",
                      "description": "要搜索的表名"
                  }
              },
              "required": ["text"]
          }
      )

• src/mysql_mcp_server_pro/handles/base.py:56-61 (registration)

  BaseHandler.__init_subclass__ automatically registers any subclass (like GetTableIndex) to the ToolRegistry when the class is defined.

  def __init_subclass__(cls, **kwargs):
      """子类初始化时自动注册到工具注册表"""
      super().__init_subclass__(**kwargs)
      if cls.name:  # 只注册有名称的工具
          ToolRegistry.register(cls)

• src/mysql_mcp_server_pro/handles/__init__.py:1-23 (registration)

  Imports GetTableIndex class (triggering automatic registration) and includes it in __all__ for easy access.

  from .execute_sql import ExecuteSQL
  from .get_chinese_initials import GetChineseInitials
  from .get_table_desc import GetTableDesc
  from .get_table_index import GetTableIndex
  from .get_table_lock import GetTableLock
  from .get_table_name import GetTableName
  from .get_db_health_running import GetDBHealthRunning
  from .get_db_health_index_usage import GetDBHealthIndexUsage
  from .use_prompt_queryTableData import UsePromptQueryTableData
  from .optimize_sql import OptimizeSql
  
  __all__ = [
      "ExecuteSQL",
      "GetChineseInitials",
      "GetTableDesc",
      "GetTableIndex",
      "GetTableLock",
      "GetTableName",
      "GetDBHealthRunning",
      "GetDBHealthIndexUsage",
      "UsePromptQueryTableData",
      "OptimizeSql"
  ]