describe_table

Instructions

Args:
    table_name: 表名
    db_config: 数据库连接配置参数，如果为None则使用默认配置
    
Returns:
    包含表结构信息的字典

Implementation Reference

• mysql-mcp.py:220-256 (handler)

  The describe_table tool handler function. Uses @mcp.tool() decorator for registration. Connects to MySQL, executes DESCRIBE {table_name}, and returns column structure information.

  @mcp.tool()
  async def describe_table(table_name: str, db_config: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
      """获取表结构
      
      Args:
          table_name: 表名
          db_config: 数据库连接配置参数，如果为None则使用默认配置
          
      Returns:
          包含表结构信息的字典
      """
      if not table_name:
          return {"error": "表名不能为空"}
          
      try:
          conn = get_connection(db_config)
          cursor = conn.cursor(dictionary=True)
          cursor.execute(f"DESCRIBE {table_name}")
          columns = cursor.fetchall()
          return {
              "success": True,
              "table": table_name,
              "columns": columns
          }
      except Error as e:
          error_message = f"获取表结构失败: {str(e)}"
          if "doesn't exist" in str(e):
              error_message += f"\n原因：表 {table_name} 不存在"
          elif "Access denied" in str(e):
              error_message += "\n原因：当前用户没有足够权限查看表结构"
          return {"error": error_message}
      except Exception as e:
          return {"error": f"获取表结构时发生未知错误: {str(e)}"}
      finally:
          if 'conn' in locals() and conn.is_connected():
              cursor.close()
              conn.close()

• mysql-mcp.py:220-221 (registration)

  Registration of the describe_table tool via @mcp.tool() decorator on the function. MCP (FastMCP) automatically registers the function as a tool.

  @mcp.tool()
  async def describe_table(table_name: str, db_config: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:

• mysql-mcp.py:222-230 (schema)

  Function signature and docstring defining the input schema: table_name (required str), db_config (optional Dict[str, Any]), and return type Dict[str, Any].

  """获取表结构
  
  Args:
      table_name: 表名
      db_config: 数据库连接配置参数，如果为None则使用默认配置
      
  Returns:
      包含表结构信息的字典
  """

• mysql-mcp.py:63-119 (helper)

  The get_connection helper function used by describe_table to establish database connections with retry logic.

  def get_connection(db_config=None):
      """获取数据库连接
      
      Args:
          db_config: 数据库连接配置参数，如果为None则使用默认配置
          
      Returns:
          数据库连接对象
      """
      # 如果没有提供配置，先尝试使用全局配置，再使用默认配置
      if db_config is None:
          if GLOBAL_DB_CONFIG is not None:
              db_config = GLOBAL_DB_CONFIG.copy()
          else:
              db_config = DEFAULT_DB_CONFIG.copy()
      else:
          # 合并用户提供的配置和全局/默认配置
          if GLOBAL_DB_CONFIG is not None:
              config = GLOBAL_DB_CONFIG.copy()
          else:
              config = DEFAULT_DB_CONFIG.copy()
          config.update(db_config)
          db_config = config
      
      retry_count = 0
      last_error = None
      max_retries = db_config.get("connect_retry_count", 3)
      
      while retry_count < max_retries:
          try:
              # 创建一个配置字典的副本，移除自定义的配置项
              db_config_copy = db_config.copy()
              db_config_copy.pop("connect_retry_count", None)
              
              # 将connection_timeout转换为mysql.connector需要的connect_timeout参数
              if "connection_timeout" in db_config_copy:
                  db_config_copy["connect_timeout"] = db_config_copy.pop("connection_timeout")
                  
              conn = mysql.connector.connect(**db_config_copy)
              return conn
          except Error as e:
              last_error = e
              retry_count += 1
              if retry_count < max_retries:
                  # 只有在还有重试机会的情况下打印重试信息
                  print(f"第 {retry_count} 次连接失败，正在重试... 错误: {e}")
      
      # 所有重试都失败后，构建详细的错误信息
      error_message = f"数据库连接错误(重试 {retry_count} 次后): {last_error}"
      if "Can't connect to MySQL server" in str(last_error):
          error_message += f"\n无法连接到MySQL服务器，请检查主机 {db_config['host']} 和端口 {db_config['port']} 是否正确"
          error_message += f"\n连接超时时间为 {db_config.get('connection_timeout', 10)} 秒"
      elif "Access denied" in str(last_error):
          error_message += f"\n访问被拒绝，请检查用户名 {db_config['user']} 和密码是否正确"
      elif "Unknown database" in str(last_error):
          error_message += f"\n未知数据库 {db_config['database']}，请确认数据库名称是否正确"
      raise Exception(error_message)

• mysql-mcp.py:9-9 (helper)

  FastMCP server initialization (mcp = FastMCP('mysql')) used by @mcp.tool() decorator for tool registration.

  mcp = FastMCP("mysql")